From a2ed5e9790c9dc3e75ecaa38ca8f7c6c82754e90 Mon Sep 17 00:00:00 2001 From: Bretton Date: Thu, 13 Aug 2026 05:58:39 -0700 Subject: [PATCH] =?UTF-8?q?wip(task15):=20cycles=20A-E=20=E2=80=94=20outbo?= =?UTF-8?q?und=20queue=20stores,=20translation=20(Note/Page/vote/Undo/Dele?= =?UTF-8?q?te),=20object+activity=20serving,=20real=20enqueuer=20on=20the?= =?UTF-8?q?=20gate=20tx?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit internal/apobject shared wire builders; consume.OutboundEnqueuer gains *sql.Tx so the enqueue commits with the rev-gate advance (or is lost). PostIntent added for task 16. Co-Authored-By: Claude Fable 5 --- internal/ap/client.go | 22 +- internal/apobject/object.go | 214 +++++++ internal/consume/comments.go | 11 +- internal/consume/dispatch.go | 36 +- internal/consume/fault_test.go | 3 +- internal/consume/jetstream_fake_test.go | 3 +- internal/consume/metrics_test.go | 4 +- internal/consume/noop.go | 3 +- internal/consume/votes.go | 4 +- .../db/migrations/020_outbound_delivery.sql | 86 +++ internal/outbound/enqueuer.go | 221 +++++++ internal/outbound/enqueuer_test.go | 200 ++++++ internal/outbound/outbound.go | 54 ++ internal/outbound/outer_acceptance_test.go | 563 ++++++++++++++++ internal/outbound/translator.go | 253 ++++++++ internal/outbound/translator_page_test.go | 242 +++++++ internal/outbound/worker.go | 104 +++ internal/personas/personas.go | 23 +- internal/personas/serving.go | 79 +++ internal/personas/serving_object_test.go | 161 +++++ internal/store/interfaces.go | 96 +++ internal/store/migrations_test.go | 45 +- internal/store/models.go | 91 +++ internal/store/outbound_activities.go | 80 +++ internal/store/outbound_deliveries.go | 293 +++++++++ internal/store/outbound_delivery_test.go | 602 ++++++++++++++++++ internal/store/outbound_objects.go | 25 + internal/testutil/db.go | 8 +- 28 files changed, 3498 insertions(+), 28 deletions(-) create mode 100644 internal/apobject/object.go create mode 100644 internal/db/migrations/020_outbound_delivery.sql create mode 100644 internal/outbound/enqueuer.go create mode 100644 internal/outbound/enqueuer_test.go create mode 100644 internal/outbound/outbound.go create mode 100644 internal/outbound/outer_acceptance_test.go create mode 100644 internal/outbound/translator.go create mode 100644 internal/outbound/translator_page_test.go create mode 100644 internal/outbound/worker.go create mode 100644 internal/personas/serving_object_test.go create mode 100644 internal/store/outbound_activities.go create mode 100644 internal/store/outbound_deliveries.go create mode 100644 internal/store/outbound_delivery_test.go diff --git a/internal/ap/client.go b/internal/ap/client.go index 0226270..6f7e769 100644 --- a/internal/ap/client.go +++ b/internal/ap/client.go @@ -594,6 +594,26 @@ func (c *Client) SendActivity(ctx context.Context, inboxURL string, activity any if c.signer == nil { return errors.NewValidationError("signer", "SendActivity requires a configured Signer") } + return c.sendActivityWith(ctx, c.signer, inboxURL, activity) +} + +// SendActivityAs signed-POSTs an activity to a remote inbox using the GIVEN +// per-actor signer, not the client's configured Signer (which stays reserved +// for the service actor's Follow/Undo). Task 15's delivery worker signs each +// activity as the persona that authored the record. Retry/status handling +// mirrors SendActivity. +func (c *Client) SendActivityAs(ctx context.Context, signer *Signer, inboxURL string, activity any) error { + if signer == nil { + return errors.NewValidationError("signer", "SendActivityAs requires a per-actor Signer") + } + return c.sendActivityWith(ctx, signer, inboxURL, activity) +} + +// sendActivityWith is the shared POST loop: it marshals the activity once, +// then retries the signed POST under the client's backoff / egress guard, +// signing each attempt with the given signer. SendActivity supplies the +// service actor's configured signer; SendActivityAs supplies a per-actor one. +func (c *Client) sendActivityWith(ctx context.Context, signer *Signer, inboxURL string, activity any) error { payload, err := json.Marshal(activity) if err != nil { return fmt.Errorf("ap: encode activity: %w", err) @@ -614,7 +634,7 @@ func (c *Client) SendActivity(ctx context.Context, inboxURL string, activity any } req.Header.Set("User-Agent", c.userAgent) req.Header.Set("Content-Type", ContentTypeActivityJSON) - if err := c.signer.SignRequest(req, payload); err != nil { + if err := signer.SignRequest(req, payload); err != nil { return err } diff --git a/internal/apobject/object.go b/internal/apobject/object.go new file mode 100644 index 0000000..9983bba --- /dev/null +++ b/internal/apobject/object.go @@ -0,0 +1,214 @@ +// Package apobject owns the AP OBJECT wire shape (Note/Page). It is the single +// source both the outbound Translator (which wraps the object in a +// Create/Update activity for delivery) and the user-origin serving surface (GET +// /ap/object/{did}/{collection}/{rkey}, via RenderObject) render from — so a +// peer that re-fetches a delivered object by id gets byte-compatible addressing +// (the same to/cc/audience split Lemmy parsed on delivery). It lives in its own +// low-level package (importing only ap + errors) so both the delivery layer and +// the identity-serving layer can share it without an import cycle. +package apobject + +import ( + "encoding/json" + "fmt" + "html" + "strings" + + "tidepool/internal/ap" + "tidepool/internal/errors" +) + +// ContextActivityStreams is the AS2 context every object/activity we emit +// carries. The Note/Page shapes use only core AS2 terms (attributedTo, source, +// mediaType, audience), so the plain context is sufficient for Lemmy 0.19.20. +const ContextActivityStreams = "https://www.w3.org/ns/activitystreams" + +// BuildNote renders a comment as an AP Note. The Note carries the community in +// cc (the Note half of the Page/Note addressing split), to ⊇ as:Public, +// attributedTo as a SINGLE STRING, and HTML content alongside its markdown +// source. inReplyTo is emitted only when a parent is known. +func BuildNote(actorID, communityAPID, parentAPID, objectURL string, record map[string]any) map[string]any { + source, _ := record["content"].(string) + note := map[string]any{ + "type": "Note", + "id": objectURL, + "attributedTo": actorID, + "to": []string{ap.PublicAudience}, + "cc": []string{communityAPID}, + "audience": communityAPID, + "mediaType": "text/html", + "content": renderHTML(source), + "source": map[string]any{ + "content": source, + "mediaType": "text/markdown", + }, + } + if parentAPID != "" { + note["inReplyTo"] = parentAPID + } + if published, ok := record["createdAt"].(string); ok && published != "" { + note["published"] = published + } + return note +} + +// BuildPage renders a post as an AP Page. The Page half of the split puts the +// community in `to` (alongside as:Public) and leaves cc empty — the crux Lemmy +// 0.19.20 keys on to tell a top-level post from a reply. name is REQUIRED +// (Lemmy rejects a titleless Page); a link embed becomes a Link attachment; +// an nsfw self-label becomes sensitive:true. +func BuildPage(actorID, communityAPID, objectURL string, record map[string]any) (map[string]any, error) { + title, _ := record["title"].(string) + if title == "" { + return nil, errors.NewValidationError("title", "a Page requires a name (Lemmy rejects a titleless post)") + } + source, _ := record["content"].(string) + page := map[string]any{ + "type": "Page", + "id": objectURL, + "attributedTo": actorID, + // The Page/Note split: community in `to`, NOT cc. + "to": []string{communityAPID, ap.PublicAudience}, + "cc": []string{}, + "name": title, + "audience": communityAPID, + "mediaType": "text/html", + "content": renderHTML(source), + "source": map[string]any{ + "content": source, + "mediaType": "text/markdown", + }, + "sensitive": hasNSFWLabel(record), + } + if attachment := linkAttachment(record); attachment != nil { + page["attachment"] = attachment + } + if published, ok := record["createdAt"].(string); ok && published != "" { + page["published"] = published + } + return page, nil +} + +// linkAttachment maps a social.coves.embed.external embed to Lemmy's +// attachment [{type:Link, href}] (Lemmy reads the FIRST attachment as the +// post's link). An image embed (social.coves.embed.images) is NOT handled here: +// its blobs live in the author's PDS and need a blob→PDS-URL seam to render as +// attachment [{type:Image, url}] — a task follow-up. A post with no external +// embed carries no attachment. +func linkAttachment(record map[string]any) []any { + embed, ok := record["embed"].(map[string]any) + if !ok { + return nil + } + if kind, _ := embed["$type"].(string); kind != "social.coves.embed.external" { + return nil + } + external, ok := embed["external"].(map[string]any) + if !ok { + return nil + } + href, _ := external["uri"].(string) + if href == "" { + return nil + } + return []any{map[string]any{"type": "Link", "href": href}} +} + +// hasNSFWLabel reports whether the record self-labels nsfw +// (com.atproto.label.defs#selfLabels with a value of "nsfw"). +func hasNSFWLabel(record map[string]any) bool { + labels, ok := record["labels"].(map[string]any) + if !ok { + return false + } + values, ok := labels["values"].([]any) + if !ok { + return false + } + for _, raw := range values { + if entry, ok := raw.(map[string]any); ok { + if val, _ := entry["val"].(string); val == "nsfw" { + return true + } + } + } + return false +} + +// RenderObject renders the AP object a served /ap/object/{did}/{collection}/{rkey} +// URL returns, from the durable outbound snapshot — the same Note/Page shape the +// Translator delivered, so a peer re-fetching the object by id sees consistent +// addressing. userOrigin derives the object's id and its author's actor id; no +// DB lookup happens (every id is in the snapshot or its at-uri). +func RenderObject(userOrigin string, snapshot []byte) (map[string]any, error) { + snap, err := ParseSnapshot(snapshot) + if err != nil { + return nil, err + } + atURI, _ := snap["atUri"].(string) + if atURI == "" { + return nil, errors.NewValidationError("snapshot.atUri", "must not be empty") + } + trimmed := strings.TrimPrefix(atURI, "at://") + objectURL := userOrigin + "/ap/object/" + trimmed + did := trimmed + if slash := strings.IndexByte(trimmed, '/'); slash >= 0 { + did = trimmed[:slash] + } + actorID := userOrigin + "/ap/actor/" + did + community, _ := snap["communityApId"].(string) + parentAPID, _ := snap["parentApId"].(string) + record, _ := snap["record"].(map[string]any) + collection, _ := snap["collection"].(string) + + var object map[string]any + if strings.Contains(collection, "postv2") { + object, err = BuildPage(actorID, community, objectURL, record) + if err != nil { + return nil, err + } + } else { + object = BuildNote(actorID, community, parentAPID, objectURL, record) + } + // Served standalone (not embedded in an activity), so it carries its own + // context. + object["@context"] = ContextActivityStreams + return object, nil +} + +// ParseSnapshot decodes the consumer's durable snapshot into a generic map so +// the object builders can read the raw record and resolved thread context +// without a round-trip through a typed AP object. +func ParseSnapshot(raw []byte) (map[string]any, error) { + if len(raw) == 0 { + return nil, errors.NewValidationError("snapshot", "must not be empty") + } + var snap map[string]any + if err := json.Unmarshal(raw, &snap); err != nil { + return nil, fmt.Errorf("apobject: decode snapshot: %w", err) + } + return snap, nil +} + +// renderHTML wraps the markdown source in the minimal HTML Lemmy stores as +// `content` (the source itself rides `source.content`). Paragraphs are split on +// blank lines and escaped so the source's own angle brackets cannot inject +// markup. +func renderHTML(source string) string { + source = strings.ReplaceAll(source, "\r\n", "\n") + blocks := strings.Split(source, "\n\n") + var b strings.Builder + for _, block := range blocks { + block = strings.TrimSpace(block) + if block == "" { + continue + } + b.WriteString("

") + b.WriteString(html.EscapeString(block)) + b.WriteString("

\n") + } + if b.Len() == 0 { + return "

