// inboundreply.go is RFC 0014 Phase 1: a reply to a notification email
// arriving as a comment. notify receives the mail, proves the thread
// binding and puts an email.reply event on its delivery stream; that
// stream is POSTed here. quarry decides what the reply becomes and
// whether the account may write it — the same split as outbound.
package api
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net/http"
"strconv"
"strings"
"time"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/webhook"
)
// ReplyStore is what the receiver remembers across deliveries: the
// secret notify signs with (empty until notify has shown one, and then
// nothing verifies), and which mails were already applied.
type ReplyStore interface {
NotifyHookSecret() (string, error)
InboundReplySeen(messageID string) (outcome string, seen bool, err error)
RecordInboundReply(messageID, txnID, account, outcome string) error
}
// ReplyEvent is notify's email.reply event as it arrives. Every other
// delivery event on the stream shares the endpoint and is ignored by
// its type.
type ReplyEvent struct {
Type string `json:"type"`
SubscriberID string `json:"subscriberId"`
TransactionID string `json:"transactionId"`
Workflow string `json:"workflow"`
Text string `json:"text"`
MessageID string `json:"messageId"`
InReplyTo string `json:"inReplyTo"`
ReceivedAt string `json:"receivedAt"`
}
// ReplyTarget is what a notification's transaction id says the reply
// answers: a forum post, so far. The TxnID quarry chose when it fired
// the notification is the correlation — nothing new is minted.
type ReplyTarget struct {
Kind string // "forum"
Hash string // the post answered
Who string // the account the notification was addressed to
}
// replyTarget reads the forge's own transaction ids back. Two shapes
// reach a person about a forum post: forum-reply:<hash>:<who> for a
// thread they follow, mention:forum:<hash>:<who> for an @mention.
func replyTarget(txn string) (ReplyTarget, bool) {
var rest string
switch {
case strings.HasPrefix(txn, "forum-reply:"):
rest = strings.TrimPrefix(txn, "forum-reply:")
case strings.HasPrefix(txn, "mention:forum:"):
rest = strings.TrimPrefix(txn, "mention:forum:")
default:
return ReplyTarget{}, false
}
hash, who, ok := strings.Cut(rest, ":")
if !ok || hash == "" || who == "" {
return ReplyTarget{}, false
}
return ReplyTarget{Kind: "forum", Hash: hash, Who: who}, true
}
const (
replyBodyMax = 64 << 10
replyMaxSkew = 5 * time.Minute
replySigHdr = "X-Notify-Signature"
replyStampHdr = "X-Notify-Timestamp"
)
// replySigned answers whether a delivery was signed by the holder of
// secret: sha256=HMAC-SHA256(secret, timestamp + "." + body), the
// outbound scheme with the timestamp folded in so a captured request
// cannot be replayed after the skew window.
func replySigned(secret, stamp, sig string, body []byte, now time.Time) bool {
ts, err := strconv.ParseInt(stamp, 10, 64)
if err != nil {
return false
}
if d := now.Sub(time.Unix(ts, 0)); d > replyMaxSkew || d < -replyMaxSkew {
return false
}
signed := append([]byte(stamp+"."), body...)
return webhook.Verify(secret, signed, sig)
}
// ReplyReceiver is the handler behind POST /api/v1/notify/events.
type ReplyReceiver struct {
Store ReplyStore
CapsFor CapsFor
// Find resolves a forum post hash to the repository holding it, the
// full post hash and its thread root. The transaction id carries no
// repository — it was a dedup key, not an address — so the forge
// looks for the post itself.
Find func(hash string) (repo, post, thread string, ok bool)
// Post writes the reply as the account; the default runs forum.reply
// through the registry so the gates and the fanout are the op's.
Post func(ctx context.Context, in forumReplyIn) (string, error)
Now func() time.Time
}
// NewReplyReceiver wires the receiver to the forge: posts are found
// across the repositories the node holds and written through r.
func NewReplyReceiver(d Deps, r *ops.Registry, store ReplyStore, capsFor CapsFor) *ReplyReceiver {
return &ReplyReceiver{
Store: store, CapsFor: capsFor,
Find: func(hash string) (string, string, string, bool) {
for _, name := range d.Repos.Names() {
repo, ok := d.Repos.Get(name)
if !ok || repo.Index == nil {
continue
}
if post, thread, ok, err := repo.Index.ForumPostThread(hash); err == nil && ok {
// A post in a companion store is the PARENT's post:
// the op, the letters and the links are all the
// parent's (RFC 0015).
if parent := d.companionParent(name); parent != "" {
name = parent
}
return name, post, thread, true
}
}
return "", "", "", false
},
Post: func(ctx context.Context, in forumReplyIn) (string, error) {
out, err := r.Invoke(ctx, "forum.reply", in)
if err != nil {
return "", err
}
res, _ := out.(forumReplyOut)
if res.Staged {
return "", errors.New("held for moderation")
}
return res.Hash, nil
},
}
}
func writeJSON(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
json.NewEncoder(w).Encode(v)
}
func (h *ReplyReceiver) now() time.Time {
if h.Now != nil {
return h.Now()
}
return time.Now()
}
// ServeHTTP answers a delivery. Every refusal carries its reason: a
// sender cannot fix what it is not told (RFC 0014). 2xx acknowledges,
// anything else asks notify to try again, so only a transient failure
// answers 5xx and a delivery that can never succeed is told 4xx once.
func (h *ReplyReceiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
body, err := io.ReadAll(io.LimitReader(req.Body, replyBodyMax+1))
if err != nil || len(body) > replyBodyMax {
writeError(w, http.StatusRequestEntityTooLarge, errors.New("delivery larger than 64 KB"))
return
}
secret, err := h.Store.NotifyHookSecret()
if err != nil {
writeError(w, http.StatusInternalServerError, err)
return
}
if secret == "" || !replySigned(secret, req.Header.Get(replyStampHdr), req.Header.Get(replySigHdr), body, h.now()) {
writeError(w, http.StatusUnauthorized, errors.New("delivery is not signed by notify"))
return
}
var ev ReplyEvent
if err := json.Unmarshal(body, &ev); err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
if ev.Type != "email.reply" {
// The stream carries every delivery event; only a reply is ours.
writeJSON(w, http.StatusOK, map[string]any{"ignored": ev.Type})
return
}
if ev.MessageID == "" {
writeError(w, http.StatusUnprocessableEntity, errors.New("reply carries no messageId"))
return
}
if outcome, seen, err := h.Store.InboundReplySeen(ev.MessageID); err != nil {
writeError(w, http.StatusInternalServerError, err)
return
} else if seen {
writeJSON(w, http.StatusOK, map[string]any{"duplicate": true, "outcome": outcome})
return
}
outcome, status, err := h.apply(req.Context(), ev)
if status >= 500 {
// Not recorded: notify redelivers and the next attempt may work.
log.Printf("inbound: reply %s from %s: %v", ev.MessageID, ev.SubscriberID, err)
writeError(w, status, err)
return
}
if err != nil {
outcome = "refused: " + err.Error()
}
if rerr := h.Store.RecordInboundReply(ev.MessageID, ev.TransactionID, ev.SubscriberID, outcome); rerr != nil {
writeError(w, http.StatusInternalServerError, rerr)
return
}
log.Printf("inbound: reply %s from %s: %s", ev.MessageID, ev.SubscriberID, outcome)
if err != nil {
writeError(w, status, err)
return
}
writeJSON(w, status, map[string]any{"outcome": outcome})
}
// apply turns one reply into a post. The status says how the refusal
// should be read: 4xx is final and recorded, 5xx is transient.
func (h *ReplyReceiver) apply(ctx context.Context, ev ReplyEvent) (outcome string, status int, err error) {
target, ok := replyTarget(ev.TransactionID)
if !ok {
return "", http.StatusUnprocessableEntity,
fmt.Errorf("no thread behind transaction %q", ev.TransactionID)
}
// The token was minted for one subscriber; the notification it
// answers was addressed to one account. They must be the same
// person, or the reply would be written as somebody else.
if target.Who != ev.SubscriberID {
return "", http.StatusForbidden,
fmt.Errorf("reply from %s to a notification addressed to %s", ev.SubscriberID, target.Who)
}
text := strings.TrimSpace(ev.Text)
if text == "" {
return "", http.StatusUnprocessableEntity, errors.New("reply is empty once quoted history is removed")
}
repo, post, thread, found := h.Find(target.Hash)
if !found {
return "", http.StatusUnprocessableEntity, fmt.Errorf("no forum post %s on this forge", target.Hash)
}
if h.CapsFor != nil && !h.CapsFor(target.Who, repo).HasAny("3") {
return "", http.StatusForbidden, fmt.Errorf("%s lacks the forum-write capability (3) on %s", target.Who, repo)
}
hash, err := h.Post(ops.WithUser(ctx, target.Who), forumReplyIn{Repo: repo, Root: thread, To: post, Body: text})
if err != nil {
if errors.Is(err, ops.ErrNotFound) || errors.Is(err, ops.ErrForbidden) || errors.Is(err, ops.ErrInvalid) {
return "", http.StatusUnprocessableEntity, err
}
return "", http.StatusInternalServerError, err
}
return "forum:" + hash, http.StatusCreated, nil
}