\n" + } + return b.String() +} diff --git a/internal/consume/comments.go b/internal/consume/comments.go index bdb7ad4..6f56823 100644 --- a/internal/consume/comments.go +++ b/internal/consume/comments.go @@ -124,7 +124,7 @@ func (d *Dispatcher) applyCommentWrite(ctx context.Context, tx *sql.Tx, did stri return fmt.Errorf("write outbound state for %s: %w", atURI, err) } - return d.enqueueComment(ctx, did, commit.Operation, stored, thread.ParentATURI, thread.ParentAPID) + return d.enqueueComment(ctx, tx, did, commit.Operation, stored, thread.ParentATURI, thread.ParentAPID) } // applyCommentDelete withdraws a comment, using ONLY state. @@ -156,13 +156,13 @@ func (d *Dispatcher) applyCommentDelete(ctx context.Context, tx *sql.Tx, did str } parent := d.parentFromSnapshot(dead.TranslatedSnapshot) - return d.enqueueComment(ctx, did, operationDelete, dead, parent.ATURI, parent.APID) + return d.enqueueComment(ctx, tx, did, operationDelete, dead, parent.ATURI, parent.APID) } // enqueueComment hands one intent to task 15. The activity id comes from the // seq the write just produced, so every applied operation gets its own stable // id and a redelivery reuses it. -func (d *Dispatcher) enqueueComment(ctx context.Context, did, operation string, stored *store.OutboundObject, parentATURI, parentAPID string) error { +func (d *Dispatcher) enqueueComment(ctx context.Context, tx *sql.Tx, did, operation string, stored *store.OutboundObject, parentATURI, parentAPID string) error { intent := CommentIntent{ Op: operation, ATURI: stored.ATURI, @@ -173,8 +173,9 @@ func (d *Dispatcher) enqueueComment(ctx context.Context, did, operation string, } // parentATURI carries the causal dependency (decision 15): delivery must // not present a reply to a peer before the thing it replies to. On a - // delete it comes from state, because the frame carries no reply refs. - if err := d.enqueuer.EnqueueActivity(ctx, did, did, parentATURI, intent); err != nil { + // delete it comes from state, because the frame carries no reply refs. The + // enqueue rides tx so it commits with the gate advance. + if err := d.enqueuer.EnqueueActivity(ctx, tx, did, did, parentATURI, intent); err != nil { return fmt.Errorf("enqueue comment intent for %s: %w", stored.ATURI, err) } return nil diff --git a/internal/consume/dispatch.go b/internal/consume/dispatch.go index ce7426a..ef4658e 100644 --- a/internal/consume/dispatch.go +++ b/internal/consume/dispatch.go @@ -80,13 +80,41 @@ type VoteIntent struct { // ActivityID reports the deterministic activity id. func (i VoteIntent) ActivityID() string { return i.ID } +// PostIntent is a Create/Update/Delete of a native post (a bridged +// social.coves.community.postv2 record). Task 15 OWNS the Page translation, but +// this consumer does not construct a PostIntent: a post rides the acceptance +// engine (task 16), which admits the post, writes the acceptance record, and +// constructs the PostIntent for the same outbound enqueue — so the shape lives +// here (beside its Comment/Vote siblings) while the producer lives there. +type PostIntent struct { + // Op is the commit operation: create, update or delete. + Op string + // ATURI is the post record's at-uri. + ATURI string + // ID is the deterministic activity id. + ID string + // CommunityAPID is the target community's AP Group id — for a Page it goes + // in `to` (the Page/Note addressing split), not `cc`. + CommunityAPID string + // Snapshot is the translated state a Delete is rebuilt from and a + // Create/Update{Page} is rendered from — the postv2 record plus resolved + // context, same envelope shape as a comment's snapshot. + Snapshot []byte +} + +// ActivityID reports the deterministic activity id. +func (i PostIntent) ActivityID() string { return i.ID } + // OutboundEnqueuer is the task 15 seam. main.go wires a logging noop until // task 15 swaps in the real delivery queue. type OutboundEnqueuer interface { - // EnqueueActivity hands one intent to delivery. orderingKey serializes - // causally related work; parentATURI carries the causal dependency - // (decision 15) so a reply is never delivered before its parent. - EnqueueActivity(ctx context.Context, actorDID, orderingKey, parentATURI string, intent Intent) error + // EnqueueActivity hands one intent to delivery ON THE CALLER'S TX — the + // enqueue must commit with the rev-gate advance the consumer is holding, or + // a rolled-back gate would strand an activity a replay cannot reproduce. + // orderingKey serializes causally related work; parentATURI carries the + // causal dependency (decision 15) so a reply is never delivered before its + // parent. + EnqueueActivity(ctx context.Context, tx *sql.Tx, actorDID, orderingKey, parentATURI string, intent Intent) error } // AcceptanceEngine is the task 16 seam. A native post to a bridged community diff --git a/internal/consume/fault_test.go b/internal/consume/fault_test.go index 811ce06..152ea5e 100644 --- a/internal/consume/fault_test.go +++ b/internal/consume/fault_test.go @@ -2,6 +2,7 @@ package consume import ( "context" + "database/sql" "tidepool/internal/store" ) @@ -21,7 +22,7 @@ type failingEnqueuer struct { calls int } -func (e *failingEnqueuer) EnqueueActivity(_ context.Context, _, _, _ string, _ Intent) error { +func (e *failingEnqueuer) EnqueueActivity(_ context.Context, _ *sql.Tx, _, _, _ string, _ Intent) error { e.calls++ return e.err } diff --git a/internal/consume/jetstream_fake_test.go b/internal/consume/jetstream_fake_test.go index c8381ee..73adcc6 100644 --- a/internal/consume/jetstream_fake_test.go +++ b/internal/consume/jetstream_fake_test.go @@ -2,6 +2,7 @@ package consume import ( "context" + "database/sql" "net/http" "net/http/httptest" "strings" @@ -143,7 +144,7 @@ type recordingEnqueuer struct { calls []recordedIntent } -func (e *recordingEnqueuer) EnqueueActivity(_ context.Context, actorDID, orderingKey, parentATURI string, intent Intent) error { +func (e *recordingEnqueuer) EnqueueActivity(_ context.Context, _ *sql.Tx, actorDID, orderingKey, parentATURI string, intent Intent) error { e.mu.Lock() defer e.mu.Unlock() e.calls = append(e.calls, recordedIntent{ diff --git a/internal/consume/metrics_test.go b/internal/consume/metrics_test.go index 86e884a..5b2bc70 100644 --- a/internal/consume/metrics_test.go +++ b/internal/consume/metrics_test.go @@ -130,13 +130,13 @@ func TestNewNoopEnqueuer_AcceptsIntentsAndDeliversNothing(t *testing.T) { require.NotNil(t, enqueuer, "a nil logger must default rather than nil-panic on the first intent") - err := enqueuer.EnqueueActivity(context.Background(), dispatchNativeDID, "key", "at://parent", + err := enqueuer.EnqueueActivity(context.Background(), nil, dispatchNativeDID, "key", "at://parent", CommentIntent{Op: "create", ATURI: "at://x", ID: "https://coves.social/ap/activity/abc"}) assert.NoError(t, err, "until task 15 lands, main wires this so the consumer still RUNS and writes its "+ "durable state — disabling the whole path instead would leave everything "+ "downstream of the cursor unexercised until delivery exists") - assert.NoError(t, enqueuer.EnqueueActivity(context.Background(), dispatchNativeDID, "key", "", + assert.NoError(t, enqueuer.EnqueueActivity(context.Background(), nil, dispatchNativeDID, "key", "", VoteIntent{Op: "undo", VoteATURI: "at://v", Direction: "up"})) } diff --git a/internal/consume/noop.go b/internal/consume/noop.go index bb1fd71..4b08770 100644 --- a/internal/consume/noop.go +++ b/internal/consume/noop.go @@ -2,6 +2,7 @@ package consume import ( "context" + "database/sql" "log/slog" ) @@ -26,7 +27,7 @@ func NewNoopEnqueuer(logger *slog.Logger) OutboundEnqueuer { // EnqueueActivity records what WOULD have been delivered and drops it. The // activity id is logged because it is the one field a later delivery has to // reproduce exactly: it is what a peer dedupes on. -func (e *noopEnqueuer) EnqueueActivity(_ context.Context, actorDID, orderingKey, parentATURI string, intent Intent) error { +func (e *noopEnqueuer) EnqueueActivity(_ context.Context, _ *sql.Tx, actorDID, orderingKey, parentATURI string, intent Intent) error { e.logger.Info("outbound intent dropped: no delivery queue wired", slog.String("actor", actorDID), slog.String("ordering_key", orderingKey), diff --git a/internal/consume/votes.go b/internal/consume/votes.go index c8b79c6..a5a7d0f 100644 --- a/internal/consume/votes.go +++ b/internal/consume/votes.go @@ -139,7 +139,7 @@ func (d *Dispatcher) applyVoteWrite(ctx context.Context, tx *sql.Tx, did string, return fmt.Errorf("write vote state for %s: %w", voteATURI, err) } - return d.enqueuer.EnqueueActivity(ctx, did, did, subjectATURI, VoteIntent{ + return d.enqueuer.EnqueueActivity(ctx, tx, did, did, subjectATURI, VoteIntent{ Op: operationCreate, VoteATURI: voteATURI, SubjectAPID: stored.SubjectAPID, @@ -187,7 +187,7 @@ func (d *Dispatcher) applyVoteDelete(ctx context.Context, tx *sql.Tx, did string return fmt.Errorf("bump vote state for %s: %w", voteATURI, err) } - return d.enqueuer.EnqueueActivity(ctx, did, did, stored.SubjectATURI, VoteIntent{ + return d.enqueuer.EnqueueActivity(ctx, tx, did, did, stored.SubjectATURI, VoteIntent{ Op: operationUndo, VoteATURI: voteATURI, SubjectAPID: stored.SubjectAPID, diff --git a/internal/db/migrations/020_outbound_delivery.sql b/internal/db/migrations/020_outbound_delivery.sql new file mode 100644 index 0000000..b993fc9 --- /dev/null +++ b/internal/db/migrations/020_outbound_delivery.sql @@ -0,0 +1,86 @@ +-- +goose Up +-- Task 15: the outbound delivery pipe, split into canonical ACTIVITIES and +-- per-inbox DELIVERIES (decision 15), plus the causal-gating marker on +-- outbound_objects. +-- +-- WHY THE SPLIT: a single globally-unique activity row cannot fan out. Task +-- 17's Delete{Person} is ONE activity delivered to MANY community inboxes, and +-- an already-delivered Create must keep serving its ORIGINAL payload after a +-- later edit (activities are immutable, objects are current). So the canonical +-- activity — the byte-stable wire payload a peer re-fetches by id — lives once +-- in outbound_activities, and every (activity, inbox) delivery attempt is a row +-- in outbound_deliveries. + +-- outbound_activities is the canonical, IMMUTABLE wire payload for one activity +-- id. It is what GET /ap/activity/{hash} serves and what a redelivery re-sends +-- verbatim, so a peer that already has the activity dedupes it on our stable +-- id. Rows are inserted ON CONFLICT DO NOTHING: the payload of an activity a +-- peer may already hold must never change under it. +CREATE TABLE outbound_activities ( + activity_id TEXT PRIMARY KEY, -- deterministic id (consume.ActivityID) + actor_did TEXT NOT NULL, -- the persona that signs the delivery + kind TEXT NOT NULL, -- Create/Update/Delete/Like/Dislike/Undo + payload JSONB NOT NULL, -- the canonical wire activity, byte-stable + parent_at_uri TEXT NOT NULL DEFAULT '', -- causal dependency ('' = none) + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- The causal-gating lookup: a delivery whose activity has a parent_at_uri is +-- ineligible until the parent's own outbound_objects row is accepted_at (a +-- BRIDGE-origin parent) — task 15's worker reads the parent chain by actor. +CREATE INDEX idx_outbound_activities_actor ON outbound_activities (actor_did); + +-- outbound_deliveries is one delivery attempt per (activity, target inbox). It +-- GENERALIZES inbox_events: the same claimed_until fencing token, the same +-- loose-index-scan per-ordering-key serialization (task 12's lesson), the same +-- SKIP LOCKED concurrency. ordering_key is the community AP id, so every +-- activity bound for one community delivers in a single serial line. +-- +-- seq is the monotonic ordering column the loose index scan descends: "the +-- head of an ordering key" is its min-seq pending row. It is a BIGSERIAL and +-- NOT part of the PK, because the PK is the natural (activity_id, target_inbox) +-- fan-out key. +CREATE TABLE outbound_deliveries ( + seq BIGSERIAL NOT NULL, + activity_id TEXT NOT NULL REFERENCES outbound_activities (activity_id) ON DELETE CASCADE, + target_inbox TEXT NOT NULL, + ordering_key TEXT NOT NULL, -- community AP id: per-community serial line + state TEXT NOT NULL DEFAULT 'pending' + CHECK (state IN ('pending', 'delivered', 'poisoned', 'cancelled')), + attempts INT NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + claimed_until TIMESTAMPTZ, -- the fencing/lease token (see inbox_events) + delivered_at TIMESTAMPTZ, + last_status_code INT, + last_error_class TEXT NOT NULL DEFAULT '', + response_excerpt TEXT NOT NULL DEFAULT '', -- bounded body sample for triage + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (activity_id, target_inbox) +); + +-- The loose-index-scan support index (task 12): one index descent per DISTINCT +-- pending ordering key jumps to each key's head, so ClaimNext is O(pending +-- keys × log N) regardless of any one community's backlog depth. Partial on +-- pending so delivered/poisoned/cancelled rows never bloat it. +CREATE INDEX idx_outbound_deliveries_queue + ON outbound_deliveries (ordering_key, seq) + WHERE state = 'pending'; + +-- CancelForActor sweeps a disabled/deleted actor's pending deliveries; it joins +-- through outbound_activities.actor_did, so the FK side is indexed above. +CREATE INDEX idx_outbound_deliveries_activity ON outbound_deliveries (activity_id); + +-- The causal marker (decision 15). accepted_at is stamped by delivery success: +-- NULL means "not yet delivered to its community", which is what gates a +-- bridge-origin child from delivering before its parent. Fediverse-origin +-- parents (no outbound_objects row) are always eligible. +ALTER TABLE outbound_objects ADD COLUMN accepted_at TIMESTAMPTZ; + +-- +goose Down +ALTER TABLE outbound_objects DROP COLUMN IF EXISTS accepted_at; +DROP INDEX IF EXISTS idx_outbound_deliveries_activity; +DROP INDEX IF EXISTS idx_outbound_deliveries_queue; +DROP TABLE IF EXISTS outbound_deliveries; +DROP INDEX IF EXISTS idx_outbound_activities_actor; +DROP TABLE IF EXISTS outbound_activities; diff --git a/internal/outbound/enqueuer.go b/internal/outbound/enqueuer.go new file mode 100644 index 0000000..d63a878 --- /dev/null +++ b/internal/outbound/enqueuer.go @@ -0,0 +1,221 @@ +package outbound + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "net/url" + "strings" + + "tidepool/internal/apobject" + "tidepool/internal/consume" + "tidepool/internal/errors" + "tidepool/internal/store" +) + +// EnqueuerOptions configures an Enqueuer. +type EnqueuerOptions struct { + // DB is the bridge database. + DB *sql.DB + // Translator renders intents into canonical AP activities. + Translator *Translator + // Inboxes resolves a community's target inbox at enqueue time (one delivery + // row per target; v2 targets are single-community). The (activity, inbox) + // pair is the delivery primary key, so the inbox is known when the row is + // written. + Inboxes InboxResolver + // Actors resolves an actor DID to its AP actor id (the single-string + // attributedTo the Translator addresses as). + Actors store.APActors + // Activities / Deliveries are the split queue. Optional: nil is + // constructed from DB. + Activities store.OutboundActivities + Deliveries store.OutboundDeliveries + // UserOrigin is AP_USER_ORIGIN. + UserOrigin string + // Logger receives drop reasons. Nil uses slog.Default(). + Logger *slog.Logger +} + +// Enqueuer is the real OutboundEnqueuer (task 15): it translates one intent and +// writes its canonical activity plus one per-target delivery, INSIDE the rev +// gate transaction the consumer hands it — the enqueue must commit with the +// gate advance or a rolled-back gate would leave the activity/delivery behind +// and a replay could not reproduce it. +type Enqueuer struct { + db *sql.DB + translator *Translator + inboxes InboxResolver + actors store.APActors + activities store.OutboundActivities + deliveries store.OutboundDeliveries + apObjects store.APObjects + userOrigin string + originHost string + logger *slog.Logger +} + +// NewEnqueuer wires an Enqueuer. +func NewEnqueuer(opts EnqueuerOptions) (*Enqueuer, error) { + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + activities := opts.Activities + if activities == nil { + activities = store.NewOutboundActivities(opts.DB) + } + deliveries := opts.Deliveries + if deliveries == nil { + deliveries = store.NewOutboundDeliveries(opts.DB) + } + // originHost labels bridge-emitted ap_objects rows (origin_instance): the + // host a re-fetch of this object dials, which is our own origin. + originHost := opts.UserOrigin + if parsed, err := url.Parse(opts.UserOrigin); err == nil && parsed.Host != "" { + originHost = strings.ToLower(parsed.Host) + } + return &Enqueuer{ + db: opts.DB, + translator: opts.Translator, + inboxes: opts.Inboxes, + actors: opts.Actors, + activities: activities, + deliveries: deliveries, + apObjects: store.NewAPObjects(opts.DB), + userOrigin: opts.UserOrigin, + originHost: originHost, + logger: logger, + }, nil +} + +// EnqueueActivity translates the intent and writes, ON THE CALLER'S TX, the +// canonical activity, the ap_objects mapping that makes the object fetchable, +// and one per-inbox delivery — so all of it commits with the gate advance or +// leaves nothing behind (a rolled-back gate must not strand a delivery a replay +// cannot reproduce). +// +// The community for both the inbox resolution and the delivery's ordering key +// is read from the intent (the per-community serial line), not the orderingKey +// argument: the consumer passes the actor DID there as a coarse hint, but a +// delivery is serialized and addressed by the target COMMUNITY. +func (e *Enqueuer) EnqueueActivity(ctx context.Context, tx *sql.Tx, actorDID, orderingKey, parentATURI string, intent consume.Intent) error { + if tx == nil { + return errors.NewValidationError("tx", "must not be nil") + } + actor, err := e.actors.GetByDID(ctx, actorDID) + if err != nil { + return fmt.Errorf("resolve actor %s: %w", actorDID, err) + } + translated, err := e.translator.Translate(actor.ActorID, intent) + if err != nil { + return fmt.Errorf("translate intent %s: %w", intent.ActivityID(), err) + } + + // The activity is the atomicity anchor: ON CONFLICT DO NOTHING, so a + // redelivered intent re-derives the same id and reports inserted=false. + // Because the activity, its mapping and its delivery all ride ONE + // transaction, an already-present activity id already has the other two — + // so a re-enqueue returns here without re-inserting the delivery (which + // would violate its PK and poison the tx). + inserted, err := e.activities.InsertTx(ctx, tx, store.OutboundActivity{ + ActivityID: intent.ActivityID(), + ActorDID: actorDID, + Kind: translated.Kind, + Payload: translated.Payload, + ParentATURI: parentATURI, + }) + if err != nil { + return fmt.Errorf("insert outbound activity %s: %w", intent.ActivityID(), err) + } + if !inserted { + return nil + } + + // The object mapping (bridge-origin) makes GET /ap/object serve the record. + // Votes have no servable object and a self-delete maps nothing new. + if mapping, ok, err := e.objectMapping(intent); err != nil { + return err + } else if ok { + if _, err := e.apObjects.PutMappingTx(ctx, tx, mapping); err != nil { + return fmt.Errorf("map outbound object %s: %w", mapping.APID, err) + } + } + + community := communityOf(intent) + inbox, err := e.inboxes.ResolveInbox(ctx, community) + if err != nil { + // No inbox means no delivery target: fail so the whole tx rolls back + // rather than writing a delivery to nowhere. + return fmt.Errorf("resolve inbox for community %s: %w", community, err) + } + if _, err := e.deliveries.EnqueueTx(ctx, tx, store.OutboundDelivery{ + ActivityID: intent.ActivityID(), + TargetInbox: inbox, + OrderingKey: community, + }); err != nil { + return fmt.Errorf("enqueue delivery for %s: %w", intent.ActivityID(), err) + } + return nil +} + +// objectMapping derives the bridge-origin ap_objects mapping for an intent that +// produces a servable object (a comment or post create/update). Votes have no +// object; a self-delete's object was mapped on its create. ok=false means no +// mapping is written. +func (e *Enqueuer) objectMapping(intent consume.Intent) (store.APObjectMapping, bool, error) { + var atURI, apType string + var snapshot []byte + switch typed := intent.(type) { + case consume.CommentIntent: + if typed.Op == "delete" { + return store.APObjectMapping{}, false, nil + } + atURI, apType, snapshot = typed.ATURI, "Note", typed.Snapshot + case consume.PostIntent: + if typed.Op == "delete" { + return store.APObjectMapping{}, false, nil + } + atURI, apType, snapshot = typed.ATURI, "Page", typed.Snapshot + default: + return store.APObjectMapping{}, false, nil + } + + trimmed := strings.TrimPrefix(atURI, "at://") + parts := strings.SplitN(trimmed, "/", 3) + if len(parts) != 3 { + return store.APObjectMapping{}, false, + errors.NewValidationError("intent.atUri", "must be at://did/collection/rkey, got "+atURI) + } + snap, err := apobject.ParseSnapshot(snapshot) + if err != nil { + return store.APObjectMapping{}, false, err + } + cid, _ := snap["cid"].(string) + return store.APObjectMapping{ + APID: e.userOrigin + "/ap/object/" + trimmed, + APType: apType, + OriginInstance: e.originHost, + Origin: store.OriginBridge, + DID: parts[0], + Collection: parts[1], + RKey: parts[2], + CID: cid, + }, true, nil +} + +// communityOf reads the target community AP id off any intent — the per- +// community serial line every delivery is ordered and addressed by. +func communityOf(intent consume.Intent) string { + switch typed := intent.(type) { + case consume.CommentIntent: + return typed.CommunityAPID + case consume.PostIntent: + return typed.CommunityAPID + case consume.VoteIntent: + return typed.CommunityAPID + default: + return "" + } +} diff --git a/internal/outbound/enqueuer_test.go b/internal/outbound/enqueuer_test.go new file mode 100644 index 0000000..dfbfe6d --- /dev/null +++ b/internal/outbound/enqueuer_test.go @@ -0,0 +1,200 @@ +package outbound + +import ( + "context" + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/consume" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// Task 15 cycle E: the real Enqueuer. It translates one intent and writes its +// canonical activity, its per-inbox delivery, AND the ap_objects mapping (so the +// object is fetchable), ALL on the caller's rev-gate transaction. The whole +// point is atomicity: the enqueue commits with the gate advance or leaves +// nothing behind, so a replay can always reproduce it. + +func enqueuerTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "outbound_deliveries", "outbound_activities", "ap_objects", "ap_actors") + return database +} + +// fakeInboxResolver records the communities it is asked to resolve and returns a +// canned inbox (or error). +type fakeInboxResolver struct { + inbox string + err error + calledWith []string +} + +func (r *fakeInboxResolver) ResolveInbox(_ context.Context, communityAPID string) (string, error) { + r.calledWith = append(r.calledWith, communityAPID) + return r.inbox, r.err +} + +// seedEnqueuerActor writes the ap_actors row the enqueuer resolves the actor id +// off of (actorDID -> attributedTo single string). +func seedEnqueuerActor(t *testing.T, conn *sql.DB) string { + t.Helper() + actorID := outUserOrigin + "/ap/actor/" + outCommenterDID + _, err := store.NewAPActors(conn).Create(context.Background(), store.APActor{ + DID: outCommenterDID, + Kind: store.ActorTypePerson, + ActorID: actorID, + NormalizedOrigin: "coves.social", + LocalPart: "alice", + RSAKeySealed: []byte("sealed-key-bytes"), + RSAKeyVersion: 1, + PublicKeyPEM: "-----BEGIN PUBLIC KEY-----\nstub\n-----END PUBLIC KEY-----\n", + }) + require.NoError(t, err, "seed ap_actors row for the commenter") + return actorID +} + +func newTestEnqueuer(t *testing.T, conn *sql.DB, inboxes InboxResolver) *Enqueuer { + t.Helper() + enq, err := NewEnqueuer(EnqueuerOptions{ + DB: conn, + Translator: NewTranslator(outUserOrigin), + Inboxes: inboxes, + Actors: store.NewAPActors(conn), + UserOrigin: outUserOrigin, + }) + require.NoError(t, err) + return enq +} + +func commentEnqueueIntent(t *testing.T) consume.CommentIntent { + return consume.CommentIntent{ + Op: "create", + ATURI: outCommentATURI, + ID: consume.ActivityID(outUserOrigin, outCommentATURI, "create", 0), + CommunityAPID: outCommunityAPID, + ParentAPID: outRootAPID, + Snapshot: commentSnapshot(t), + } +} + +func TestEnqueuer_CommittedTxWritesActivityDeliveryAndMapping(t *testing.T) { + conn := enqueuerTestDB(t) + ctx := context.Background() + seedEnqueuerActor(t, conn) + + resolver := &fakeInboxResolver{inbox: outSharedInbox} + enq := newTestEnqueuer(t, conn, resolver) + intent := commentEnqueueIntent(t) + + tx, err := conn.BeginTx(ctx, nil) + require.NoError(t, err) + require.NoError(t, enq.EnqueueActivity(ctx, tx, outCommenterDID, outCommunityAPID, outRootATURI, intent), + "EnqueueActivity must write its rows on the given tx") + require.NoError(t, tx.Commit()) + + // The inbox resolver was consulted with the community, and its answer is the + // delivery target. + require.Contains(t, resolver.calledWith, outCommunityAPID, + "the enqueuer resolves the community's inbox at enqueue time (the delivery PK needs it)") + + require.Equal(t, 1, count(t, conn, "outbound_activities"), "exactly one canonical activity") + require.Equal(t, 1, count(t, conn, "outbound_deliveries"), "exactly one per-inbox delivery") + + activity, err := store.NewOutboundActivities(conn).Get(ctx, intent.ID) + require.NoError(t, err) + require.NotNil(t, activity) + assert.Equal(t, outCommenterDID, activity.ActorDID) + assert.Equal(t, "Create", activity.Kind, "the translated kind rides the activity row") + assert.Equal(t, outRootATURI, activity.ParentATURI, "the causal parent is copied onto the row") + assert.NotEmpty(t, activity.Payload, "the canonical payload is stored") + + delivery, err := store.NewOutboundDeliveries(conn).Get(ctx, intent.ID, outSharedInbox) + require.NoError(t, err, "the delivery is keyed by (activity, resolved inbox)") + require.NotNil(t, delivery) + assert.Equal(t, outSharedInbox, delivery.TargetInbox, "target_inbox is the resolver's answer") + assert.Equal(t, outCommunityAPID, delivery.OrderingKey, "ordering_key is the community AP id") + assert.Equal(t, store.DeliveryStatePending, delivery.State) + + // The object is fetchable: an ap_objects mapping was written, origin=bridge. + mapping, err := store.NewAPObjects(conn).GetByAPID(ctx, outCommentAPID) + require.NoError(t, err, + "the enqueuer writes an ap_objects mapping so GET /ap/object can serve the record") + require.NotNil(t, mapping) + assert.Equal(t, store.OriginBridge, mapping.Origin, "a bridge-emitted object must say origin=bridge") + assert.Equal(t, outCommenterDID, mapping.DID) +} + +func TestEnqueuer_RolledBackTxWritesNothing(t *testing.T) { + conn := enqueuerTestDB(t) + ctx := context.Background() + seedEnqueuerActor(t, conn) + + enq := newTestEnqueuer(t, conn, &fakeInboxResolver{inbox: outSharedInbox}) + intent := commentEnqueueIntent(t) + + tx, err := conn.BeginTx(ctx, nil) + require.NoError(t, err) + require.NoError(t, enq.EnqueueActivity(ctx, tx, outCommenterDID, outCommunityAPID, outRootATURI, intent)) + require.NoError(t, tx.Rollback()) + + assert.Zero(t, count(t, conn, "outbound_activities"), + "a rolled-back gate tx must leave NO activity — this is the seam that makes replay safe") + assert.Zero(t, count(t, conn, "outbound_deliveries"), "...and no delivery") + assert.Zero(t, count(t, conn, "ap_objects"), "...and no mapping") +} + +func TestEnqueuer_ReEnqueueIsIdempotent(t *testing.T) { + conn := enqueuerTestDB(t) + ctx := context.Background() + seedEnqueuerActor(t, conn) + + enq := newTestEnqueuer(t, conn, &fakeInboxResolver{inbox: outSharedInbox}) + intent := commentEnqueueIntent(t) + + for i := 0; i < 2; i++ { + tx, err := conn.BeginTx(ctx, nil) + require.NoError(t, err) + require.NoError(t, enq.EnqueueActivity(ctx, tx, outCommenterDID, outCommunityAPID, outRootATURI, intent), + "a redelivered intent re-derives the same activity id and must not error") + require.NoError(t, tx.Commit()) + } + + assert.Equal(t, 1, count(t, conn, "outbound_activities"), + "a re-enqueue of the same activity id must not write a second activity (ON CONFLICT DO NOTHING)") + assert.Equal(t, 1, count(t, conn, "outbound_deliveries"), + "...nor a duplicate delivery for the same (activity, inbox)") +} + +func TestEnqueuer_InboxResolutionFailureRollsBack(t *testing.T) { + conn := enqueuerTestDB(t) + ctx := context.Background() + seedEnqueuerActor(t, conn) + + enq := newTestEnqueuer(t, conn, &fakeInboxResolver{err: assert.AnError}) + intent := commentEnqueueIntent(t) + + tx, err := conn.BeginTx(ctx, nil) + require.NoError(t, err) + err = enq.EnqueueActivity(ctx, tx, outCommenterDID, outCommunityAPID, outRootATURI, intent) + require.Error(t, err, + "a community whose inbox cannot be resolved has no delivery target — the enqueue must fail, "+ + "not write a delivery to nowhere") + _ = tx.Rollback() + + assert.Zero(t, count(t, conn, "outbound_activities"), "a failed enqueue leaves no partial rows") + assert.Zero(t, count(t, conn, "outbound_deliveries")) +} + +func count(t *testing.T, conn *sql.DB, table string) int { + t.Helper() + var n int + require.NoError(t, conn.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM `+table).Scan(&n)) + return n +} diff --git a/internal/outbound/outbound.go b/internal/outbound/outbound.go new file mode 100644 index 0000000..6f20c47 --- /dev/null +++ b/internal/outbound/outbound.go @@ -0,0 +1,54 @@ +// Package outbound is the reverse materializer and the pipe that carries it +// (task 15, decisions 12/15/16): social.coves.* records already summarized into +// consume.Intents are translated into ActivityPub vocabulary, written to the +// split activities/deliveries queue, and delivered to community inboxes with +// per-actor HTTP signatures, per-community ordering, explicit causal +// dependencies, and at-least-once semantics. +// +// The three moving parts: +// +// - Translator renders a consume.Intent into a canonical AP activity +// (Create/Update/Delete{Note}, Like/Dislike/Undo). It owns the wire format +// — the Lemmy quirks (to⊇Public required on Notes, attributedTo a single +// string, source+content duality, no summary on Delete) live here. +// - Enqueuer translates at enqueue time and writes ONE outbound_activities +// row plus ONE outbound_deliveries row per target, INSIDE the caller's rev +// gate transaction so the enqueue commits with the gate advance or not at +// all. +// - Worker claims a delivery, rechecks consent, enforces causal gating, +// signs with the actor's key and POSTs to the community inbox, then records +// the terminal outcome under the delivery's fencing token. +package outbound + +import ( + "context" + + "tidepool/internal/ap" +) + +// SignerProvider yields the per-actor AP Signer a delivery is signed with. The +// worker signs each delivery as the PERSONA that authored the record, not as +// the service actor — Lemmy attributes the activity to the signing key's owner. +// personas.Service satisfies this (its actorSigner, exported for task 15). +type SignerProvider interface { + // SignerFor returns the Signer whose keyId is "{actorID}#main-key" for the + // persona minted under did. A DID with no minted actor is an error + // satisfying errors.IsNotFound. + SignerFor(ctx context.Context, did string) (*ap.Signer, error) +} + +// InboxResolver resolves a community's target inbox from its Group actor +// document, preferring endpoints.sharedInbox, cached with a TTL. On a +// 401/404/410 the worker asks it to re-resolve ONCE before poisoning, so an +// endpoint rotation is not mistaken for a dead inbox. +type InboxResolver interface { + // ResolveInbox returns the AP inbox URL for a community AP Group id. + ResolveInbox(ctx context.Context, communityAPID string) (inbox string, err error) +} + +// ActivitySender POSTs a signed activity to a remote inbox as a specific +// persona. *ap.Client satisfies it via SendActivityAs (the per-actor signing +// path, distinct from the service actor's configured-Signer SendActivity). +type ActivitySender interface { + SendActivityAs(ctx context.Context, signer *ap.Signer, inbox string, activity any) error +} diff --git a/internal/outbound/outer_acceptance_test.go b/internal/outbound/outer_acceptance_test.go new file mode 100644 index 0000000..965fb3f --- /dev/null +++ b/internal/outbound/outer_acceptance_test.go @@ -0,0 +1,563 @@ +package outbound + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/consume" + "tidepool/internal/identity" + "tidepool/internal/personas" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// The world this acceptance test builds. +const ( + outUserOrigin = "https://coves.social" + outCovesHost = "coves.social" + + // The bridged community, hosted on a fake Lemmy. Its Group actor document + // advertises endpoints.sharedInbox pointing back at itself. + outCommunityAPID = "https://lemmy.world/c/technology" + outLemmyHost = "lemmy.world" + outCommunityName = "technology" + outCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + outSharedInbox = "https://lemmy.world/c/technology/inbox" + + // The parent post the comment replies to — a BRIDGE-origin object (Tidepool + // federated it), so it causally gates its child until it is accepted. Its + // inReplyTo is our coves.social object URL. + outRootAuthorDID = "did:plc:ewvi7nxzyoun6zhxrhs64oiz" + outRootRKey = "3lzroot2222aa" + outRootCID = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + outRootATURI = "at://" + outRootAuthorDID + "/social.coves.community.postv2/" + outRootRKey + outRootAPID = outUserOrigin + "/ap/object/" + outRootAuthorDID + "/social.coves.community.postv2/" + outRootRKey + + // The commenter — a native Coves user minted lazily on first interaction. + outCommenterDID = "did:plc:7iza6de2dwap2sbkpav7c6c6" + outCommenterHandle = "alice.coves.social" + outCommentRKey = "3lzcmnt3333bb" + outCommentCID = "bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4" + outCommentRev = "3lzcmntrev001" + outCommentATURI = "at://" + outCommenterDID + "/social.coves.community.comment/" + outCommentRKey + outCommentAPID = outUserOrigin + "/ap/object/" + outCommenterDID + "/social.coves.community.comment/" + outCommentRKey +) + +// outKEK seals minted actors' AP RSA keys (32 bytes, AES-256). +var outKEK = []byte("0123456789abcdef0123456789abcdef") + +// TestOutboundDeliversASignedNativeComment is the OUTER acceptance test for +// task 15. +// +// GIVEN a bridged community (its Group doc served by a fake Lemmy that verifies +// the HTTP signatures it receives), a native commenter minted as a coves.social +// Person actor, and a CommentIntent enqueued THROUGH THE REAL ENQUEUER inside a +// transaction, WHEN the delivery worker runs, THEN: +// +// 1. the tx contract holds: an enqueue whose tx rolls back writes NO +// outbound_activities / outbound_deliveries rows; a committed one writes +// exactly one of each, pending; +// 2. the fake Lemmy receives EXACTLY ONE Create{Note} whose HTTP signature +// verifies against the actor document coves.social serves; +// 3. the inner Note is addressed the way Lemmy 0.19.20 demands: to ⊇ Public, +// cc ⊇ the community, audience = the community, attributedTo = the actor id +// as a SINGLE STRING, inReplyTo = the parent AP id, content AND source both +// carried; +// 4. the outbound_deliveries row reaches 'delivered'; +// 5. a REPLAY (redeliver) is deduped by activity id — the fake Lemmy's +// duplicate-activity response leaves the delivery 'delivered', never +// poisoned. +// +// No network: the only hosts the AP clients may dial (coves.social and +// lemmy.world) are rewritten onto the two httptest listeners. +// +// SEAMS this outer test drives through test doubles (production shapes GREEN +// must supply / wire): outbound.SignerProvider (a sealed-key unseal, mirroring +// personas.actorSigner), outbound.InboxResolver (reads endpoints.sharedInbox +// off the Group doc), outbound.ActivitySender (ap.Client.SendActivityAs), and +// the real outbound.Enqueuer / outbound.Translator / outbound.Worker. +func TestOutboundDeliversASignedNativeComment(t *testing.T) { + conn := outboundAcceptanceDB(t) + ctx := context.Background() + + // --- The user origin: personas.Service mints + serves the actor. --- + custodian, err := identity.NewCustodian(outKEK) + require.NoError(t, err, "build custodian") + svc, err := personas.New(personas.Options{DB: conn, Custodian: custodian, UserOrigin: outUserOrigin}) + require.NoError(t, err, "build personas service") + + actor, err := svc.CreateActorForDID(ctx, outCommenterDID, outCommenterHandle) + require.NoError(t, err, "mint the commenter's actor") + actorID := actor.ActorID + require.Equal(t, outUserOrigin+"/ap/actor/"+outCommenterDID, actorID) + + personasServer := httptest.NewServer(svc) + t.Cleanup(personasServer.Close) + + // --- The bridged community's home: a fake Lemmy. --- + lemmy := &fakeLemmy{ + host: outLemmyHost, + communityAPI: outCommunityAPID, + sharedInbox: outSharedInbox, + seen: map[string]bool{}, + } + lemmyServer := httptest.NewServer(lemmy) + t.Cleanup(lemmyServer.Close) + + // --- One AP client, both hosts rewritten onto the httptest listeners. --- + routes := map[string]string{ + outCovesHost: personasServer.Listener.Addr().String(), + outLemmyHost: lemmyServer.Listener.Addr().String(), + } + client := ap.NewClient(ap.ClientOptions{ + HTTPClient: &http.Client{Transport: hostRewrite{routes: routes}}, + }) + // The fake Lemmy verifies inbound signatures by fetching the signer's actor + // document off coves.social through this same client. + lemmy.verifier = ap.NewVerifier(client) + + // --- The parent post: a bridge-origin object, already accepted so its child + // is causally eligible. --- + seedAcceptedParent(t, conn) + + // --- Wire the real enqueuer + worker. --- + enqueuer, err := NewEnqueuer(EnqueuerOptions{ + DB: conn, + Translator: NewTranslator(outUserOrigin), + Inboxes: inboxResolver{client: client}, + Actors: store.NewAPActors(conn), + UserOrigin: outUserOrigin, + }) + require.NoError(t, err, "build enqueuer") + + worker, err := NewWorker(WorkerOptions{ + DB: conn, + Actors: store.NewAPActors(conn), + Signers: sealedSigners{actors: store.NewAPActors(conn), custodian: custodian}, + Inboxes: inboxResolver{client: client}, + Sender: client, + Lease: time.Minute, + }) + require.NoError(t, err, "build worker") + + intent := consume.CommentIntent{ + Op: "create", + ATURI: outCommentATURI, + ID: consume.ActivityID(outUserOrigin, outCommentATURI, "create", 0), + CommunityAPID: outCommunityAPID, + ParentAPID: outRootAPID, + Snapshot: commentSnapshot(t), + } + + // ------------------------------------------------------------------- + // 1a. Tx contract: a rolled-back enqueue leaves NOTHING behind. + // ------------------------------------------------------------------- + tx, err := conn.BeginTx(ctx, nil) + require.NoError(t, err) + require.NoError(t, enqueuer.EnqueueActivity(ctx, tx, outCommenterDID, outCommunityAPID, outRootATURI, intent), + "enqueue must write its activity + delivery on the given tx") + require.NoError(t, tx.Rollback()) + + assert.Zero(t, countRows(t, conn, "outbound_activities"), + "an enqueue whose gate tx rolls back must leave NO activity row — a replay could not "+ + "reproduce it under an advanced gate") + assert.Zero(t, countRows(t, conn, "outbound_deliveries"), + "...and no delivery row either") + + // ------------------------------------------------------------------- + // 1b. Tx contract: a committed enqueue writes exactly one of each. + // ------------------------------------------------------------------- + tx, err = conn.BeginTx(ctx, nil) + require.NoError(t, err) + require.NoError(t, enqueuer.EnqueueActivity(ctx, tx, outCommenterDID, outCommunityAPID, outRootATURI, intent)) + require.NoError(t, tx.Commit()) + + require.Equal(t, 1, countRows(t, conn, "outbound_activities"), + "a committed comment enqueue writes exactly one canonical activity") + require.Equal(t, 1, countRows(t, conn, "outbound_deliveries"), + "...and exactly one per-inbox delivery (v2 targets are single-community)") + + activities := store.NewOutboundActivities(conn) + storedActivity, err := activities.Get(ctx, intent.ID) + require.NoError(t, err, "the activity is keyed by the deterministic id the intent carries") + require.NotNil(t, storedActivity) + assert.Equal(t, outCommenterDID, storedActivity.ActorDID, "the signing persona rides the activity") + + deliveries := store.NewOutboundDeliveries(conn) + pending, err := deliveries.Get(ctx, intent.ID, outSharedInbox) + require.NoError(t, err, + "the delivery is addressed to the community's sharedInbox, resolved from its Group doc") + require.NotNil(t, pending) + assert.Equal(t, store.DeliveryStatePending, pending.State) + assert.Equal(t, outCommunityAPID, pending.OrderingKey, + "the ordering key is the community AP id — the per-community serial line") + + // ------------------------------------------------------------------- + // 2-4. Run the worker: one signed Create{Note} is delivered. + // ------------------------------------------------------------------- + drainWorker(t, ctx, worker) + + require.Equal(t, 1, lemmy.postCount(), + "exactly one activity must be POSTed to the community inbox, got %d", lemmy.postCount()) + + rec := lemmy.lastActivity(t) + assert.NoError(t, rec.verifyErr, + "the delivery's HTTP signature must verify against the actor document coves.social serves") + assert.Equal(t, actorID, rec.verifiedActorID, + "Verify attributes the signature to the minted persona") + assert.Equal(t, "Create", rec.body["type"], "a comment create federates as a Create activity") + + note := asMap(t, rec.body["object"], "Create.object") + assert.Equal(t, "Note", note["type"], "a comment renders as a Note") + + to := asStringSet(t, note["to"], "Note.to") + assert.Contains(t, to, ap.PublicAudience, + "Lemmy REQUIRES a Note's `to` to include as:Public (verify_is_public rejects otherwise)") + + cc := asStringSet(t, note["cc"], "Note.cc") + assert.Contains(t, cc, outCommunityAPID, "the community is cc'd (the announce_create_note shape)") + + assert.Equal(t, outCommunityAPID, note["audience"], + "audience = the community AP id, or Lemmy fetches every to/cc URL hunting for the community") + + attributedTo, isString := note["attributedTo"].(string) + assert.True(t, isString, + "attributedTo MUST be a single string — an array parses as the Peertube variant") + assert.Equal(t, actorID, attributedTo, "attributedTo is the persona's actor id") + + assert.Equal(t, outRootAPID, note["inReplyTo"], + "inReplyTo is the parent's coves.social object URL (a bridge-origin parent)") + + assert.NotEmpty(t, note["content"], "the HTML content is carried") + source := asMap(t, note["source"], "Note.source") + assert.NotEmpty(t, source["content"], "the markdown source is carried alongside the HTML") + assert.Equal(t, "text/markdown", source["mediaType"], + "Lemmy uses source verbatim when present (content/source duality)") + + delivered, err := deliveries.Get(ctx, intent.ID, outSharedInbox) + require.NoError(t, err) + require.NotNil(t, delivered) + assert.Equal(t, store.DeliveryStateDelivered, delivered.State, + "a delivered activity's row reaches the delivered state") + + // ------------------------------------------------------------------- + // 5. Replay: a redelivery of the same activity id is deduped, not poisoned. + // ------------------------------------------------------------------- + resetDeliveryToPending(t, conn, intent.ID, outSharedInbox) + drainWorker(t, ctx, worker) + + assert.Equal(t, 2, lemmy.postCount(), + "the redelivery POSTs the same activity again (dedupe is the peer's job, keyed on our id)") + + redelivered, err := deliveries.Get(ctx, intent.ID, outSharedInbox) + require.NoError(t, err) + require.NotNil(t, redelivered) + assert.Equal(t, store.DeliveryStateDelivered, redelivered.State, + "Lemmy's duplicate-activity response must classify as DELIVERED, never poisoned — a crash "+ + "between deliver and mark is expected and safe under at-least-once") +} + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +func outboundAcceptanceDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "outbound_deliveries", "outbound_activities", "outbound_objects", + "ap_actors", "communities") + return database +} + +// seedAcceptedParent writes the parent post's outbound state and stamps its +// accepted_at directly (SetAccepted is a task-15 seam under test), so the child +// comment is causally eligible: only a BRIDGE-origin parent gates, and only +// until it is accepted. +func seedAcceptedParent(t *testing.T, conn *sql.DB) { + t.Helper() + ctx := context.Background() + _, err := store.NewOutboundObjects(conn).Upsert(ctx, store.OutboundObject{ + ATURI: outRootATURI, + APObjectID: outRootAPID, + LastCID: outRootCID, + LastRev: "3lzhead000001", + CommunityDID: outCommunityDID, + CommunityAPID: outCommunityAPID, + TranslatedSnapshot: []byte(`{"type":"Page","name":"parent post"}`), + }) + require.NoError(t, err, "seed parent outbound object") + _, err = conn.ExecContext(ctx, + `UPDATE outbound_objects SET accepted_at = now() WHERE at_uri = $1`, outRootATURI) + require.NoError(t, err, "stamp the parent accepted (raw, so the causal gate opens)") +} + +// commentSnapshot is the durable state the consumer stored for the comment (the +// commentSnapshot shape from consume/comments.go): the record plus the resolved +// thread context the Translator renders the Note from. +func commentSnapshot(t *testing.T) []byte { + t.Helper() + snap, err := json.Marshal(map[string]any{ + "atUri": outCommentATURI, + "cid": outCommentCID, + "rev": outCommentRev, + "collection": "social.coves.community.comment", + "record": map[string]any{ + "$type": "social.coves.community.comment", + "reply": map[string]any{ + "root": map[string]any{"uri": outRootATURI, "cid": outRootCID}, + "parent": map[string]any{"uri": outRootATURI, "cid": outRootCID}, + }, + "content": "first reply from atproto", + "createdAt": "2026-08-12T10:00:00.000Z", + }, + "parentAtUri": outRootATURI, + "parentApId": outRootAPID, + "communityApId": outCommunityAPID, + }) + require.NoError(t, err) + return snap +} + +// drainWorker runs DeliverNext until the queue is empty. +func drainWorker(t *testing.T, ctx context.Context, worker *Worker) { + t.Helper() + for i := 0; i < 20; i++ { + worked, err := worker.DeliverNext(ctx) + require.NoError(t, err, "DeliverNext must not error on a healthy delivery") + if !worked { + return + } + } + t.Fatal("worker did not drain within 20 iterations") +} + +func resetDeliveryToPending(t *testing.T, conn *sql.DB, activityID, inbox string) { + t.Helper() + _, err := conn.ExecContext(context.Background(), ` + UPDATE outbound_deliveries + SET state = 'pending', delivered_at = NULL, claimed_until = NULL, + next_attempt_at = now() + WHERE activity_id = $1 AND target_inbox = $2`, activityID, inbox) + require.NoError(t, err, "reset the delivery to pending to force a redelivery") +} + +func countRows(t *testing.T, conn *sql.DB, table string) int { + t.Helper() + var n int + require.NoError(t, conn.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM `+table).Scan(&n)) + return n +} + +// --------------------------------------------------------------------------- +// Test doubles for the outbound seams +// --------------------------------------------------------------------------- + +// hostRewrite sends requests for known hosts to their httptest listeners while +// preserving the request URL and Host, so the AP client — and the signatures it +// produces and verifies — believe they are talking to the real origins. It is +// deliberately NOT an *http.Transport, so ap.NewClient's guardedTransport passes +// it through unchanged. Anything not routed is refused: this test never touches +// the network. +type hostRewrite struct { + routes map[string]string +} + +func (rt hostRewrite) RoundTrip(req *http.Request) (*http.Response, error) { + target, ok := rt.routes[strings.ToLower(req.URL.Hostname())] + if !ok { + return nil, fmt.Errorf("refusing outbound request to %s: no route", req.URL) + } + clone := req.Clone(req.Context()) + clone.Host = req.URL.Host // preserve the authority the server verifies against + clone.URL.Scheme = "http" + clone.URL.Host = target + return http.DefaultTransport.RoundTrip(clone) +} + +// sealedSigners is the SignerProvider: it unseals the persona's AP RSA key, +// mirroring personas.actorSigner (the exported signer accessor is a task-13/15 +// seam GREEN wires — the test unseals directly to stay self-contained). +type sealedSigners struct { + actors store.APActors + custodian *identity.Custodian +} + +func (s sealedSigners) SignerFor(ctx context.Context, did string) (*ap.Signer, error) { + actor, err := s.actors.GetByDID(ctx, did) + if err != nil { + return nil, err + } + key, err := s.custodian.DecryptActorRSAKey(did, actor.RSAKeySealed) + if err != nil { + return nil, err + } + return ap.NewSigner(actor.ActorID+"#main-key", key), nil +} + +// inboxResolver reads endpoints.sharedInbox off the community's served Group +// document (the InboxResolver seam; a production one caches with a TTL). +type inboxResolver struct { + client *ap.Client +} + +func (r inboxResolver) ResolveInbox(ctx context.Context, communityAPID string) (string, error) { + doc, err := r.client.FetchActor(ctx, communityAPID) + if err != nil { + return "", err + } + inbox := doc.SharedInboxOrInbox() + if inbox == "" { + return "", fmt.Errorf("community %s advertises no inbox", communityAPID) + } + return inbox, nil +} + +// --------------------------------------------------------------------------- +// The fake Lemmy +// --------------------------------------------------------------------------- + +type receivedActivity struct { + body map[string]any + verifiedActorID string + verifyErr error +} + +type fakeLemmy struct { + host string + communityAPI string + sharedInbox string + + verifier *ap.Verifier + + mu sync.Mutex + received []receivedActivity + seen map[string]bool // activity ids already accepted (dedupe) +} + +func (l *fakeLemmy) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet && r.URL.Path == "/c/"+outCommunityName: + l.serveGroup(w) + case r.Method == http.MethodPost && r.URL.Path == "/c/"+outCommunityName+"/inbox": + l.serveInbox(w, r) + default: + http.Error(w, "not found", http.StatusNotFound) + } +} + +func (l *fakeLemmy) serveGroup(w http.ResponseWriter) { + doc := map[string]any{ + "@context": "https://www.w3.org/ns/activitystreams", + "id": l.communityAPI, + "type": "Group", + "preferredUsername": outCommunityName, + "inbox": l.sharedInbox, + "endpoints": map[string]any{"sharedInbox": l.sharedInbox}, + } + w.Header().Set("Content-Type", ap.ContentTypeActivityJSON) + _ = json.NewEncoder(w).Encode(doc) +} + +func (l *fakeLemmy) serveInbox(w http.ResponseWriter, r *http.Request) { + body := make([]byte, r.ContentLength) + if r.ContentLength > 0 { + _, _ = readFull(r, body) + } + + actorID, verifyErr := l.verifier.Verify(r.Context(), r, body) + + var parsed map[string]any + _ = json.Unmarshal(body, &parsed) + + l.mu.Lock() + l.received = append(l.received, receivedActivity{body: parsed, verifiedActorID: actorID, verifyErr: verifyErr}) + activityID, _ := parsed["id"].(string) + duplicate := l.seen[activityID] + if activityID != "" { + l.seen[activityID] = true + } + l.mu.Unlock() + + if duplicate { + // Lemmy's received_activity dedupe rejects an activity id it already + // processed. The exact wire code/message is a GREEN classification + // detail; the pinned CONTRACT is that the sender treats it as DELIVERED. + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"activity was already received"}`)) + return + } + w.WriteHeader(http.StatusAccepted) +} + +func (l *fakeLemmy) postCount() int { + l.mu.Lock() + defer l.mu.Unlock() + return len(l.received) +} + +func (l *fakeLemmy) lastActivity(t *testing.T) receivedActivity { + t.Helper() + l.mu.Lock() + defer l.mu.Unlock() + require.NotEmpty(t, l.received, "the fake Lemmy received no activity") + return l.received[len(l.received)-1] +} + +// readFull reads exactly len(buf) bytes from the request body. +func readFull(r *http.Request, buf []byte) (int, error) { + total := 0 + for total < len(buf) { + n, err := r.Body.Read(buf[total:]) + total += n + if err != nil { + return total, err + } + } + return total, nil +} + +// --------------------------------------------------------------------------- +// JSON assertion helpers +// --------------------------------------------------------------------------- + +func asMap(t *testing.T, v any, what string) map[string]any { + t.Helper() + m, ok := v.(map[string]any) + require.True(t, ok, "%s must be a JSON object, got %T", what, v) + return m +} + +// asStringSet coerces an AP addressing field (a string or an array of strings) +// into a set for ⊇ assertions. +func asStringSet(t *testing.T, v any, what string) []string { + t.Helper() + switch typed := v.(type) { + case string: + return []string{typed} + case []any: + out := make([]string, 0, len(typed)) + for _, item := range typed { + s, ok := item.(string) + require.True(t, ok, "%s array must hold strings, got %T", what, item) + out = append(out, s) + } + return out + default: + require.Failf(t, "bad addressing shape", "%s must be a string or array of strings, got %T", what, v) + return nil + } +} diff --git a/internal/outbound/translator.go b/internal/outbound/translator.go new file mode 100644 index 0000000..6c24108 --- /dev/null +++ b/internal/outbound/translator.go @@ -0,0 +1,253 @@ +package outbound + +import ( + "encoding/json" + "fmt" + "strings" + + "tidepool/internal/ap" + "tidepool/internal/apobject" + "tidepool/internal/consume" + "tidepool/internal/errors" +) + +// contextActivityStreams is the AS2 context every activity we emit carries at +// its top level. The Note/Page/vote shapes use only core AS2 terms +// (attributedTo, source, mediaType, audience, Like/Dislike/Undo), so the plain +// context is sufficient for Lemmy 0.19.20 to parse them. +const contextActivityStreams = apobject.ContextActivityStreams + +// TranslatedActivity is a rendered AP activity ready to persist and deliver: +// its canonical wire form (Payload, byte-stable once stored) plus the coarse +// facts the queue indexes on. +type TranslatedActivity struct { + // ActivityID is the deterministic id the intent carries (consume.ActivityID). + ActivityID string + // Kind is the AP activity type: Create, Update, Delete, Like, Dislike, Undo. + Kind string + // Payload is the canonical wire activity JSON. + Payload []byte + // ParentATURI is the causal dependency copied onto the activity row. + ParentATURI string +} + +// Translator renders a consume.Intent into an AP activity. It owns the wire +// format and every Lemmy quirk (verified against Lemmy 0.19.20): a Note's `to` +// MUST include as:Public with the community in cc, a Page carries the community +// in `to`, `attributedTo` is a SINGLE STRING, HTML `content` AND +// `source:{content,mediaType:text/markdown}` are both carried, `audience` is +// the community AP id, a Delete carries NO `summary`, and an Undo embeds the +// full inner vote object. +type Translator struct { + userOrigin string +} + +// NewTranslator builds a Translator that mints ids and self-references under +// userOrigin (AP_USER_ORIGIN). +func NewTranslator(userOrigin string) *Translator { + return &Translator{userOrigin: userOrigin} +} + +// Translate renders one intent into a canonical activity, addressed and signed +// as actorID (the persona's AP actor id, a single string). It hand-builds the +// wire JSON from the intent's already-resolved ids and its snapshot's raw +// record — no DB lookups and no round-trip through ap.Object, so the byte shape +// is exactly what Lemmy demanded. +func (t *Translator) Translate(actorID string, intent consume.Intent) (*TranslatedActivity, error) { + if actorID == "" { + return nil, errors.NewValidationError("actorID", "must not be empty") + } + switch typed := intent.(type) { + case consume.CommentIntent: + return t.comment(actorID, typed) + case consume.PostIntent: + return t.post(actorID, typed) + case consume.VoteIntent: + return t.vote(actorID, typed) + default: + return nil, errors.NewValidationError("intent", fmt.Sprintf("no translation for %T", intent)) + } +} + +// comment renders a native comment as Create/Update{Note} (create/update) or a +// bare Delete (delete). +func (t *Translator) comment(actorID string, intent consume.CommentIntent) (*TranslatedActivity, error) { + if intent.CommunityAPID == "" { + return nil, errors.NewValidationError("communityApId", "must not be empty") + } + snap, err := apobject.ParseSnapshot(intent.Snapshot) + if err != nil { + return nil, err + } + atURI, _ := snap["atUri"].(string) + if atURI == "" { + return nil, errors.NewValidationError("snapshot.atUri", "must not be empty") + } + objectURL := t.objectURL(atURI) + parentATURI, _ := snap["parentAtUri"].(string) + + if intent.Op == "delete" { + return t.finish(intent.ID, "Delete", parentATURI, + t.deleteActivity(intent.ID, actorID, intent.CommunityAPID, objectURL)) + } + + record, _ := snap["record"].(map[string]any) + note := apobject.BuildNote(actorID, intent.CommunityAPID, intent.ParentAPID, objectURL, record) + kind := "Create" + if intent.Op == "update" { + kind = "Update" + } + return t.finish(intent.ID, kind, parentATURI, + t.wrapActivity(kind, intent.ID, actorID, intent.CommunityAPID, note)) +} + +// post renders a native post as Create/Update{Page} or a bare Delete. +func (t *Translator) post(actorID string, intent consume.PostIntent) (*TranslatedActivity, error) { + if intent.CommunityAPID == "" { + return nil, errors.NewValidationError("communityApId", "must not be empty") + } + snap, err := apobject.ParseSnapshot(intent.Snapshot) + if err != nil { + return nil, err + } + atURI, _ := snap["atUri"].(string) + if atURI == "" { + return nil, errors.NewValidationError("snapshot.atUri", "must not be empty") + } + objectURL := t.objectURL(atURI) + + if intent.Op == "delete" { + // A post has no causal parent, so the activity row carries none. + return t.finish(intent.ID, "Delete", "", + t.deleteActivity(intent.ID, actorID, intent.CommunityAPID, objectURL)) + } + + record, _ := snap["record"].(map[string]any) + page, err := apobject.BuildPage(actorID, intent.CommunityAPID, objectURL, record) + if err != nil { + return nil, err + } + kind := "Create" + if intent.Op == "update" { + kind = "Update" + } + return t.finish(intent.ID, kind, "", + t.wrapActivity(kind, intent.ID, actorID, intent.CommunityAPID, page)) +} + +// vote renders a Like/Dislike (create) or the Undo of one (undo). An Undo +// embeds the FULL inner vote object reconstructed from state — a bare-URL inner +// object fails Lemmy's parse, so the inner {type, id, actor, object} is spelled +// out. +func (t *Translator) vote(actorID string, intent consume.VoteIntent) (*TranslatedActivity, error) { + if intent.CommunityAPID == "" { + return nil, errors.NewValidationError("communityApId", "must not be empty") + } + if intent.SubjectAPID == "" { + return nil, errors.NewValidationError("subjectApId", "must not be empty") + } + voteType, err := voteActivityType(intent.Direction) + if err != nil { + return nil, err + } + community := intent.CommunityAPID + + if intent.Op == "undo" { + if intent.InnerActivityID == "" { + return nil, errors.NewValidationError("innerActivityId", "an Undo must name the activity it withdraws") + } + activity := map[string]any{ + "@context": contextActivityStreams, + "id": intent.ID, + "type": "Undo", + "actor": actorID, + "to": []string{ap.PublicAudience}, + "cc": []string{community}, + "audience": community, + "object": map[string]any{ + "type": voteType, + "id": intent.InnerActivityID, + "actor": actorID, + "object": intent.SubjectAPID, + "audience": community, + }, + } + return t.finish(intent.ID, "Undo", "", activity) + } + + activity := map[string]any{ + "@context": contextActivityStreams, + "id": intent.ID, + "type": voteType, + "actor": actorID, + "object": intent.SubjectAPID, + "to": []string{ap.PublicAudience}, + "cc": []string{community}, + "audience": community, + } + return t.finish(intent.ID, voteType, "", activity) +} + +// wrapActivity wraps a rendered object (Note/Page) in the outer Create/Update +// activity, addressed to Public with the community cc'd and set as audience. +func (t *Translator) wrapActivity(kind, id, actorID, communityAPID string, object map[string]any) map[string]any { + return map[string]any{ + "@context": contextActivityStreams, + "id": id, + "type": kind, + "actor": actorID, + "to": []string{ap.PublicAudience}, + "cc": []string{communityAPID}, + "audience": communityAPID, + "object": object, + } +} + +// deleteActivity is a self-delete: a Delete of the bare object URL and NOTHING +// else. Lemmy reads a `summary` on a Delete as a MOD-REMOVAL reason, so a +// self-delete must omit it or it looks like moderation. +func (t *Translator) deleteActivity(id, actorID, communityAPID, objectURL string) map[string]any { + return map[string]any{ + "@context": contextActivityStreams, + "id": id, + "type": "Delete", + "actor": actorID, + "to": []string{ap.PublicAudience}, + "cc": []string{communityAPID}, + "audience": communityAPID, + "object": objectURL, + } +} + +// finish marshals the built activity into the canonical payload. +func (t *Translator) finish(activityID, kind, parentATURI string, activity map[string]any) (*TranslatedActivity, error) { + payload, err := json.Marshal(activity) + if err != nil { + return nil, fmt.Errorf("translate %s: encode activity: %w", activityID, err) + } + return &TranslatedActivity{ + ActivityID: activityID, + Kind: kind, + Payload: payload, + ParentATURI: parentATURI, + }, nil +} + +// objectURL is the AP id a native record federates as: the served +// /ap/object/{did}/{collection}/{rkey} URL, which is exactly the at-uri's three +// parts hung under the origin. +func (t *Translator) objectURL(atURI string) string { + return t.userOrigin + "/ap/object/" + strings.TrimPrefix(atURI, "at://") +} + +// voteActivityType maps a vote direction to its AP activity type. +func voteActivityType(direction string) (string, error) { + switch direction { + case "up": + return "Like", nil + case "down": + return "Dislike", nil + default: + return "", errors.NewValidationError("direction", "must be up or down, got "+direction) + } +} diff --git a/internal/outbound/translator_page_test.go b/internal/outbound/translator_page_test.go new file mode 100644 index 0000000..113e269 --- /dev/null +++ b/internal/outbound/translator_page_test.go @@ -0,0 +1,242 @@ +package outbound + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/consume" +) + +// Task 15 cycle B (Page half): the Translator renders a native POST as +// Create/Update{Page} or a bare Delete. The Page/Note split is the crux — +// verified against internal/ap/testdata/announce_create_page_lemmy_world.json: +// a Page carries the community in `to` (alongside as:Public), where a Note +// carries it in `cc`. name is REQUIRED (Lemmy rejects a titleless Page), a link +// embed becomes attachment [{type:Link,href}], nsfw becomes sensitive, audience +// is the community, and attributedTo is a SINGLE STRING. +// +// The producer of a PostIntent is task 16's acceptance engine; task 15 owns +// this translation, so it is pinned here at the translator boundary. + +const ( + pageUserOrigin = "https://coves.social" + pageCommunityAPI = "https://lemmy.world/c/technology" + pageActorID = "https://coves.social/ap/actor/did:plc:ewvi7nxzyoun6zhxrhs64oiz" + pageAuthorDID = "did:plc:ewvi7nxzyoun6zhxrhs64oiz" + pagePostRKey = "3lzpost22222aa" + pagePostCID = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + pagePostATURI = "at://" + pageAuthorDID + "/social.coves.community.postv2/" + pagePostRKey + pagePostAPID = pageUserOrigin + "/ap/object/" + pageAuthorDID + "/social.coves.community.postv2/" + pagePostRKey + pageLinkHref = "https://www.tomshardware.com/pc-components/dram/samsung-sk-hynix-and-micron-face-a-third-dram-price-fixing-lawsuit" + pageTitle = "Inside the history of DRAM price-fixing lawsuits" + pageBody = "two decades of failed cases, and why HBM changes the math" +) + +// postSnapshot builds the durable snapshot the acceptance engine stores for a +// post (the same envelope commentSnapshot uses): the postv2 record plus context. +func postSnapshot(t *testing.T, record map[string]any) []byte { + t.Helper() + snap, err := json.Marshal(map[string]any{ + "atUri": pagePostATURI, + "cid": pagePostCID, + "rev": "3lzpostrev0001", + "collection": "social.coves.community.postv2", + "record": record, + "communityApId": pageCommunityAPI, + }) + require.NoError(t, err) + return snap +} + +func linkPostRecord() map[string]any { + return map[string]any{ + "$type": "social.coves.community.postv2", + "community": "did:plc:44ybard66vv44zksje25o7dz", + "title": pageTitle, + "content": pageBody, + "embed": map[string]any{ + "$type": "social.coves.embed.external", + "external": map[string]any{"uri": pageLinkHref, "title": "Tom's Hardware"}, + }, + "createdAt": "2026-07-07T03:27:37.028Z", + } +} + +func TestTranslator_Page(t *testing.T) { + tr := NewTranslator(pageUserOrigin) + + cases := []struct { + name string + intent consume.PostIntent + wantKind string + validate func(t *testing.T, activity map[string]any) + }{ + { + name: "create link post -> Create{Page}", + intent: consume.PostIntent{ + Op: "create", + ATURI: pagePostATURI, + ID: consume.ActivityID(pageUserOrigin, pagePostATURI, "create", 0), + CommunityAPID: pageCommunityAPI, + Snapshot: postSnapshot(t, linkPostRecord()), + }, + wantKind: "Create", + validate: func(t *testing.T, activity map[string]any) { + page := mustMap(t, activity["object"], "object") + assert.Equal(t, "Page", page["type"], "a post renders as a Page, not a Note") + assert.Equal(t, pagePostAPID, page["id"], "the Page id is the served object URL") + + to := mustStringSet(t, page["to"], "Page.to") + assert.Contains(t, to, pageCommunityAPI, + "the Page/Note split: a Page carries the community in `to` (a Note carries it in cc)") + assert.Contains(t, to, ap.PublicAudience, "to must also include as:Public") + + cc := mustStringSet(t, page["cc"], "Page.cc") + assert.NotContains(t, cc, pageCommunityAPI, + "the community must NOT be in a Page's cc — that is the Note shape") + + assert.Equal(t, pageTitle, page["name"], + "name is REQUIRED and comes from the record's title") + assert.Equal(t, pageCommunityAPI, page["audience"], "audience = the community AP id") + + attributedTo, isString := page["attributedTo"].(string) + assert.True(t, isString, "attributedTo MUST be a single string, not an array") + assert.Equal(t, pageActorID, attributedTo) + + assert.NotEmpty(t, page["content"], "the HTML content is carried") + source := mustMap(t, page["source"], "Page.source") + assert.Equal(t, pageBody, source["content"], "the markdown source is carried verbatim") + assert.Equal(t, "text/markdown", source["mediaType"]) + + attach := mustSlice(t, page["attachment"], "Page.attachment") + require.NotEmpty(t, attach, "a link post carries a Link attachment") + link := mustMap(t, attach[0], "attachment[0]") + assert.Equal(t, "Link", link["type"], + "a link embed maps to attachment [{type:Link,href}] (Lemmy reads the FIRST attachment)") + assert.Equal(t, pageLinkHref, link["href"]) + + assert.Equal(t, false, page["sensitive"], "a non-nsfw post is sensitive:false") + }, + }, + { + name: "create nsfw text post -> sensitive", + intent: consume.PostIntent{ + Op: "create", + ATURI: pagePostATURI, + ID: consume.ActivityID(pageUserOrigin, pagePostATURI, "create", 0), + CommunityAPID: pageCommunityAPI, + Snapshot: postSnapshot(t, map[string]any{ + "$type": "social.coves.community.postv2", + "community": "did:plc:44ybard66vv44zksje25o7dz", + "title": "a spicy text post", + "content": "body text", + "labels": map[string]any{ + "$type": "com.atproto.label.defs#selfLabels", + "values": []any{map[string]any{"val": "nsfw"}}, + }, + "createdAt": "2026-07-07T03:27:37.028Z", + }), + }, + wantKind: "Create", + validate: func(t *testing.T, activity map[string]any) { + page := mustMap(t, activity["object"], "object") + assert.Equal(t, "a spicy text post", page["name"]) + assert.Equal(t, true, page["sensitive"], + "a post self-labelled nsfw renders sensitive:true") + }, + }, + { + name: "update -> Update{Page}", + intent: consume.PostIntent{ + Op: "update", + ATURI: pagePostATURI, + ID: consume.ActivityID(pageUserOrigin, pagePostATURI, "update", 1), + CommunityAPID: pageCommunityAPI, + Snapshot: postSnapshot(t, linkPostRecord()), + }, + wantKind: "Update", + validate: func(t *testing.T, activity map[string]any) { + page := mustMap(t, activity["object"], "object") + assert.Equal(t, "Page", page["type"], "an edit re-federates the same Page shape") + to := mustStringSet(t, page["to"], "Page.to") + assert.Contains(t, to, pageCommunityAPI, "the Page/Note split holds on Update too") + }, + }, + { + name: "delete -> Delete with NO summary", + intent: consume.PostIntent{ + Op: "delete", + ATURI: pagePostATURI, + ID: consume.ActivityID(pageUserOrigin, pagePostATURI, "delete", 1), + CommunityAPID: pageCommunityAPI, + Snapshot: postSnapshot(t, linkPostRecord()), + }, + wantKind: "Delete", + validate: func(t *testing.T, activity map[string]any) { + assert.Equal(t, pagePostAPID, activity["object"], + "a self-delete's object is the bare object URL") + _, hasSummary := activity["summary"] + assert.False(t, hasSummary, + "a self-delete carries NO summary: Lemmy reads a summary as a mod-removal reason") + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + out, err := tr.Translate(pageActorID, tc.intent) + require.NoError(t, err, "the Page translator must render %s", tc.name) + require.NotNil(t, out) + assert.Equal(t, tc.wantKind, out.Kind) + assert.Equal(t, tc.intent.ID, out.ActivityID) + + var activity map[string]any + require.NoError(t, json.Unmarshal(out.Payload, &activity), + "the payload must be valid activity JSON") + assert.Equal(t, tc.wantKind, activity["type"], "the outer activity type") + assert.Equal(t, pageActorID, activity["actor"], "the activity actor is the persona") + tc.validate(t, activity) + }) + } +} + +// ---- local JSON helpers (asMap/asStringSet live in the outer test file) ---- + +func mustMap(t *testing.T, v any, what string) map[string]any { + t.Helper() + m, ok := v.(map[string]any) + require.True(t, ok, "%s must be a JSON object, got %T", what, v) + return m +} + +func mustSlice(t *testing.T, v any, what string) []any { + t.Helper() + s, ok := v.([]any) + require.True(t, ok, "%s must be a JSON array, got %T", what, v) + return s +} + +func mustStringSet(t *testing.T, v any, what string) []string { + t.Helper() + switch typed := v.(type) { + case nil: + return nil + case string: + return []string{typed} + case []any: + out := make([]string, 0, len(typed)) + for _, item := range typed { + s, ok := item.(string) + require.True(t, ok, "%s array must hold strings, got %T", what, item) + out = append(out, s) + } + return out + default: + require.Failf(t, "bad shape", "%s must be a string or array, got %T", what, v) + return nil + } +} diff --git a/internal/outbound/worker.go b/internal/outbound/worker.go new file mode 100644 index 0000000..619bdf3 --- /dev/null +++ b/internal/outbound/worker.go @@ -0,0 +1,104 @@ +package outbound + +import ( + "context" + "database/sql" + "log/slog" + "time" + + "tidepool/internal/store" +) + +// DefaultLease bounds a delivery claim: long enough for a POST + retries, short +// enough that a crashed worker's delivery is re-claimable. +const DefaultLease = 2 * time.Minute + +// WorkerOptions configures a Worker. +type WorkerOptions struct { + // DB is the bridge database. + DB *sql.DB + // Activities / Deliveries are the split queue. Optional: nil from DB. + Activities store.OutboundActivities + Deliveries store.OutboundDeliveries + // Objects gates causally (a bridge-origin parent must be accepted before + // its child delivers) and is stamped accepted on delivery success. Optional. + Objects store.OutboundObjects + // Actors is the consent recheck at claim time: a disabled/paused actor's + // create/update is cancelled, not delivered (delete/undo are exempt). + // Optional: nil from DB. + Actors store.APActors + // Signers yields the per-actor Signer each delivery is signed with. + Signers SignerProvider + // Inboxes re-resolves a rotated inbox once before poisoning. + Inboxes InboxResolver + // Sender POSTs the signed activity. *ap.Client satisfies it. + Sender ActivitySender + // Lease overrides DefaultLease. + Lease time.Duration + // Logger receives per-delivery outcomes. Nil uses slog.Default(). + Logger *slog.Logger +} + +// Worker claims one delivery at a time and carries it to a terminal state. It +// is the at-least-once engine: a crash between POST and MarkDelivered redelivers +// (Lemmy dedupes on our stable activity id, and its duplicate-activity response +// is classified DELIVERED, not poisoned). +type Worker struct { + db *sql.DB + activities store.OutboundActivities + deliveries store.OutboundDeliveries + objects store.OutboundObjects + actors store.APActors + signers SignerProvider + inboxes InboxResolver + sender ActivitySender + lease time.Duration + logger *slog.Logger +} + +// NewWorker wires a Worker. +func NewWorker(opts WorkerOptions) (*Worker, error) { + logger := opts.Logger + if logger == nil { + logger = slog.Default() + } + lease := opts.Lease + if lease <= 0 { + lease = DefaultLease + } + activities := opts.Activities + if activities == nil { + activities = store.NewOutboundActivities(opts.DB) + } + deliveries := opts.Deliveries + if deliveries == nil { + deliveries = store.NewOutboundDeliveries(opts.DB) + } + objects := opts.Objects + if objects == nil { + objects = store.NewOutboundObjects(opts.DB) + } + return &Worker{ + db: opts.DB, + activities: activities, + deliveries: deliveries, + objects: objects, + actors: opts.Actors, + signers: opts.Signers, + inboxes: opts.Inboxes, + sender: opts.Sender, + lease: lease, + logger: logger, + }, nil +} + +// DeliverNext claims one processable delivery and carries it to a terminal +// state (delivered / poisoned / cancelled) or reschedules it. It returns +// worked=true when a delivery was claimed and handled, worked=false (nil error) +// when the queue held nothing claimable. +// +// STUB (task 15 RED): GREEN claims, rechecks consent (exempting delete/undo), +// enforces causal gating, signs and POSTs, then records the fenced outcome. +func (w *Worker) DeliverNext(ctx context.Context) (worked bool, err error) { + return false, nil +} diff --git a/internal/personas/personas.go b/internal/personas/personas.go index 1bb3e3d..1f014b3 100644 --- a/internal/personas/personas.go +++ b/internal/personas/personas.go @@ -81,7 +81,12 @@ type Service struct { // inboxHandler is the ingest inbox this origin's shared inbox dispatches // to. Nil means the route 404s. inboxHandler http.Handler - logger *slog.Logger + // outboundObjects / outboundActivities back the task-15 serving surface: + // GET /ap/object renders from the object snapshot, GET /ap/activity serves + // the canonical payload. Constructed from Options.DB (no new Options field). + outboundObjects store.OutboundObjects + outboundActivities store.OutboundActivities + logger *slog.Logger } // New builds a Service. The origin is canonicalized once here — config @@ -120,9 +125,11 @@ func New(opts Options) (*Service, error) { userOrigin: origin, userHost: host, - serviceActor: opts.ServiceActor, - inboxHandler: opts.InboxHandler, - logger: logger, + serviceActor: opts.ServiceActor, + inboxHandler: opts.InboxHandler, + outboundObjects: store.NewOutboundObjects(opts.DB), + outboundActivities: store.NewOutboundActivities(opts.DB), + logger: logger, }, nil } @@ -223,6 +230,14 @@ func suffixedLocalPart(base string, attempt int) string { return base + "-" + strconv.Itoa(attempt) } +// ActorSigner is the exported per-actor signer accessor (task 15 seam): the +// outbound delivery worker signs each activity as the persona that authored the +// record, not as the service actor. It is actorSigner promoted to the package +// surface; internal callers keep using the unexported form. +func (s *Service) ActorSigner(ctx context.Context, did string) (*ap.Signer, error) { + return s.actorSigner(ctx, did) +} + // actorSigner unseals a minted actor's RSA key and returns a Signer whose // keyID is "{actor_id}#main-key" — the same id the actor document publishes, // so a verifier that fetches the document finds the key it needs there. diff --git a/internal/personas/serving.go b/internal/personas/serving.go index d1a5957..61cb1cc 100644 --- a/internal/personas/serving.go +++ b/internal/personas/serving.go @@ -10,6 +10,7 @@ import ( "time" "tidepool/internal/ap" + "tidepool/internal/apobject" "tidepool/internal/errors" "tidepool/internal/store" ) @@ -23,6 +24,11 @@ const ( actorPathPrefix = "/ap/actor/" outboxSuffix = "/outbox" inboxPath = "/ap/inbox" + // objectPathPrefix and activityPathPrefix are the outbound serving surface + // (task 15): a peer re-fetches a native object or activity by the id the + // bridge minted under it. + objectPathPrefix = "/ap/object/" + activityPathPrefix = "/ap/activity/" // jrdContentType is WebFinger's media type (v1 precedent: // ingest/inbox.go's service-actor webfinger). @@ -78,6 +84,16 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } s.handleInbox(w, r) + case strings.HasPrefix(path, objectPathPrefix): + if !isGET(w, r) { + return + } + s.handleObject(w, r, strings.TrimPrefix(path, objectPathPrefix)) + case strings.HasPrefix(path, activityPathPrefix): + if !isGET(w, r) { + return + } + s.handleActivity(w, r, strings.TrimPrefix(path, activityPathPrefix)) case strings.HasPrefix(path, actorPathPrefix): rest := strings.TrimPrefix(path, actorPathPrefix) if !isGET(w, r) { @@ -97,6 +113,69 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } +// handleObject serves a native record as its AP object (Note/Page), rendered +// from the outbound_objects SNAPSHOT so it survives a PDS outage — Lemmy +// re-fetches a delivered object by id, and the snapshot is the byte-stable +// source the delivery itself was built from. A tombstoned row is 410 Gone (the +// record was deleted, and serving the stale body would resurrect it); an object +// we hold no state for is 404. Serving is bound to the origin's own host, like +// the actor document: the object id sits on this authority, and answering under +// another Host would publish a cross-authority claim. +func (s *Service) handleObject(w http.ResponseWriter, r *http.Request, rest string) { + if normalizeHost(r.Host) != s.userHost { + http.NotFound(w, r) + return + } + // rest is did/collection/rkey — the at-uri's three parts. + parts := strings.SplitN(rest, "/", 3) + if len(parts) != 3 || parts[0] == "" || parts[1] == "" || parts[2] == "" { + http.NotFound(w, r) + return + } + atURI := "at://" + parts[0] + "/" + parts[1] + "/" + parts[2] + + object, err := s.outboundObjects.GetByATURI(r.Context(), atURI) + if err != nil { + s.writeStoreError(w, r, err) + return + } + if object.IsTombstoned() { + http.Error(w, "gone", http.StatusGone) + return + } + + doc, err := apobject.RenderObject(s.userOrigin, object.TranslatedSnapshot) + if err != nil { + s.logger.Error("failed to render served object from snapshot", + "at_uri", atURI, "host", r.Host, "error", err) + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + writeJSON(w, ap.ContentTypeActivityJSON, doc) +} + +// handleActivity serves a canonical activity payload VERBATIM. Activities are +// immutable (objects are current): an already-served activity must not change +// when the object it created is later edited, so the stored payload is written +// byte-for-byte. An activity id we never minted is 404. +func (s *Service) handleActivity(w http.ResponseWriter, r *http.Request, hash string) { + if normalizeHost(r.Host) != s.userHost { + http.NotFound(w, r) + return + } + if hash == "" || strings.Contains(hash, "/") { + http.NotFound(w, r) + return + } + activity, err := s.outboundActivities.Get(r.Context(), s.userOrigin+activityPathPrefix+hash) + if err != nil { + s.writeStoreError(w, r, err) + return + } + w.Header().Set("Content-Type", ap.ContentTypeActivityJSON) + _, _ = w.Write(activity.Payload) +} + func isGET(w http.ResponseWriter, r *http.Request) bool { return requireMethod(w, r, http.MethodGet) } diff --git a/internal/personas/serving_object_test.go b/internal/personas/serving_object_test.go new file mode 100644 index 0000000..cb9a758 --- /dev/null +++ b/internal/personas/serving_object_test.go @@ -0,0 +1,161 @@ +package personas + +import ( + "context" + "database/sql" + "encoding/json" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/identity" + "tidepool/internal/store" + "tidepool/internal/testutil" +) + +// Task 15 cycle D: the user-origin object/activity serving surface. Lemmy +// re-fetches a native record by id, so coves.social must serve: +// +// - GET /ap/object/{did}/{collection}/{rkey} → the AP object rendered from the +// outbound_objects SNAPSHOT (survives PDS outages); a tombstoned row → 410; +// - GET /ap/activity/{hash} → the CANONICAL immutable payload from +// outbound_activities, byte-for-byte (an edited object must NOT change an +// already-served activity — activities are immutable, objects are current); +// - unknown object/activity → 404. + +const ( + serveObjectDID = "did:plc:7iza6de2dwap2sbkpav7c6c6" + serveObjectCollection = "social.coves.community.comment" + serveObjectRKey = "3lzserveobj001" + serveObjectATURI = "at://" + serveObjectDID + "/" + serveObjectCollection + "/" + serveObjectRKey + serveCommunityDID = "did:plc:44ybard66vv44zksje25o7dz" + serveCommunityAPID = "https://lemmy.world/c/technology" + serveActivityHash = "0f1e2d3c4b5a69788796a5b4c3d2e1f00f1e2d3c4b5a69788796a5b4c3d2e1f0" +) + +func servingObjectTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, "ap_actors", "outbound_objects", "outbound_activities") + return database +} + +func newServingService(t *testing.T, conn *sql.DB) *Service { + t.Helper() + custodian, err := identity.NewCustodian([]byte("0123456789abcdef0123456789abcdef")) + require.NoError(t, err) + svc, err := New(Options{DB: conn, Custodian: custodian, UserOrigin: userOrigin}) + require.NoError(t, err) + return svc +} + +func serveObjectAPID() string { + return userOrigin + "/ap/object/" + serveObjectDID + "/" + serveObjectCollection + "/" + serveObjectRKey +} + +func seedServedObject(t *testing.T, conn *sql.DB) { + t.Helper() + _, err := store.NewOutboundObjects(conn).Upsert(context.Background(), store.OutboundObject{ + ATURI: serveObjectATURI, + APObjectID: serveObjectAPID(), + CommunityDID: serveCommunityDID, + CommunityAPID: serveCommunityAPID, + TranslatedSnapshot: []byte(`{ + "atUri": "` + serveObjectATURI + `", + "collection": "` + serveObjectCollection + `", + "record": {"$type":"social.coves.community.comment","content":"served from snapshot", + "reply":{"root":{"uri":"at://x/y/z"},"parent":{"uri":"at://x/y/z"}}}, + "parentApId": "https://lemmy.world/post/1", + "communityApId": "` + serveCommunityAPID + `" + }`), + }) + require.NoError(t, err, "seed outbound_objects snapshot") +} + +func TestServing_ObjectRendersFromSnapshot(t *testing.T) { + conn := servingObjectTestDB(t) + svc := newServingService(t, conn) + seedServedObject(t, conn) + + path := "/ap/object/" + serveObjectDID + "/" + serveObjectCollection + "/" + serveObjectRKey + rec := serveOnHost(svc, userHost, http.MethodGet, path, + http.Header{"Accept": []string{ap.ContentTypeActivityJSON}}) + + require.Equal(t, http.StatusOK, rec.Code, + "GET %s (Host %s) must render the native object from its snapshot; body=%s", + path, userHost, rec.Body.String()) + assert.Contains(t, rec.Header().Get("Content-Type"), "activity+json", + "the object is served as application/activity+json") + + var doc map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &doc), + "the served object must be JSON, got %s", rec.Body.String()) + assert.Equal(t, serveObjectAPID(), doc["id"], + "the served object's id is the coves.social object URL") + assert.NotEmpty(t, doc["type"], "the rendered object carries an AP type (Note/Page)") +} + +func TestServing_TombstonedObjectIs410(t *testing.T) { + conn := servingObjectTestDB(t) + svc := newServingService(t, conn) + seedServedObject(t, conn) + + _, err := store.NewOutboundObjects(conn).Tombstone(context.Background(), serveObjectATURI) + require.NoError(t, err, "tombstone the served object") + + path := "/ap/object/" + serveObjectDID + "/" + serveObjectCollection + "/" + serveObjectRKey + rec := serveOnHost(svc, userHost, http.MethodGet, path, nil) + assert.Equal(t, http.StatusGone, rec.Code, + "a tombstoned outbound_objects row must serve 410 Gone, not the stale body") +} + +func TestServing_UnknownObjectIs404(t *testing.T) { + conn := servingObjectTestDB(t) + svc := newServingService(t, conn) + + path := "/ap/object/" + serveObjectDID + "/" + serveObjectCollection + "/nonexistent" + rec := serveOnHost(svc, userHost, http.MethodGet, path, nil) + assert.Equal(t, http.StatusNotFound, rec.Code, + "an object we hold no state for must 404, not 200 with an empty body") +} + +func TestServing_ActivityServesCanonicalPayloadVerbatim(t *testing.T) { + conn := servingObjectTestDB(t) + svc := newServingService(t, conn) + + activityID := userOrigin + "/ap/activity/" + serveActivityHash + payload := []byte(`{"@context":"https://www.w3.org/ns/activitystreams",` + + `"id":"` + activityID + `","type":"Create","actor":"` + serveObjectAPID() + `",` + + `"object":{"type":"Note","content":"original, pre-edit"}}`) + inserted, err := store.NewOutboundActivities(conn).Insert(context.Background(), store.OutboundActivity{ + ActivityID: activityID, + ActorDID: serveObjectDID, + Kind: "Create", + Payload: payload, + }) + require.NoError(t, err) + require.True(t, inserted) + + path := "/ap/activity/" + serveActivityHash + rec := serveOnHost(svc, userHost, http.MethodGet, path, + http.Header{"Accept": []string{ap.ContentTypeActivityJSON}}) + + require.Equal(t, http.StatusOK, rec.Code, + "GET %s must serve the canonical activity payload; body=%s", path, rec.Body.String()) + assert.Contains(t, rec.Header().Get("Content-Type"), "activity+json") + assert.JSONEq(t, string(payload), rec.Body.String(), + "the activity payload is served verbatim — activities are immutable, so an edited object "+ + "must NOT change an already-served activity") +} + +func TestServing_UnknownActivityIs404(t *testing.T) { + conn := servingObjectTestDB(t) + svc := newServingService(t, conn) + + rec := serveOnHost(svc, userHost, http.MethodGet, "/ap/activity/deadbeef", nil) + assert.Equal(t, http.StatusNotFound, rec.Code, + "an activity id we never minted must 404") +} diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go index a4f91fe..c3f540b 100644 --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -332,6 +332,15 @@ type OutboundObjects interface { // TombstoneTx is Tombstone on an existing transaction. A nil tx is an // error satisfying errors.IsValidation. TombstoneTx(ctx context.Context, tx *sql.Tx, atURI string) (*OutboundObject, error) + + // SetAccepted stamps accepted_at — the causal-gating marker (task 15, + // decision 15). Delivery SUCCESS sets it; a NULL accepted_at means the + // object has not yet been delivered to its community, which is what keeps a + // BRIDGE-origin child (a reply) ineligible until its parent lands. + // Stamping an already-accepted row preserves the original time (a + // redelivery must not move the causal boundary). A missing row is an error + // satisfying errors.IsNotFound. + SetAccepted(ctx context.Context, atURI string) error } // OutboundVotes persists the state an outbound Undo is rebuilt from (decision @@ -437,3 +446,90 @@ type Tombstones interface { // any observed redelivery horizon. Prune(ctx context.Context, cutoff time.Time) (int64, error) } + +// OutboundActivities persists the canonical, immutable wire payloads outbound +// deliveries fan out from (task 15, decision 15). One activity id maps to one +// payload byte-string that GET /ap/activity/{hash} serves and a redelivery +// re-sends verbatim; a peer dedupes on the stable id. The payload never +// changes once written — an edit is a NEW activity, not a rewrite. +type OutboundActivities interface { + // Insert idempotently writes one activity. It returns inserted=true when a + // new row was written and inserted=false (no error) when the activity id + // already existed: the ON CONFLICT DO NOTHING is deliberate — the payload + // of an activity a peer may already hold must never be overwritten. + Insert(ctx context.Context, activity OutboundActivity) (inserted bool, err error) + + // InsertTx is Insert on an existing transaction — the seam the enqueuer + // uses so the activity, its deliveries and the rev-gate advance land in ONE + // commit (an enqueue whose gate tx rolls back must leave no activity or + // delivery row). A nil tx is an error satisfying errors.IsValidation. + InsertTx(ctx context.Context, tx *sql.Tx, activity OutboundActivity) (inserted bool, err error) + + // Get returns the canonical activity for an id. A miss is an error + // satisfying errors.IsNotFound. + Get(ctx context.Context, activityID string) (*OutboundActivity, error) +} + +// OutboundDeliveries is the per-inbox delivery queue (task 15). It generalizes +// the inbox_events fenced work queue: claimed_until fencing, per-ordering-key +// serialization via a loose index scan, SKIP LOCKED concurrency. The ordering +// key is the community AP id, so all deliveries bound for one community form a +// single serial line. +type OutboundDeliveries interface { + // Enqueue writes one pending delivery keyed on (ActivityID, TargetInbox) + // and returns the stored row. A duplicate (activity, inbox) is an error + // satisfying errors.IsAlreadyExists. + Enqueue(ctx context.Context, delivery OutboundDelivery) (*OutboundDelivery, error) + + // EnqueueTx is Enqueue on an existing transaction — rides the enqueuer's + // gate tx. A nil tx is an error satisfying errors.IsValidation. + EnqueueTx(ctx context.Context, tx *sql.Tx, delivery OutboundDelivery) (*OutboundDelivery, error) + + // ClaimNext atomically claims the oldest processable delivery and + // increments its attempt counter. A delivery is processable when it is + // pending, past its next_attempt_at, unleased (or the lease expired), and — + // the per-community ordering guarantee — is the head (min Seq) of its + // ordering key among pending rows: a younger delivery on a key is invisible + // while an older PENDING sibling exists, and a delivered/poisoned/cancelled + // sibling stops blocking. An empty queue returns an error satisfying + // errors.IsNotFound. + // + // The returned delivery's ClaimedUntil is the fencing/claim token: the + // Mark*/Release methods require it so a worker whose lease expired and was + // re-claimed by another cannot clobber the newer attempt's outcome. + ClaimNext(ctx context.Context, lease time.Duration) (*OutboundDelivery, error) + + // MarkDelivered stamps the delivery delivered (delivered_at set, lease + // cleared), recording lastStatusCode. claimToken must equal the claim's + // ClaimedUntil. It returns exists=false for a missing (activity, inbox); + // applied=false (no error) when the claim was stale or the row already + // terminal, so the outcome was discarded without a clobber. + MarkDelivered(ctx context.Context, activityID, targetInbox string, lastStatusCode int, claimToken time.Time) (exists, applied bool, err error) + + // Release records a transient failure and schedules the retry (error class, + // status and excerpt stored, lease cleared, next_attempt_at set), leaving + // the delivery pending. claimToken must equal the claim's ClaimedUntil. + // Same (exists, applied) split as MarkDelivered. + Release(ctx context.Context, activityID, targetInbox, errorClass, excerpt string, lastStatusCode int, nextAttempt, claimToken time.Time) (exists, applied bool, err error) + + // MarkPoisoned permanently fails the delivery (state=poisoned, lease + // cleared, error class/status/excerpt stored). Poisoned rows are skipped by + // ClaimNext and stop blocking their ordering key. claimToken must equal the + // claim's ClaimedUntil. Same (exists, applied) split. + MarkPoisoned(ctx context.Context, activityID, targetInbox, errorClass, excerpt string, lastStatusCode int, claimToken time.Time) (exists, applied bool, err error) + + // CancelForActor moves every PENDING delivery of the actor's activities to + // cancelled (the consent/kill-switch withdrawal — a disabled or paused + // actor's create/update work is parked, never poisoned). Terminal + // deliveries are untouched. Returns how many rows were cancelled. + CancelForActor(ctx context.Context, actorDID string) (int64, error) + + // CancelForCommunity moves every PENDING delivery on an ordering key (a + // community AP id) to cancelled — a community deleted or unfollowed out + // from under pending work. Returns how many rows were cancelled. + CancelForCommunity(ctx context.Context, orderingKey string) (int64, error) + + // Get returns the delivery for an (activity, inbox) pair. A miss is an + // error satisfying errors.IsNotFound. + Get(ctx context.Context, activityID, targetInbox string) (*OutboundDelivery, error) +} diff --git a/internal/store/migrations_test.go b/internal/store/migrations_test.go index 0f19fad..8657c86 100644 --- a/internal/store/migrations_test.go +++ b/internal/store/migrations_test.go @@ -29,7 +29,9 @@ func TestMigrations_UpDownUp(t *testing.T) { -- task 14 (migration 018): the consumer's own state and the -- outbound state deletes are rebuilt from. 'consumer_cursors', 'jetstream_record_revs', 'jetstream_dead_letters', - 'outbound_objects', 'outbound_votes', 'federation_prefs') + 'outbound_objects', 'outbound_votes', 'federation_prefs', + -- task 15 (migration 020): the outbound delivery queue. + 'outbound_activities', 'outbound_deliveries') `).Scan(&remaining) require.NoError(t, err) assert.Zero(t, remaining, "down migrations must drop every Tidepool table") @@ -40,20 +42,43 @@ func TestMigrations_UpDownUp(t *testing.T) { // a migration whose Down forgets a table is caught only when its Up // created one. Assert the newest tables exist after the re-up so the two // halves stay in step. - for _, table := range []string{ - "consumer_cursors", "jetstream_record_revs", "jetstream_dead_letters", - "outbound_objects", "outbound_votes", "federation_prefs", + for _, tc := range []struct { + table string + migration string + }{ + {"consumer_cursors", "018"}, + {"jetstream_record_revs", "018"}, + {"jetstream_dead_letters", "018"}, + {"outbound_objects", "018"}, + {"outbound_votes", "018"}, + {"federation_prefs", "018"}, + {"outbound_activities", "020"}, + {"outbound_deliveries", "020"}, } { var exists bool err = database.QueryRowContext(ctx, ` SELECT EXISTS ( SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1 - )`, table).Scan(&exists) + )`, tc.table).Scan(&exists) require.NoError(t, err) - assert.True(t, exists, "migration 018 must create %q", table) + assert.True(t, exists, "migration %s must create %q", tc.migration, tc.table) } + // The causal-gating marker migration 020 ALTERs onto outbound_objects: + // NULL accepted_at is what keeps a bridge-origin child ineligible until its + // parent lands. + var acceptedAtExists bool + err = database.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'outbound_objects' + AND column_name = 'accepted_at' + )`).Scan(&acceptedAtExists) + require.NoError(t, err) + assert.True(t, acceptedAtExists, "migration 020 must add outbound_objects.accepted_at") + // Leave the schema usable and prove it is: exercise a write. repo := NewAPObjects(database) _, err = repo.PutMapping(ctx, testMapping()) @@ -105,6 +130,14 @@ func TestMigrations_UniqueConstraintNames(t *testing.T) { "outbound_votes_pkey", "outbound_votes_actor_subject_key", "federation_prefs_pkey", + + // Task 15 (migration 020). The activity id is globally unique (one + // canonical payload fans out to many inboxes); the delivery PK is the + // (activity, inbox) fan-out key; the partial queue index is the + // loose-index-scan support the generalized ClaimNext depends on. + "outbound_activities_pkey", + "outbound_deliveries_pkey", + "idx_outbound_deliveries_queue", } for _, name := range expected { var exists bool diff --git a/internal/store/models.go b/internal/store/models.go index d33d4fa..bdfa32d 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -320,3 +320,94 @@ type InboxEvent struct { ProcessedAt *time.Time Error string // last processing error; empty if none } + +// DeliveryState tracks a single per-inbox delivery attempt through its +// terminal fates (task 15, decision 15). pending is the only non-terminal +// state; delivered/poisoned/cancelled are all final. cancelled (not poisoned) +// is the kill-switch/consent outcome — a delivery parked because the actor +// opted out or a community was unfollowed, never a failure the operator must +// triage. +type DeliveryState string + +const ( + // DeliveryStatePending means the delivery is queued or backing off. + DeliveryStatePending DeliveryState = "pending" + // DeliveryStateDelivered means a peer accepted the activity (including + // Lemmy's duplicate-activity response, which is a success by our stable + // id). + DeliveryStateDelivered DeliveryState = "delivered" + // DeliveryStatePoisoned means the delivery permanently failed (a 4xx, an + // attempt-cap breach, an unaccepted or poisoned parent). + DeliveryStatePoisoned DeliveryState = "poisoned" + // DeliveryStateCancelled means a consent/kill-switch withdrawal parked the + // delivery: create/update for a disabled or paused actor, or a community + // unfollowed out from under pending work. Never a failure. + DeliveryStateCancelled DeliveryState = "cancelled" +) + +// Valid reports whether the value is a known delivery state. +func (s DeliveryState) Valid() bool { + switch s { + case DeliveryStatePending, DeliveryStateDelivered, DeliveryStatePoisoned, DeliveryStateCancelled: + return true + } + return false +} + +// OutboundActivity is the canonical, IMMUTABLE wire payload for one activity id +// (task 15, decision 15). One row fans out to many outbound_deliveries; GET +// /ap/activity/{hash} serves Payload verbatim, and a redelivery re-sends it +// byte-for-byte so a peer dedupes on the stable id. Its payload never changes +// once written — a later edit is a NEW activity, not a rewrite of this one. +type OutboundActivity struct { + // ActivityID is the deterministic AP activity id (consume.ActivityID) and + // the row's primary key. + ActivityID string + // ActorDID is the persona whose key signs every delivery of this activity. + ActorDID string + // Kind is the AP activity type: Create, Update, Delete, Like, Dislike, Undo. + Kind string + // Payload is the canonical wire activity JSON, byte-stable after first write. + Payload []byte + // ParentATURI is the causal dependency (decision 15): a delivery for this + // activity is ineligible until the parent's mapping is accepted. "" = none. + ParentATURI string + CreatedAt time.Time +} + +// OutboundDelivery is one delivery attempt of an activity to one inbox (task +// 15). It generalizes the inbox_events queue: ClaimedUntil is the same fencing +// token, OrderingKey (the community AP id) serializes deliveries per community, +// and the loose-index-scan head is the min-Seq pending row of a key. +type OutboundDelivery struct { + // Seq is the monotonic ordering column the per-key serialization descends. + Seq int64 + // ActivityID + TargetInbox are the composite primary key: one activity + // fans out to many inboxes. + ActivityID string + TargetInbox string + // OrderingKey is the community AP id — deliveries sharing it are handled + // strictly in Seq order. + OrderingKey string + // State is the delivery's fate (pending until terminal). + State DeliveryState + // Attempts counts how many times a worker claimed this delivery. + Attempts int + // NextAttemptAt is the retry-backoff schedule; claimable when <= now. + NextAttemptAt time.Time + // ClaimedUntil is the current worker lease AND the fencing/claim token; + // nil/past means unclaimed. MarkDelivered/Release/MarkPoisoned require it. + ClaimedUntil *time.Time + // DeliveredAt stamps the successful delivery. + DeliveredAt *time.Time + // LastStatusCode is the last HTTP status seen (nil before any attempt + // produced one). + LastStatusCode *int + // LastErrorClass is a coarse retry-taxonomy label (transport, 4xx, 5xx, + // duplicate, attempt_cap, parent_unaccepted, parent_poisoned). + LastErrorClass string + // ResponseExcerpt is a bounded sample of the peer's response body. + ResponseExcerpt string + CreatedAt time.Time + UpdatedAt time.Time +} diff --git a/internal/store/outbound_activities.go b/internal/store/outbound_activities.go new file mode 100644 index 0000000..b4558bd --- /dev/null +++ b/internal/store/outbound_activities.go @@ -0,0 +1,80 @@ +package store + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + + "tidepool/internal/errors" +) + +type postgresOutboundActivities struct { + db *sql.DB +} + +// NewOutboundActivities creates the postgres-backed outbound_activities +// repository. +func NewOutboundActivities(db *sql.DB) OutboundActivities { + return &postgresOutboundActivities{db: db} +} + +const outboundActivityColumns = ` + activity_id, actor_did, kind, payload, parent_at_uri, created_at` + +func (r *postgresOutboundActivities) Insert(ctx context.Context, activity OutboundActivity) (bool, error) { + return r.insert(ctx, r.db, activity) +} + +func (r *postgresOutboundActivities) InsertTx(ctx context.Context, tx *sql.Tx, activity OutboundActivity) (bool, error) { + if tx == nil { + return false, errors.NewValidationError("tx", "must not be nil") + } + return r.insert(ctx, tx, activity) +} + +func (r *postgresOutboundActivities) insert(ctx context.Context, q execer, activity OutboundActivity) (bool, error) { + // ON CONFLICT DO NOTHING makes the canonical payload immutable: a + // redelivery re-derives the SAME id and re-inserts, and the row a peer may + // already hold must never be rewritten under it. inserted reports whether a + // fresh row landed (rows affected == 1), never overwriting. + query := ` + INSERT INTO outbound_activities (activity_id, actor_did, kind, payload, parent_at_uri) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (activity_id) DO NOTHING` + + result, err := q.ExecContext(ctx, query, + activity.ActivityID, activity.ActorDID, activity.Kind, activity.Payload, activity.ParentATURI) + if err != nil { + return false, fmt.Errorf("insert outbound_activity %q: %w", activity.ActivityID, err) + } + affected, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("insert outbound_activity %q: rows affected: %w", activity.ActivityID, err) + } + return affected == 1, nil +} + +func (r *postgresOutboundActivities) Get(ctx context.Context, activityID string) (*OutboundActivity, error) { + query := `SELECT` + outboundActivityColumns + ` FROM outbound_activities WHERE activity_id = $1` + activity, err := scanOutboundActivity(r.db.QueryRowContext(ctx, query, activityID)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_activity", activityID) + } + return nil, fmt.Errorf("get outbound_activity %q: %w", activityID, err) + } + return activity, nil +} + +func scanOutboundActivity(row rowScanner) (*OutboundActivity, error) { + var activity OutboundActivity + err := row.Scan( + &activity.ActivityID, &activity.ActorDID, &activity.Kind, + &activity.Payload, &activity.ParentATURI, &activity.CreatedAt, + ) + if err != nil { + return nil, err + } + return &activity, nil +} diff --git a/internal/store/outbound_deliveries.go b/internal/store/outbound_deliveries.go new file mode 100644 index 0000000..3dd1044 --- /dev/null +++ b/internal/store/outbound_deliveries.go @@ -0,0 +1,293 @@ +package store + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + "time" + + "tidepool/internal/errors" +) + +type postgresOutboundDeliveries struct { + db *sql.DB +} + +// NewOutboundDeliveries creates the postgres-backed outbound_deliveries +// repository — the per-inbox delivery queue (task 15). +func NewOutboundDeliveries(db *sql.DB) OutboundDeliveries { + return &postgresOutboundDeliveries{db: db} +} + +// deliveryColumns is the SELECT/RETURNING list for a full OutboundDelivery row. +const deliveryColumns = ` + seq, activity_id, target_inbox, ordering_key, state, attempts, + next_attempt_at, claimed_until, delivered_at, last_status_code, + last_error_class, response_excerpt, created_at, updated_at` + +func (r *postgresOutboundDeliveries) Enqueue(ctx context.Context, delivery OutboundDelivery) (*OutboundDelivery, error) { + return r.enqueue(ctx, r.db, delivery) +} + +func (r *postgresOutboundDeliveries) EnqueueTx(ctx context.Context, tx *sql.Tx, delivery OutboundDelivery) (*OutboundDelivery, error) { + if tx == nil { + return nil, errors.NewValidationError("tx", "must not be nil") + } + return r.enqueue(ctx, tx, delivery) +} + +func (r *postgresOutboundDeliveries) enqueue(ctx context.Context, q execer, delivery OutboundDelivery) (*OutboundDelivery, error) { + // A fresh delivery is pending, unattempted, unclaimed: state, attempts, + // next_attempt_at and seq are all defaulted by the table. A duplicate + // (activity, inbox) pair violates the PK — mapped to AlreadyExists rather + // than pre-checked, since a SELECT-then-INSERT races a concurrent enqueue. + query := ` + INSERT INTO outbound_deliveries (activity_id, target_inbox, ordering_key) + VALUES ($1, $2, $3) + RETURNING` + deliveryColumns + + stored, err := scanOutboundDelivery(q.QueryRowContext(ctx, query, + delivery.ActivityID, delivery.TargetInbox, delivery.OrderingKey)) + if err != nil { + if _, ok := uniqueViolation(err); ok { + return nil, errors.NewConflictError("outbound_delivery", "activity_inbox", + delivery.ActivityID+" "+delivery.TargetInbox) + } + return nil, fmt.Errorf("enqueue outbound_delivery %q -> %q: %w", + delivery.ActivityID, delivery.TargetInbox, err) + } + return stored, nil +} + +func (r *postgresOutboundDeliveries) ClaimNext(ctx context.Context, lease time.Duration) (*OutboundDelivery, error) { + if lease <= 0 { + return nil, errors.NewValidationError("lease", "must be positive") + } + + // Generalizes inbox_events.ClaimNext (task 12's lesson): the candidate is + // the head of an ordering key — its min-seq PENDING row — that is also past + // its next_attempt_at and unleased. Per-community serialization is + // head-of-line: while an older pending sibling on a key exists (queued, + // leased, or backing off) every younger delivery on that key is invisible; + // a delivered/poisoned/cancelled sibling leaves the partial index and stops + // blocking. + // + // The heads are found by a recursive CTE emulating a loose index scan over + // idx_outbound_deliveries_queue (ordering_key, seq WHERE state='pending'): + // one index descent per DISTINCT pending key jumps to each key's head, so + // the claim is O(pending keys × log N) regardless of any one community's + // backlog depth — never O(backlog) as a per-row NOT EXISTS would be. They + // are materialized with ARRAY(...) — not a plain IN or a correlated EXISTS — + // so the planner fetches exactly those rows by seq. The outer SELECT + // re-applies every claimability predicate on the locked row (a claim + // committed between the CTE snapshot and the lock is then seen and skipped). + // FOR UPDATE ... SKIP LOCKED lets concurrent workers race without + // serializing on row locks; the UPDATE stamps the lease and counts the + // attempt atomically. + query := ` + UPDATE outbound_deliveries + SET claimed_until = CURRENT_TIMESTAMP + make_interval(secs => $1), + attempts = attempts + 1, + updated_at = now() + WHERE seq = ( + SELECT c.seq FROM outbound_deliveries c + WHERE c.seq = ANY (ARRAY( + WITH RECURSIVE key_heads AS ( + SELECT h.seq, h.ordering_key FROM ( + SELECT e.seq, e.ordering_key + FROM outbound_deliveries e + WHERE e.state = 'pending' + ORDER BY e.ordering_key, e.seq + LIMIT 1 + ) h + UNION ALL + SELECT n.seq, n.ordering_key FROM key_heads k + CROSS JOIN LATERAL ( + SELECT e.seq, e.ordering_key + FROM outbound_deliveries e + WHERE e.state = 'pending' + AND e.ordering_key > k.ordering_key + ORDER BY e.ordering_key, e.seq + LIMIT 1 + ) n + ) + SELECT seq FROM key_heads)) + AND c.state = 'pending' + AND c.next_attempt_at <= CURRENT_TIMESTAMP + AND (c.claimed_until IS NULL OR c.claimed_until <= CURRENT_TIMESTAMP) + ORDER BY c.seq + LIMIT 1 + FOR UPDATE OF c SKIP LOCKED) + RETURNING` + deliveryColumns + + delivery, err := scanOutboundDelivery(r.db.QueryRowContext(ctx, query, lease.Seconds())) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_delivery", "next claimable") + } + return nil, fmt.Errorf("claim next outbound_delivery: %w", err) + } + return delivery, nil +} + +func (r *postgresOutboundDeliveries) MarkDelivered(ctx context.Context, activityID, targetInbox string, lastStatusCode int, claimToken time.Time) (bool, bool, error) { + // Fencing: only the worker still holding the claim (claimed_until == + // claimToken) may record the outcome, and only while the row is + // non-terminal (state = 'pending'). A stale worker whose lease lapsed and + // was re-claimed no longer matches, writes 0 rows, and is reported + // applied=false so it cannot clobber the newer attempt. + query := ` + WITH updated AS ( + UPDATE outbound_deliveries + SET state = 'delivered', delivered_at = CURRENT_TIMESTAMP, + claimed_until = NULL, last_status_code = $3, updated_at = now() + WHERE activity_id = $1 AND target_inbox = $2 + AND state = 'pending' + AND claimed_until = $4 + RETURNING 1 + ) + SELECT + EXISTS (SELECT 1 FROM outbound_deliveries WHERE activity_id = $1 AND target_inbox = $2), + EXISTS (SELECT 1 FROM updated)` + + return r.markResult(ctx, "mark delivered", query, + activityID, targetInbox, lastStatusCode, claimToken.UTC()) +} + +func (r *postgresOutboundDeliveries) Release(ctx context.Context, activityID, targetInbox, errorClass, excerpt string, lastStatusCode int, nextAttempt, claimToken time.Time) (bool, bool, error) { + // Fencing + non-terminal guard: only the current claim holder reschedules + // (claimed_until == claimToken, state = 'pending'), so a stale worker's late + // release cannot resurrect a delivery a newer attempt already drove to a + // terminal state. The row stays pending with the lease cleared so a retry + // can re-claim after the backoff. + query := ` + WITH updated AS ( + UPDATE outbound_deliveries + SET claimed_until = NULL, next_attempt_at = $3, + last_error_class = $4, response_excerpt = $5, + last_status_code = $6, updated_at = now() + WHERE activity_id = $1 AND target_inbox = $2 + AND state = 'pending' + AND claimed_until = $7 + RETURNING 1 + ) + SELECT + EXISTS (SELECT 1 FROM outbound_deliveries WHERE activity_id = $1 AND target_inbox = $2), + EXISTS (SELECT 1 FROM updated)` + + return r.markResult(ctx, "release", query, + activityID, targetInbox, nextAttempt.UTC(), errorClass, excerpt, lastStatusCode, claimToken.UTC()) +} + +func (r *postgresOutboundDeliveries) MarkPoisoned(ctx context.Context, activityID, targetInbox, errorClass, excerpt string, lastStatusCode int, claimToken time.Time) (bool, bool, error) { + // Fencing + non-terminal guard: only the current claim holder may poison + // (claimed_until == claimToken, state = 'pending'). A poisoned delivery + // leaves the pending partial index and stops blocking its ordering key. + query := ` + WITH updated AS ( + UPDATE outbound_deliveries + SET state = 'poisoned', claimed_until = NULL, + last_error_class = $3, response_excerpt = $4, + last_status_code = $5, updated_at = now() + WHERE activity_id = $1 AND target_inbox = $2 + AND state = 'pending' + AND claimed_until = $6 + RETURNING 1 + ) + SELECT + EXISTS (SELECT 1 FROM outbound_deliveries WHERE activity_id = $1 AND target_inbox = $2), + EXISTS (SELECT 1 FROM updated)` + + return r.markResult(ctx, "poison", query, + activityID, targetInbox, errorClass, excerpt, lastStatusCode, claimToken.UTC()) +} + +// markResult runs a fenced (exists, applied) mark statement and maps a missing +// (activity, inbox) pair to NotFound. +func (r *postgresOutboundDeliveries) markResult(ctx context.Context, op, query string, args ...any) (bool, bool, error) { + var exists, applied bool + if err := r.db.QueryRowContext(ctx, query, args...).Scan(&exists, &applied); err != nil { + return false, false, fmt.Errorf("%s outbound_delivery: %w", op, err) + } + return exists, applied, nil +} + +func (r *postgresOutboundDeliveries) CancelForActor(ctx context.Context, actorDID string) (int64, error) { + // Consent/kill-switch withdrawal: park the actor's PENDING work as + // cancelled (never poisoned — this is not a failure). Terminal deliveries + // are left untouched. Joined through outbound_activities.actor_did. + query := ` + UPDATE outbound_deliveries d + SET state = 'cancelled', claimed_until = NULL, updated_at = now() + FROM outbound_activities a + WHERE d.activity_id = a.activity_id + AND a.actor_did = $1 + AND d.state = 'pending'` + + return r.cancel(ctx, "cancel outbound_deliveries for actor", query, actorDID) +} + +func (r *postgresOutboundDeliveries) CancelForCommunity(ctx context.Context, orderingKey string) (int64, error) { + // A community deleted or unfollowed out from under pending work: park every + // PENDING delivery on the ordering key as cancelled, leaving terminal rows. + query := ` + UPDATE outbound_deliveries + SET state = 'cancelled', claimed_until = NULL, updated_at = now() + WHERE ordering_key = $1 AND state = 'pending'` + + return r.cancel(ctx, "cancel outbound_deliveries for community", query, orderingKey) +} + +func (r *postgresOutboundDeliveries) cancel(ctx context.Context, op, query string, arg string) (int64, error) { + result, err := r.db.ExecContext(ctx, query, arg) + if err != nil { + return 0, fmt.Errorf("%s %q: %w", op, arg, err) + } + affected, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("%s %q: rows affected: %w", op, arg, err) + } + return affected, nil +} + +func (r *postgresOutboundDeliveries) Get(ctx context.Context, activityID, targetInbox string) (*OutboundDelivery, error) { + query := `SELECT` + deliveryColumns + ` + FROM outbound_deliveries WHERE activity_id = $1 AND target_inbox = $2` + delivery, err := scanOutboundDelivery(r.db.QueryRowContext(ctx, query, activityID, targetInbox)) + if err != nil { + if stderrors.Is(err, sql.ErrNoRows) { + return nil, errors.NewNotFoundError("outbound_delivery", activityID+" -> "+targetInbox) + } + return nil, fmt.Errorf("get outbound_delivery %q -> %q: %w", activityID, targetInbox, err) + } + return delivery, nil +} + +func scanOutboundDelivery(row rowScanner) (*OutboundDelivery, error) { + var delivery OutboundDelivery + var state string + var claimedUntil, deliveredAt sql.NullTime + var lastStatus sql.NullInt64 + err := row.Scan( + &delivery.Seq, &delivery.ActivityID, &delivery.TargetInbox, &delivery.OrderingKey, + &state, &delivery.Attempts, &delivery.NextAttemptAt, &claimedUntil, &deliveredAt, + &lastStatus, &delivery.LastErrorClass, &delivery.ResponseExcerpt, + &delivery.CreatedAt, &delivery.UpdatedAt, + ) + if err != nil { + return nil, err + } + delivery.State = DeliveryState(state) + if claimedUntil.Valid { + delivery.ClaimedUntil = &claimedUntil.Time + } + if deliveredAt.Valid { + delivery.DeliveredAt = &deliveredAt.Time + } + if lastStatus.Valid { + code := int(lastStatus.Int64) + delivery.LastStatusCode = &code + } + return &delivery, nil +} diff --git a/internal/store/outbound_delivery_test.go b/internal/store/outbound_delivery_test.go new file mode 100644 index 0000000..5d37cfb --- /dev/null +++ b/internal/store/outbound_delivery_test.go @@ -0,0 +1,602 @@ +package store + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/errors" + "tidepool/internal/testutil" +) + +// Task 15 cycle A: the outbound DELIVERY queue — canonical activities +// (immutable wire payloads) fanned out to per-inbox deliveries (decision 15). +// The delivery half generalizes the inbox_events fenced queue: claimed_until +// fencing, per-ordering-key serialization via a loose index scan, SKIP LOCKED. +// Every pin here is really about "can at-least-once delivery survive a crash +// between deliver and mark, without double-delivering or clobbering a re-claim?" + +// deliveryTestDB returns a migrated connection with task 15's tables emptied. +// The two are truncated in ONE statement so the FK (deliveries → activities) +// is satisfied without CASCADE, and RESTART IDENTITY resets the delivery seq so +// ordering assertions are deterministic across runs. +func deliveryTestDB(t *testing.T) *sql.DB { + t.Helper() + database := testutil.DB(t) + testutil.Truncate(t, database, + "outbound_deliveries", "outbound_activities", "outbound_objects") + return database +} + +const ( + delTargetInbox = "https://lemmy.world/c/technology/inbox" + delOrderingKey = "https://lemmy.world/c/technology" + delParentATURI = "at://" + testCommunityDID + "/social.coves.community.postv2/3lzpostaaaaaa" +) + +// Activity ids embed a 64-hex digest built at runtime (repeatHex is a func), so +// these are vars, not consts. +var ( + delActivityID = "https://coves.social/ap/activity/" + repeatHex('a') + delOtherActivityID = "https://coves.social/ap/activity/" + repeatHex('b') +) + +func testActivity() OutboundActivity { + return OutboundActivity{ + ActivityID: delActivityID, + ActorDID: testDID, + Kind: "Create", + Payload: []byte(`{"type":"Create","object":{"type":"Note","content":"hi"}}`), + ParentATURI: delParentATURI, + } +} + +func testDelivery() OutboundDelivery { + return OutboundDelivery{ + ActivityID: delActivityID, + TargetInbox: delTargetInbox, + OrderingKey: delOrderingKey, + } +} + +// seedActivity inserts the canonical activity a delivery fans out from (the FK +// target), failing the test if the insert did not report a fresh row. +func seedActivity(t *testing.T, repo OutboundActivities, activity OutboundActivity) { + t.Helper() + inserted, err := repo.Insert(context.Background(), activity) + require.NoError(t, err, "seed activity %s", activity.ActivityID) + require.True(t, inserted, "seed activity %s must be a fresh insert", activity.ActivityID) +} + +// --------------------------------------------------------------------------- +// outbound_activities — the immutable canonical payload +// --------------------------------------------------------------------------- + +func TestOutboundActivities_InsertIsIdempotentAndImmutable(t *testing.T) { + database := deliveryTestDB(t) + repo := NewOutboundActivities(database) + ctx := context.Background() + + inserted, err := repo.Insert(ctx, testActivity()) + require.NoError(t, err, "first insert of %s", delActivityID) + assert.True(t, inserted, "a fresh activity id must report inserted=true") + + // A redelivery re-derives the SAME id and re-inserts. The payload a peer may + // already hold must NOT be overwritten — ON CONFLICT DO NOTHING. + rewrite := testActivity() + rewrite.Payload = []byte(`{"type":"Create","object":{"type":"Note","content":"TAMPERED"}}`) + inserted, err = repo.Insert(ctx, rewrite) + require.NoError(t, err, "re-insert of an existing activity id is not an error") + assert.False(t, inserted, + "an already-stored activity id must report inserted=false, not overwrite the row") + + got, err := repo.Get(ctx, delActivityID) + require.NoError(t, err) + require.NotNil(t, got) + assert.JSONEq(t, `{"type":"Create","object":{"type":"Note","content":"hi"}}`, string(got.Payload), + "the canonical payload is immutable: a re-insert must not rewrite it (activities are "+ + "byte-stable across later edits)") + assert.Equal(t, testDID, got.ActorDID, "the signing persona is recorded") + assert.Equal(t, "Create", got.Kind) + assert.Equal(t, delParentATURI, got.ParentATURI, + "the causal dependency rides the activity row") +} + +func TestOutboundActivities_GetMissingIsNotFound(t *testing.T) { + database := deliveryTestDB(t) + repo := NewOutboundActivities(database) + + _, err := repo.Get(context.Background(), delActivityID) + require.Error(t, err, "an unknown activity id must not silently return a zero row") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) +} + +func TestOutboundActivities_InsertTxRidesTheTransaction(t *testing.T) { + database := deliveryTestDB(t) + repo := NewOutboundActivities(database) + ctx := context.Background() + + // Rolled back: the enqueue rides the rev-gate claim, so a gate rollback + // must leave NO activity row — otherwise a replay finds the activity but no + // gate advance. + tx, err := database.BeginTx(ctx, nil) + require.NoError(t, err) + inserted, err := repo.InsertTx(ctx, tx, testActivity()) + require.NoError(t, err, "InsertTx inside a transaction") + assert.True(t, inserted) + require.NoError(t, tx.Rollback()) + + _, err = repo.Get(ctx, delActivityID) + require.Error(t, err, "a rolled-back InsertTx must leave no activity row") + assert.True(t, errors.IsNotFound(err), "want NotFound after rollback, got %v", err) + + // Committed: the row lands. + tx, err = database.BeginTx(ctx, nil) + require.NoError(t, err) + _, err = repo.InsertTx(ctx, tx, testActivity()) + require.NoError(t, err) + require.NoError(t, tx.Commit()) + + got, err := repo.Get(ctx, delActivityID) + require.NoError(t, err, "a committed InsertTx must be visible") + require.NotNil(t, got) + + _, err = repo.InsertTx(ctx, nil, testActivity()) + require.Error(t, err, "InsertTx with a nil tx must not silently fall back to the pool") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) +} + +// --------------------------------------------------------------------------- +// outbound_deliveries — enqueue + read +// --------------------------------------------------------------------------- + +func TestOutboundDeliveries_EnqueueStartsPending(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + + stored, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err, "enqueue delivery for %s", delActivityID) + require.NotNil(t, stored, "Enqueue must return the stored row") + assert.Equal(t, DeliveryStatePending, stored.State, "a fresh delivery is pending") + assert.Equal(t, 0, stored.Attempts, "no attempt has been made yet") + assert.Nil(t, stored.ClaimedUntil, "a fresh delivery is unclaimed") + assert.Nil(t, stored.DeliveredAt) + assert.Positive(t, stored.Seq, "the monotonic ordering seq is assigned on enqueue") + + got, err := repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, delOrderingKey, got.OrderingKey) + + // The (activity, inbox) pair is the primary key: a second enqueue of the + // same pair must refuse, not spawn a duplicate delivery. + _, err = repo.Enqueue(ctx, testDelivery()) + require.Error(t, err, "a duplicate (activity, inbox) delivery must be refused") + assert.True(t, errors.IsAlreadyExists(err), "want AlreadyExists, got %v", err) +} + +func TestOutboundDeliveries_EnqueueTxRidesTheTransaction(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + + tx, err := database.BeginTx(ctx, nil) + require.NoError(t, err) + _, err = repo.EnqueueTx(ctx, tx, testDelivery()) + require.NoError(t, err) + require.NoError(t, tx.Rollback()) + + _, err = repo.Get(ctx, delActivityID, delTargetInbox) + require.Error(t, err, "a rolled-back EnqueueTx must leave no delivery row") + assert.True(t, errors.IsNotFound(err), "want NotFound after rollback, got %v", err) + + _, err = repo.EnqueueTx(ctx, nil, testDelivery()) + require.Error(t, err, "EnqueueTx with a nil tx must not silently fall back to the pool") + assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) +} + +// --------------------------------------------------------------------------- +// outbound_deliveries — ClaimNext, fencing, per-key serialization +// --------------------------------------------------------------------------- + +func TestOutboundDeliveries_ClaimNextStampsFencingAndBumpsAttempts(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + _, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err) + + claimed, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err, "ClaimNext must return the head-of-key pending delivery") + require.NotNil(t, claimed) + assert.Equal(t, delActivityID, claimed.ActivityID) + assert.Equal(t, delTargetInbox, claimed.TargetInbox) + assert.Equal(t, 1, claimed.Attempts, "ClaimNext increments the attempt counter") + require.NotNil(t, claimed.ClaimedUntil, + "the claim must stamp a fencing token (claimed_until) the Mark* methods verify") + assert.True(t, claimed.ClaimedUntil.After(time.Now().Add(-time.Second)), + "the lease is stamped into the future") + + // A claimed, unexpired delivery is invisible to a second claim (SKIP LOCKED + // + the lease guard) — no two workers may hold the same delivery. + _, err = repo.ClaimNext(ctx, time.Minute) + require.Error(t, err, "a claimed, unexpired delivery must be invisible to a second ClaimNext") + assert.True(t, errors.IsNotFound(err), "want NotFound (empty queue), got %v", err) +} + +func TestOutboundDeliveries_PerOrderingKeySerialization(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + // Two deliveries on the SAME ordering key (the same community). The younger + // one must be invisible while the older PENDING sibling is unclaimed. + seedActivity(t, activities, testActivity()) + older := testDelivery() + _, err := repo.Enqueue(ctx, older) + require.NoError(t, err) + + second := testActivity() + second.ActivityID = delOtherActivityID + seedActivity(t, activities, second) + younger := OutboundDelivery{ + ActivityID: delOtherActivityID, + TargetInbox: delTargetInbox, + OrderingKey: delOrderingKey, + } + _, err = repo.Enqueue(ctx, younger) + require.NoError(t, err) + + // The head of the key is the older delivery. + claimed, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + assert.Equal(t, delActivityID, claimed.ActivityID, + "ClaimNext returns the min-seq pending row of the ordering key") + + // While the older sibling is claimed (pending, leased), the younger is + // still blocked: per-community serialization is head-of-line. + _, err = repo.ClaimNext(ctx, time.Minute) + require.Error(t, err, + "a younger delivery on a key must be invisible while an older pending sibling exists") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) + + // Once the older sibling reaches a terminal state (delivered), the younger + // unblocks. + exists, applied, err := repo.MarkDelivered(ctx, claimed.ActivityID, claimed.TargetInbox, 202, *claimed.ClaimedUntil) + require.NoError(t, err) + require.True(t, exists) + require.True(t, applied, "the claim holder marks its delivery delivered") + + next, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err, "a delivered sibling stops blocking the ordering key") + require.NotNil(t, next) + assert.Equal(t, delOtherActivityID, next.ActivityID, + "the younger delivery becomes the head once the older is terminal") +} + +func TestOutboundDeliveries_MarkDeliveredFencing(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + _, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err) + + claimed, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed.ClaimedUntil) + staleToken := *claimed.ClaimedUntil + + // A STALE token (a worker whose lease lapsed and was re-claimed by another) + // must not clobber the newer attempt: applied=false, no error, state + // untouched. + wrongToken := staleToken.Add(-time.Hour) + exists, applied, err := repo.MarkDelivered(ctx, delActivityID, delTargetInbox, 202, wrongToken) + require.NoError(t, err) + assert.True(t, exists, "the delivery row exists") + assert.False(t, applied, "a stale fencing token must be a no-op, not a clobber") + + got, err := repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + assert.Equal(t, DeliveryStatePending, got.State, "the stale mark must not have moved state") + + // The current claim holder marks it delivered. + exists, applied, err = repo.MarkDelivered(ctx, delActivityID, delTargetInbox, 202, staleToken) + require.NoError(t, err) + assert.True(t, exists) + assert.True(t, applied, "the current claim holder's mark applies") + + got, err = repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + assert.Equal(t, DeliveryStateDelivered, got.State) + require.NotNil(t, got.DeliveredAt, "delivered_at is stamped") + require.NotNil(t, got.LastStatusCode) + assert.Equal(t, 202, *got.LastStatusCode, "the accepting status is recorded") + assert.Nil(t, got.ClaimedUntil, "the lease is cleared on a terminal outcome") + + // A missing (activity, inbox) reports exists=false. + exists, applied, err = repo.MarkDelivered(ctx, delOtherActivityID, delTargetInbox, 202, staleToken) + require.NoError(t, err) + assert.False(t, exists, "a missing delivery reports exists=false") + assert.False(t, applied) +} + +func TestOutboundDeliveries_ReleaseReschedulesUnderFencing(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + _, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err) + + claimed, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed.ClaimedUntil) + token := *claimed.ClaimedUntil + + next := time.Now().Add(30 * time.Second).UTC() + exists, applied, err := repo.Release(ctx, delActivityID, delTargetInbox, "5xx", "bad gateway", 502, next, token) + require.NoError(t, err) + assert.True(t, exists) + assert.True(t, applied, "the claim holder reschedules the retry") + + got, err := repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + assert.Equal(t, DeliveryStatePending, got.State, "a released delivery stays pending for retry") + assert.Nil(t, got.ClaimedUntil, "the lease is cleared so a retry can re-claim") + assert.Equal(t, "5xx", got.LastErrorClass, "the retry taxonomy label is recorded") + require.NotNil(t, got.LastStatusCode) + assert.Equal(t, 502, *got.LastStatusCode) + assert.WithinDuration(t, next, got.NextAttemptAt, time.Second, "the backoff schedule is stored") + + // A stale token cannot reschedule. + _, applied, err = repo.Release(ctx, delActivityID, delTargetInbox, "5xx", "x", 502, + next, token.Add(-time.Hour)) + require.NoError(t, err) + assert.False(t, applied, "a stale fencing token must not reschedule") +} + +func TestOutboundDeliveries_MarkPoisonedStopsBlockingKey(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + // Older poisons; younger on the same key must then unblock (a poisoned + // sibling stops blocking, exactly like inbox_events). + seedActivity(t, activities, testActivity()) + _, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err) + + second := testActivity() + second.ActivityID = delOtherActivityID + seedActivity(t, activities, second) + _, err = repo.Enqueue(ctx, OutboundDelivery{ + ActivityID: delOtherActivityID, + TargetInbox: delTargetInbox, + OrderingKey: delOrderingKey, + }) + require.NoError(t, err) + + claimed, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.Equal(t, delActivityID, claimed.ActivityID) + + exists, applied, err := repo.MarkPoisoned(ctx, delActivityID, delTargetInbox, "4xx", "bad request", 400, *claimed.ClaimedUntil) + require.NoError(t, err) + assert.True(t, exists) + assert.True(t, applied) + + got, err := repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + assert.Equal(t, DeliveryStatePoisoned, got.State) + assert.Equal(t, "4xx", got.LastErrorClass) + + next, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err, "a poisoned sibling must stop blocking its ordering key") + require.NotNil(t, next) + assert.Equal(t, delOtherActivityID, next.ActivityID) + + // A stale token cannot poison a delivery a retry already completed. + _, applied, err = repo.MarkPoisoned(ctx, delActivityID, delTargetInbox, "4xx", "x", 400, + claimed.ClaimedUntil.Add(-time.Hour)) + require.NoError(t, err) + assert.False(t, applied, "a stale fencing token must not poison") +} + +// --------------------------------------------------------------------------- +// outbound_deliveries — consent / kill-switch cancellation +// --------------------------------------------------------------------------- + +func TestOutboundDeliveries_CancelForActorParksPending(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + // Three deliveries on DISTINCT ordering keys, so head-of-line blocking never + // leaves a testDID delivery stranded pending by accident — the kill switch + // must cancel EVERY pending delivery of the actor, so the fixture must not + // smuggle in a second pending sibling and pretend it is terminal: + // - testDID / key A → driven to DELIVERED (terminal, must survive Cancel); + // - testDID / key B → left PENDING (the one Cancel parks); + // - testSecondDID/key C → left PENDING (out of scope, must survive). + deliveredKey := "https://lemmy.world/c/technology" + pendingKey := "https://lemmy.world/c/science" + otherKey := "https://lemmy.world/c/gaming" + + // testDID's soon-to-be-delivered activity, on key A. + delivered := testActivity() // actor testDID, id delActivityID + seedActivity(t, activities, delivered) + _, err := repo.Enqueue(ctx, OutboundDelivery{ + ActivityID: delActivityID, + TargetInbox: delTargetInbox + "/delivered", + OrderingKey: deliveredKey, + }) + require.NoError(t, err) + + // testDID's still-pending activity, on key B. + pending := testActivity() + pending.ActivityID = delOtherActivityID + seedActivity(t, activities, pending) + _, err = repo.Enqueue(ctx, OutboundDelivery{ + ActivityID: delOtherActivityID, + TargetInbox: delTargetInbox + "/pending", + OrderingKey: pendingKey, + }) + require.NoError(t, err) + + // A DIFFERENT actor's pending activity, on key C. + otherActor := OutboundActivity{ + ActivityID: "https://coves.social/ap/activity/" + repeatHex('c'), + ActorDID: testSecondDID, + Kind: "Create", + Payload: []byte(`{"type":"Create"}`), + } + seedActivity(t, activities, otherActor) + _, err = repo.Enqueue(ctx, OutboundDelivery{ + ActivityID: otherActor.ActivityID, + TargetInbox: delTargetInbox + "/other", + OrderingKey: otherKey, + }) + require.NoError(t, err) + + // Deliver testDID's key-A delivery so Cancel must leave that terminal row + // alone. It is the head of its own key, so ClaimNext reaches it; loop until + // it is the one claimed (distinct keys → whichever head has the lower seq + // comes first, and key A was enqueued first). + c, err := repo.ClaimNext(ctx, time.Minute) + require.NoError(t, err) + require.Equal(t, delActivityID, c.ActivityID, + "the first-enqueued delivery is the lowest-seq head across the keys") + _, applied, err := repo.MarkDelivered(ctx, c.ActivityID, c.TargetInbox, 202, *c.ClaimedUntil) + require.NoError(t, err) + require.True(t, applied) + + cancelled, err := repo.CancelForActor(ctx, testDID) + require.NoError(t, err, "CancelForActor sweeps EVERY pending delivery of the actor") + assert.EqualValues(t, 1, cancelled, + "exactly the one still-pending delivery for testDID is cancelled (the delivered one is "+ + "terminal, the other actor's is out of scope)") + + // The pending testDID delivery is parked. + got, err := repo.Get(ctx, delOtherActivityID, delTargetInbox+"/pending") + require.NoError(t, err) + assert.Equal(t, DeliveryStateCancelled, got.State, "a cancelled delivery is parked, not poisoned") + + // The DELIVERED testDID delivery is untouched — sparing a pending head would + // be a broken kill switch, but a TERMINAL row must never be rewritten. + got, err = repo.Get(ctx, delActivityID, delTargetInbox+"/delivered") + require.NoError(t, err) + assert.Equal(t, DeliveryStateDelivered, got.State, + "CancelForActor must not touch terminal deliveries") + + // The other actor's pending delivery is untouched. + got, err = repo.Get(ctx, otherActor.ActivityID, delTargetInbox+"/other") + require.NoError(t, err) + assert.Equal(t, DeliveryStatePending, got.State, "CancelForActor is scoped to the named actor") +} + +func TestOutboundDeliveries_CancelForCommunityParksPending(t *testing.T) { + database := deliveryTestDB(t) + activities := NewOutboundActivities(database) + repo := NewOutboundDeliveries(database) + ctx := context.Background() + + seedActivity(t, activities, testActivity()) + _, err := repo.Enqueue(ctx, testDelivery()) + require.NoError(t, err) + + // A delivery on a DIFFERENT ordering key must survive. + other := testActivity() + other.ActivityID = delOtherActivityID + seedActivity(t, activities, other) + otherKey := "https://lemmy.world/c/science" + _, err = repo.Enqueue(ctx, OutboundDelivery{ + ActivityID: delOtherActivityID, + TargetInbox: "https://lemmy.world/c/science/inbox", + OrderingKey: otherKey, + }) + require.NoError(t, err) + + cancelled, err := repo.CancelForCommunity(ctx, delOrderingKey) + require.NoError(t, err, "CancelForCommunity parks a community's pending work (unfollow/delete)") + assert.EqualValues(t, 1, cancelled) + + got, err := repo.Get(ctx, delActivityID, delTargetInbox) + require.NoError(t, err) + assert.Equal(t, DeliveryStateCancelled, got.State) + + got, err = repo.Get(ctx, delOtherActivityID, "https://lemmy.world/c/science/inbox") + require.NoError(t, err) + assert.Equal(t, DeliveryStatePending, got.State, + "CancelForCommunity is scoped to the named ordering key") +} + +// --------------------------------------------------------------------------- +// outbound_objects.accepted_at — the causal-gating marker (migration 020) +// --------------------------------------------------------------------------- + +func TestOutboundObjects_SetAcceptedStampsAndIsIdempotent(t *testing.T) { + database := deliveryTestDB(t) + repo := NewOutboundObjects(database) + ctx := context.Background() + + _, err := repo.Upsert(ctx, testOutboundObject()) + require.NoError(t, err) + + // accepted_at starts NULL: a not-yet-delivered object gates its bridge-origin + // children. + assert.Nil(t, rawAcceptedAt(t, database, testCommentATURI), + "a freshly written outbound object must have a NULL accepted_at (not yet delivered)") + + require.NoError(t, repo.SetAccepted(ctx, testCommentATURI), + "delivery success stamps accepted_at") + first := rawAcceptedAt(t, database, testCommentATURI) + require.NotNil(t, first, "SetAccepted must stamp accepted_at") + + // Idempotent: a redelivery re-stamps but must NOT move the causal boundary. + require.NoError(t, repo.SetAccepted(ctx, testCommentATURI)) + second := rawAcceptedAt(t, database, testCommentATURI) + require.NotNil(t, second) + assert.Equal(t, *first, *second, + "re-accepting preserves the original accepted_at: a redelivery must not move the marker") + + err = repo.SetAccepted(ctx, "at://did:plc:nobody/social.coves.community.comment/nope") + require.Error(t, err, "accepting an object we hold no state for is a bug, not a no-op") + assert.True(t, errors.IsNotFound(err), "want NotFound, got %v", err) +} + +// rawAcceptedAt reads outbound_objects.accepted_at directly so the pin does not +// depend on the struct/scan carrying the new column yet. +func rawAcceptedAt(t *testing.T, database *sql.DB, atURI string) *time.Time { + t.Helper() + var accepted sql.NullTime + err := database.QueryRowContext(context.Background(), + `SELECT accepted_at FROM outbound_objects WHERE at_uri = $1`, atURI).Scan(&accepted) + require.NoError(t, err, "read accepted_at for %s", atURI) + if !accepted.Valid { + return nil + } + return &accepted.Time +} diff --git a/internal/store/outbound_objects.go b/internal/store/outbound_objects.go index 9b6543a..ddd807a 100644 --- a/internal/store/outbound_objects.go +++ b/internal/store/outbound_objects.go @@ -134,6 +134,31 @@ func (r *postgresOutboundObjects) tombstone(ctx context.Context, q execer, atURI return object, nil } +// SetAccepted stamps accepted_at, the causal-gating marker (task 15). +// +// COALESCE preserves the original time on a redelivery: accepted_at is the +// causal boundary a bridge-origin child gates on, and a re-accept must not move +// it. A missing object is NotFound, not a no-op — accepting an object we hold no +// state for is a bug, since there is nothing whose children we could unblock. +func (r *postgresOutboundObjects) SetAccepted(ctx context.Context, atURI string) error { + query := ` + UPDATE outbound_objects + SET accepted_at = COALESCE(accepted_at, now()) + WHERE at_uri = $1` + result, err := r.db.ExecContext(ctx, query, atURI) + if err != nil { + return fmt.Errorf("set accepted_at for outbound_object %q: %w", atURI, err) + } + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("set accepted_at for outbound_object %q: rows affected: %w", atURI, err) + } + if affected == 0 { + return errors.NewNotFoundError("outbound_object", atURI) + } + return nil +} + func scanOutboundObject(row rowScanner) (*OutboundObject, error) { var object OutboundObject var tombstonedAt sql.NullTime diff --git a/internal/testutil/db.go b/internal/testutil/db.go index 74a9ddd..93cbe6e 100644 --- a/internal/testutil/db.go +++ b/internal/testutil/db.go @@ -88,8 +88,14 @@ func Truncate(t testing.TB, conn *sql.DB, tables ...string) { if len(tables) == 0 { return } + // CASCADE so truncating a table that is the target of a foreign key also + // clears its referencing tables (e.g. outbound_activities ← outbound_ + // deliveries): setup-time truncation wants a clean slate, and a caller that + // lists the parent but not the child would otherwise fail with a 0A000 FK + // error. Truncation runs before any seeding, so cascading never wipes data a + // test meant to keep. _, err := conn.ExecContext(context.Background(), - fmt.Sprintf(`TRUNCATE %s RESTART IDENTITY`, strings.Join(tables, ", "))) + fmt.Sprintf(`TRUNCATE %s RESTART IDENTITY CASCADE`, strings.Join(tables, ", "))) if err != nil { t.Fatalf("truncate test tables: %v", err) } -- 2.51.2