diff --git a/internal/consume/comments.go b/internal/consume/comments.go index dd9d61d..d3348a1 100644 --- a/internal/consume/comments.go +++ b/internal/consume/comments.go @@ -13,33 +13,49 @@ import ( // The native-comment path: a social.coves.community.comment record in a Coves // user's own repo that belongs to a thread the bridge federates. // -// Cycle F implements CREATE. Update, delete, the Lemmy depth cap and parents -// that are themselves comments are cycle H; they extend this file rather than -// replacing it, because the four steps below are the same for all of them. +// The DELETE case is what outbound_objects exists for. A Jetstream delete +// commit carries the repo DID, the collection and the rkey and NOTHING else: +// no record body, no CID, no reply refs. Every fact the Delete activity needs +// — the AP id, the community it is addressed to, the parent it hangs under — +// has to be read back out of state written at create time. + +// maxCommentDepth is Lemmy's comment nesting limit (0.19.20). A comment AT the +// cap still federates; one below it never can, so it is made visible in the +// DLQ rather than dropped at debug. +const maxCommentDepth = 50 // handleComment applies one comment commit. +func (d *Dispatcher) handleComment(ctx context.Context, did string, commit *CommitEvent) error { + switch commit.Operation { + case operationCreate, operationUpdate: + return d.applyCommentWrite(ctx, did, commit) + case operationDelete: + return d.applyCommentDelete(ctx, did, commit) + default: + d.logger.Debug("unknown comment operation", + slog.String("operation", commit.Operation), slog.String("did", did)) + return nil + } +} + +// applyCommentWrite handles a create or an update — the two operations that +// push content OUTWARD. // // The ORDER of the steps is the design, and each one is a gate on the next: // // 1. the opt-out gate, so a user who said no never gets an AP identity -// created for them; +// created for them and never has new content sent on their behalf; // 2. the thread resolution, so a comment in a NATIVE community — which this // bridge has no business federating — is skipped before anything is // written or minted; -// 3. the lazy mint, THROUGH handle verification, because the local part it +// 3. the depth cap, before any state or identity exists for a comment Lemmy +// will never accept; +// 4. the lazy mint, THROUGH handle verification, because the local part it // derives is frozen at creation; -// 4. the outbound state, then the intent — state first, because the intent is -// built from it and a delete one day has nothing else to be built from. -func (d *Dispatcher) handleComment(ctx context.Context, did string, commit *CommitEvent) error { - if commit.Operation != operationCreate { - // Update and delete are cycle H. Returning nil rather than an error - // keeps the cursor moving; the rev gate has already claimed this - // revision, which is what a later handler needs to stay ordered. - d.logger.Debug("comment operation not handled yet", - slog.String("operation", commit.Operation), slog.String("did", did)) - return nil - } - +// 5. the outbound state, then the intent — state first, because the intent's +// activity id comes from the seq the write bumps, and a delete one day has +// nothing else to be built from. +func (d *Dispatcher) applyCommentWrite(ctx context.Context, did string, commit *CommitEvent) error { federating, err := d.mayFederate(ctx, did) if err != nil { return err @@ -48,63 +64,111 @@ func (d *Dispatcher) handleComment(ctx context.Context, did string, commit *Comm // The residual split-thread case, explicitly chosen (decision 11): the // comment stays on the atproto side and the Lemmy side never sees it. d.logger.Debug("skipping comment from an opted-out author", - slog.String("did", did), slog.String("rkey", commit.RKey)) + slog.String("did", did), slog.String("rkey", commit.RKey), + slog.String("operation", commit.Operation)) return nil } - thread, err := d.resolveThread(ctx, commit) + atURI := commitRecordURI(did, commit) + thread, err := d.commentThread(ctx, atURI, commit) if err != nil { return err } if thread == nil { - // Most native comments live in native communities. Dead-lettering - // every one of them would bury the queue in events that are working - // exactly as intended. - d.logger.Debug("skipping comment with no federated parent", - slog.String("did", did), slog.String("rkey", commit.RKey)) + // Most native comments live in native communities, and most updates + // with no prior state are edits to a comment that never federated. + // Dead-lettering either would bury the queue in events working exactly + // as intended. + d.logger.Debug("skipping comment with no federated thread", + slog.String("did", did), slog.String("rkey", commit.RKey), + slog.String("operation", commit.Operation)) return nil } + if thread.Depth > maxCommentDepth { + // Named "depth" on purpose: the connector stores err.Error() as the + // dead letter's last_error, and that string is all an operator + // triaging the queue has to go on. + return fmt.Errorf("%w: comment %s is at depth %d, beyond Lemmy's cap of %d", + ErrPermanentEvent, atURI, thread.Depth, maxCommentDepth) + } + if err := d.ensureActor(ctx, did); err != nil { return err } - atURI := commitRecordURI(did, commit) snapshot, err := commentSnapshot(atURI, commit, thread) if err != nil { return err } - stored, err := d.objects.Upsert(ctx, store.OutboundObject{ ATURI: atURI, APObjectID: d.apObjectID(did, commit), LastCID: commit.CID, LastRev: commit.Rev, - // The community comes from the PARENT's mapping, never from anything - // the comment asserts about itself: a record can claim any community - // it likes, but its parent's mapping is what the bridge already - // federated. + // The community and the depth come from the THREAD, never from + // anything the record asserts about itself — and on an update they + // come from the state the create left, because an edit never moves a + // comment between communities or up the thread. CommunityDID: thread.CommunityDID, CommunityAPID: thread.CommunityAPID, TranslatedSnapshot: snapshot, - Depth: thread.Depth + 1, + Depth: thread.Depth, }) if err != nil { return fmt.Errorf("write outbound state for %s: %w", atURI, err) } + return d.enqueueComment(ctx, did, commit.Operation, stored, thread.ParentATURI, thread.ParentAPID) +} + +// applyCommentDelete withdraws a comment, using ONLY state. +// +// The opt-out gate is deliberately absent. A delete only ever REMOVES content, +// so it is always safe to send, and it is the only way a user who has opted +// out can retract what is already on the fediverse — blocking it would leave +// the peer's copy standing forever, the exact opposite of what asking to stop +// federating means. +func (d *Dispatcher) applyCommentDelete(ctx context.Context, did string, commit *CommitEvent) error { + atURI := commitRecordURI(did, commit) + + // Tombstone returns the state the Delete is built from in the same + // statement that stamps it, and is idempotent: a redelivered delete + // preserves the original tombstone time AND seq, so it reuses the activity + // id the first one sent and the peer recognises it as the same activity. + dead, err := d.objects.Tombstone(ctx, atURI) + if errors.IsNotFound(err) { + // A comment this bridge never federated. There is nothing to withdraw, + // and most native comment deletes are exactly this. + d.logger.Debug("skipping delete for a comment with no outbound state", + slog.String("did", did), slog.String("rkey", commit.RKey)) + return nil + } + if err != nil { + return fmt.Errorf("tombstone outbound state for %s: %w", atURI, err) + } + + parent := parentFromSnapshot(dead.TranslatedSnapshot) + return d.enqueueComment(ctx, 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 { intent := CommentIntent{ - Op: operationCreate, - ATURI: atURI, - ID: ActivityID(d.userOrigin, atURI, operationCreate, stored.LastActivitySeq), - CommunityAPID: thread.CommunityAPID, - ParentAPID: thread.ParentAPID, - Snapshot: snapshot, + Op: operation, + ATURI: stored.ATURI, + ID: ActivityID(d.userOrigin, stored.ATURI, operation, stored.LastActivitySeq), + CommunityAPID: stored.CommunityAPID, + ParentAPID: parentAPID, + Snapshot: stored.TranslatedSnapshot, } // parentATURI carries the causal dependency (decision 15): delivery must - // not present a reply to a peer before the thing it replies to. - if err := d.enqueuer.EnqueueActivity(ctx, did, did, thread.ParentATURI, intent); err != nil { - return fmt.Errorf("enqueue comment intent for %s: %w", atURI, err) + // 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 { + return fmt.Errorf("enqueue comment intent for %s: %w", stored.ATURI, err) } return nil } @@ -116,59 +180,110 @@ type resolvedThread struct { ParentAPID string CommunityDID string CommunityAPID string - // Depth is the PARENT's reply depth; the comment sits one below it. + // Depth is THIS comment's depth: the parent's recorded depth plus one. Depth int } -// resolveThread resolves the comment's parent through ap_objects and the -// parent's community through communities. A nil thread with a nil error means -// "not federated here" — a skip, not a failure. -func (d *Dispatcher) resolveThread(ctx context.Context, commit *CommitEvent) (*resolvedThread, error) { +// commentThread resolves the thread context for a create or an update. +// +// An UPDATE reads it back from the state its create wrote rather than +// re-resolving: the record could name a different parent, and an edit is not +// allowed to move a comment between communities or up the thread. It also +// means an edit to a comment that never federated (the author was opted out at +// the time, or it predates the bridge) resolves to nothing and is skipped. +func (d *Dispatcher) commentThread(ctx context.Context, atURI string, commit *CommitEvent) (*resolvedThread, error) { + if commit.Operation == operationUpdate { + stored, err := d.objects.GetByATURI(ctx, atURI) + if errors.IsNotFound(err) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("read outbound state for %s: %w", atURI, err) + } + parent := parentFromSnapshot(stored.TranslatedSnapshot) + return &resolvedThread{ + ParentATURI: parent.ATURI, + ParentAPID: parent.APID, + CommunityDID: stored.CommunityDID, + CommunityAPID: stored.CommunityAPID, + Depth: stored.Depth, + }, nil + } + return d.resolveParent(ctx, commit) +} + +// resolveParent finds the thing a new comment replies to. A parent can live in +// EITHER of two places, and both are legitimate: +// +// - ap_objects: content materialized from the fediverse (a Lemmy post or +// comment), or bridge-origin content mapped at write time; +// - outbound_objects: a NATIVE postv2 the acceptance engine admitted, or an +// earlier native comment. Nothing maps those into ap_objects — they were +// never materialized from the fediverse — so their outbound row is the +// only evidence they federate at all. +// +// A nil thread with a nil error means "not federated here": a skip, not a +// failure. +func (d *Dispatcher) resolveParent(ctx context.Context, commit *CommitEvent) (*resolvedThread, error) { parentATURI := replyRef(commit.Record, "parent") if parentATURI == "" { - // A comment with no reply.parent is a top-level comment shape this - // task does not federate; root-only replies fall back to the root. + // A root-only reply hangs directly under the thread root. parentATURI = replyRef(commit.Record, "root") } if parentATURI == "" { return nil, nil } - parent, err := d.objectMappings.GetByATURI(ctx, parentATURI) - if errors.IsNotFound(err) { - return nil, nil - } - if err != nil { + mapping, err := d.objectMappings.GetByATURI(ctx, parentATURI) + if err != nil && !errors.IsNotFound(err) { return nil, fmt.Errorf("resolve comment parent %s: %w", parentATURI, err) } - if parent.CommunityDID == "" { - return nil, nil + if err == nil && mapping.CommunityDID != "" { + community, err := d.communities.GetByDID(ctx, mapping.CommunityDID) + if errors.IsNotFound(err) { + // The parent is mapped but its community is not one this bridge + // federates, so there is nowhere to deliver to. + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("resolve community %s: %w", mapping.CommunityDID, err) + } + return &resolvedThread{ + ParentATURI: parentATURI, + ParentAPID: mapping.APID, + CommunityDID: mapping.CommunityDID, + CommunityAPID: community.APGroupID, + Depth: d.parentDepth(ctx, parentATURI) + 1, + }, nil } - community, err := d.communities.GetByDID(ctx, parent.CommunityDID) + state, err := d.objects.GetByATURI(ctx, parentATURI) if errors.IsNotFound(err) { - // The parent is mapped but its community is not one this bridge - // federates, so there is nowhere to deliver to. return nil, nil } if err != nil { - return nil, fmt.Errorf("resolve community %s: %w", parent.CommunityDID, err) + return nil, fmt.Errorf("read parent outbound state %s: %w", parentATURI, err) } - - thread := &resolvedThread{ + return &resolvedThread{ ParentATURI: parentATURI, - ParentAPID: parent.APID, - CommunityDID: parent.CommunityDID, - CommunityAPID: community.APGroupID, - } - // A parent that is itself a federated comment carries its own depth; a - // parent that is a post has none, and its replies are depth 1. - if parentState, err := d.objects.GetByATURI(ctx, parentATURI); err == nil { - thread.Depth = parentState.Depth - } else if !errors.IsNotFound(err) { - return nil, fmt.Errorf("read parent outbound state %s: %w", parentATURI, err) + ParentAPID: state.APObjectID, + CommunityDID: state.CommunityDID, + CommunityAPID: state.CommunityAPID, + Depth: state.Depth + 1, + }, nil +} + +// parentDepth reads a mapped parent's recorded depth, which exists only if the +// bridge federated it outward too. A parent with no outbound row is a post or +// a Lemmy object at the top of what this bridge tracks, so its replies are +// depth 1. Reading the parent's recorded depth is what keeps the cap O(1) +// instead of walking the thread on every comment. +func (d *Dispatcher) parentDepth(ctx context.Context, parentATURI string) int { + state, err := d.objects.GetByATURI(ctx, parentATURI) + if err != nil { + return 0 } - return thread, nil + return state.Depth } // ensureActor makes sure the author has an AP identity, resolving their handle @@ -210,9 +325,11 @@ func (d *Dispatcher) apObjectID(did string, commit *CommitEvent) string { } // commentSnapshot is the durable state a Delete or a restore is rebuilt from. -// It keeps the RECORD as it arrived plus the context that was resolved around -// it, because the delete commit that arrives one day carries neither. -// Rendering it into ActivityPub vocabulary is task 15's; this is the input. +// It keeps the RECORD as it arrived plus the context resolved around it, +// because the delete commit that arrives one day carries neither — including +// the PARENT, whose at-uri and AP id the Delete needs for causal ordering and +// addressing. Rendering all of it into ActivityPub vocabulary is task 15's; +// this is the input. func commentSnapshot(atURI string, commit *CommitEvent, thread *resolvedThread) ([]byte, error) { snapshot, err := json.Marshal(map[string]any{ "atUri": atURI, @@ -220,6 +337,7 @@ func commentSnapshot(atURI string, commit *CommitEvent, thread *resolvedThread) "rev": commit.Rev, "collection": commit.Collection, "record": commit.Record, + "parentAtUri": thread.ParentATURI, "parentApId": thread.ParentAPID, "communityApId": thread.CommunityAPID, }) @@ -231,6 +349,22 @@ func commentSnapshot(atURI string, commit *CommitEvent, thread *resolvedThread) return snapshot, nil } +// snapshotParent is the parent reference carried in a stored snapshot. +type snapshotParent struct { + ATURI string `json:"parentAtUri"` + APID string `json:"parentApId"` +} + +// parentFromSnapshot reads the parent back out of stored state. An unreadable +// or older snapshot yields empty strings rather than an error: the delete +// still has to go out, and delivery without the causal hint is better than a +// retraction that never leaves. +func parentFromSnapshot(snapshot []byte) snapshotParent { + var parent snapshotParent + _ = json.Unmarshal(snapshot, &parent) + return parent +} + // replyRef reads reply.{name}.uri out of a decoded comment record. func replyRef(record map[string]any, name string) string { reply, ok := record["reply"].(map[string]any) @@ -245,5 +379,8 @@ func replyRef(record map[string]any, name string) string { return uri } -// operationCreate is the commit operation that carries a new record. -const operationCreate = "create" +// The commit operations this consumer distinguishes. +const ( + operationCreate = "create" + operationUpdate = "update" +) diff --git a/internal/consume/comments_test.go b/internal/consume/comments_test.go new file mode 100644 index 0000000..babafdf --- /dev/null +++ b/internal/consume/comments_test.go @@ -0,0 +1,447 @@ +package consume + +import ( + "context" + "database/sql" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/store" +) + +// Task 14 cycle H: the rest of the comment path — update, delete, the Lemmy +// depth cap, and the second place a parent can live. +// +// The delete case is the one the whole outbound_objects table exists for. A +// Jetstream delete commit carries the repo DID, the collection and the rkey +// and NOTHING else: no record body, no CID, no reply refs. Every fact the +// Delete activity needs has to be read back out of state written at create +// time, and this file is where that is proven. + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +// commentFrameFull builds a comment commit of any operation. A delete carries +// no record at all, which is the point. +func commentFrameFull(did, rev, rkey, operation, content, parentATURI string) []byte { + if operation == "delete" { + return []byte(fmt.Sprintf( + `{"did":%q,"time_us":9100,"kind":"commit","commit":{"rev":%q,"operation":"delete",`+ + `"collection":"social.coves.community.comment","rkey":%q}}`, + did, rev, rkey)) + } + return []byte(fmt.Sprintf( + `{"did":%q,"time_us":9100,"kind":"commit","commit":{"rev":%q,"operation":%q,`+ + `"collection":"social.coves.community.comment","rkey":%q,`+ + `"cid":"bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4",`+ + `"record":{"$type":"social.coves.community.comment","reply":{`+ + `"root":{"uri":%q,"cid":%q},"parent":{"uri":%q,"cid":%q}},`+ + `"content":%q,"createdAt":"2026-08-13T10:00:00.000Z"}}}`, + did, rev, operation, rkey, acceptRootATURI, acceptRootCID, + parentATURI, acceptRootCID, content)) +} + +func commentATURIFor(did, rkey string) string { + return "at://" + did + "/" + CollectionComment + "/" + rkey +} + +// seedOutboundParent writes the outbound_objects row the ACCEPTANCE ENGINE (a +// native postv2 root) or an earlier native comment would have left behind. +// Nothing maps these into ap_objects — they were never materialized from the +// fediverse — so this row is the only record that they federate at all. +func seedOutboundParent(t *testing.T, database *sql.DB, atURI string, depth int) *store.OutboundObject { + t.Helper() + stored, err := store.NewOutboundObjects(database).Upsert(context.Background(), store.OutboundObject{ + ATURI: atURI, + APObjectID: acceptUserOrigin + "/ap/object/" + atURI, + CommunityDID: acceptCommunityDID, + CommunityAPID: acceptCommunityAPID, + TranslatedSnapshot: []byte(`{"seeded":"parent"}`), + Depth: depth, + }) + require.NoError(t, err, "seed outbound parent %s", atURI) + require.NotNil(t, stored) + return stored +} + +// createComment runs the create half so update/delete tests start from real +// state rather than a hand-written row. +func createComment(t *testing.T, fixture *dispatchFixture, rkey, content string) { + t.Helper() + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, rkey, "create", content, acceptRootATURI))) +} + +// --------------------------------------------------------------------------- +// H1 — update +// --------------------------------------------------------------------------- + +func TestCommentUpdate_BumpsSeqAndReplacesTheSnapshot(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + + const rkey = "3lzcmntupd001" + atURI := commentATURIFor(dispatchNativeDID, rkey) + + createComment(t, fixture, rkey, "first draft") + created, err := store.NewOutboundObjects(database).GetByATURI(context.Background(), atURI) + require.NoError(t, err) + require.Equal(t, 0, created.LastActivitySeq) + + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRevHigher, rkey, "update", "edited text", acceptRootATURI))) + + updated, err := store.NewOutboundObjects(database).GetByATURI(context.Background(), atURI) + require.NoError(t, err) + require.NotNil(t, updated) + + assert.Equal(t, 1, updated.LastActivitySeq, + "an applied update is a SECOND activity: reusing the create's seq would reuse "+ + "its id, and a peer that already has that id would drop the edit") + assert.Contains(t, string(updated.TranslatedSnapshot), "edited text", + "the snapshot is replaced, because it is what a later Delete is rebuilt from") + assert.NotContains(t, string(updated.TranslatedSnapshot), "first draft") + assert.Equal(t, dispatchRevHigher, updated.LastRev) + + assert.Equal(t, created.CommunityDID, updated.CommunityDID, + "an edit never moves a comment between communities") + assert.Equal(t, created.CommunityAPID, updated.CommunityAPID) + assert.Equal(t, created.Depth, updated.Depth, "nor changes where it sits in the thread") + assert.Nil(t, updated.TombstonedAt) + + calls := fixture.enqueuer.Calls() + require.Len(t, calls, 2, "one intent for the create, one for the update") + intent, ok := calls[1].Intent.(CommentIntent) + require.True(t, ok, "want CommentIntent, got %T", calls[1].Intent) + assert.Equal(t, "update", intent.Op) + assert.Equal(t, atURI, intent.ATURI) + assert.Equal(t, ActivityID(acceptUserOrigin, atURI, "update", 1), intent.ActivityID(), + "the id is derived from the op and the BUMPED seq") + assert.NotEqual(t, calls[0].Intent.ActivityID(), intent.ActivityID(), + "and is therefore distinct from the create's") +} + +func TestCommentUpdate_OfAnUnknownCommentIsSkipped(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + + // An edit whose create was never federated — the author opted out at the + // time, or the create predates the bridge. + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, "3lzcmntupd002", "update", "edited", acceptRootATURI)), + "an update with no prior state is not an error") +} + +// --------------------------------------------------------------------------- +// H2 — delete, rebuilt entirely from state +// --------------------------------------------------------------------------- + +func TestCommentDelete_IsBuiltFromStateBecauseTheFrameCarriesNothing(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + + const rkey = "3lzcmntdel001" + atURI := commentATURIFor(dispatchNativeDID, rkey) + + createComment(t, fixture, rkey, "goodbye cruel world") + require.Len(t, fixture.enqueuer.Calls(), 1) + + // The delete frame has no record, no CID, no reply refs. Everything the + // Delete activity needs must come out of outbound_objects. + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRevHigher, rkey, "delete", "", ""))) + + dead, err := store.NewOutboundObjects(database).GetByATURI(context.Background(), atURI) + require.NoError(t, err, "the row SURVIVES the delete — it is what a replayed create "+ + "is rejected against") + require.NotNil(t, dead) + require.NotNil(t, dead.TombstonedAt, "tombstoned_at is stamped") + assert.Equal(t, 1, dead.LastActivitySeq, "the Delete is the next activity") + + calls := fixture.enqueuer.Calls() + require.Len(t, calls, 2, "exactly one delete intent") + call := calls[1] + + assert.Equal(t, dispatchNativeDID, call.ActorDID) + assert.Equal(t, acceptRootATURI, call.ParentATURI, + "the parent at-uri comes from STATE: the delete frame does not carry reply refs, "+ + "so without it the causal ordering task 15 needs would be lost") + + intent, ok := call.Intent.(CommentIntent) + require.True(t, ok, "want CommentIntent, got %T", call.Intent) + assert.Equal(t, "delete", intent.Op) + assert.Equal(t, atURI, intent.ATURI) + assert.Equal(t, acceptCommunityAPID, intent.CommunityAPID, + "the community the Delete is addressed to comes from state") + assert.Equal(t, acceptRootAPID, intent.ParentAPID) + assert.Equal(t, ActivityID(acceptUserOrigin, atURI, "delete", 1), intent.ActivityID(), + "the id derives from the seq the tombstone bumped") + assert.Contains(t, string(intent.Snapshot), "goodbye cruel world", + "and the snapshot travels with it, because task 15 renders the Delete from the "+ + "object it is deleting") +} + +func TestCommentDelete_IsIdempotent(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + + const rkey = "3lzcmntdel002" + atURI := commentATURIFor(dispatchNativeDID, rkey) + + createComment(t, fixture, rkey, "bye") + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRevHigher, rkey, "delete", "", ""))) + + first, err := store.NewOutboundObjects(database).GetByATURI(context.Background(), atURI) + require.NoError(t, err) + require.NotNil(t, first) + require.NotNil(t, first.TombstonedAt, "the first delete must tombstone the row") + firstCalls := fixture.enqueuer.Calls() + require.Len(t, firstCalls, 2, "create then delete") + firstIntent := firstCalls[1].Intent.ActivityID() + + // A second delete with a HIGHER rev clears the gate and reaches the + // handler — a redelivery Jetstream is entitled to make. + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, "3lzrev0000003", rkey, "delete", "", ""))) + + second, err := store.NewOutboundObjects(database).GetByATURI(context.Background(), atURI) + require.NoError(t, err) + require.NotNil(t, second) + require.NotNil(t, second.TombstonedAt) + assert.Equal(t, *first.TombstonedAt, *second.TombstonedAt, + "the original tombstone time is preserved") + assert.Equal(t, first.LastActivitySeq, second.LastActivitySeq, + "and the seq does NOT bump again") + + lastCalls := fixture.enqueuer.Calls() + require.NotEmpty(t, lastCalls) + last := lastCalls[len(lastCalls)-1] + assert.Equal(t, firstIntent, last.Intent.ActivityID(), + "so a redelivered delete reuses the id the first one sent, and the peer "+ + "recognises it as the same activity instead of processing it twice") +} + +func TestCommentDelete_OfAnUnknownCommentIsSkipped(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + + // A comment this bridge never federated — a native thread, an opted-out + // author, or content older than the bridge. + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, "3lzcmntdel003", "delete", "", "")), + "a delete with no state is a skip: there is nothing to withdraw, and most "+ + "native comment deletes are exactly this") + + assert.Empty(t, fixture.enqueuer.Calls(), + "nothing may be enqueued for an object no peer was ever told about") +} + +// --------------------------------------------------------------------------- +// H3 — the Lemmy depth cap +// --------------------------------------------------------------------------- + +func TestCommentDepth_CountsFromTheParentsRecordedDepth(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + // A reply to a comment that is itself five deep. + parentATURI := commentATURIFor(acceptRootAuthorDID, "3lzcmntpar005") + seedOutboundParent(t, database, parentATURI, 5) + + const rkey = "3lzcmntdep001" + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, rkey, "create", "deep reply", parentATURI))) + + stored, err := store.NewOutboundObjects(database).GetByATURI( + context.Background(), commentATURIFor(dispatchNativeDID, rkey)) + require.NoError(t, err) + require.NotNil(t, stored) + assert.Equal(t, 6, stored.Depth, + "depth is the parent's plus one, read from the parent's own recorded depth "+ + "rather than by walking the thread on every comment") +} + +func TestCommentDepth_AtTheCapStillFederates(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + parentATURI := commentATURIFor(acceptRootAuthorDID, "3lzcmntpar049") + seedOutboundParent(t, database, parentATURI, 49) + + const rkey = "3lzcmntdep050" + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, rkey, "create", "at the limit", parentATURI)), + "depth 50 is the last one Lemmy accepts, so it must federate") + + stored, err := store.NewOutboundObjects(database).GetByATURI( + context.Background(), commentATURIFor(dispatchNativeDID, rkey)) + require.NoError(t, err) + require.NotNil(t, stored) + assert.Equal(t, 50, stored.Depth) + assert.Len(t, fixture.enqueuer.Calls(), 1) +} + +func TestCommentDepth_BeyondTheCapDeadLettersWithANamedReason(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + parentATURI := commentATURIFor(acceptRootAuthorDID, "3lzcmntpar050") + seedOutboundParent(t, database, parentATURI, 50) + + const rkey = "3lzcmntdep051" + err := fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, rkey, "create", "too deep", parentATURI)) + + require.Error(t, err, + "Lemmy caps comment depth at 50. A deeper comment must be VISIBLE — the DLQ — "+ + "rather than dropped at debug like an ordinary skip, because it is a real "+ + "comment a real user wrote that will never appear") + assert.Contains(t, err.Error(), "depth", + "the reason must name depth: the connector stores err.Error() as the dead "+ + "letter's last_error, and that string is the only thing an operator "+ + "triaging the queue has to go on") + assert.ErrorIs(t, err, ErrPermanentEvent, + "a comment cannot become shallower, so retrying it ten times only delays the "+ + "same answer") + + assert.Equal(t, 1, countRows(t, database, "outbound_objects"), + "and no state is written for it — the seeded parent stays the only row") + assert.Empty(t, fixture.enqueuer.Calls()) +} + +// --------------------------------------------------------------------------- +// H4 — the two places a parent can live +// --------------------------------------------------------------------------- + +func TestCommentParent_ResolvesFromOutboundStateWhenNotMapped(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + // A NATIVE postv2 root the acceptance engine admitted. It was never + // materialized from the fediverse, so it has no ap_objects mapping at all + // — its outbound_objects row is the only evidence it federates. + rootATURI := "at://" + acceptRootAuthorDID + "/" + CollectionPostV2 + "/3lznativert01" + parent := seedOutboundParent(t, database, rootATURI, 0) + require.Zero(t, countRows(t, database, "ap_objects"), + "the fixture deliberately maps nothing") + + const rkey = "3lzcmntsrc001" + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRev, rkey, "create", "reply to a native post", rootATURI))) + + stored, err := store.NewOutboundObjects(database).GetByATURI( + context.Background(), commentATURIFor(dispatchNativeDID, rkey)) + require.NoError(t, err, + "a reply to a native post must federate: those posts ARE the bridged content "+ + "the engine just accepted, and dropping their replies would leave every "+ + "native thread half-bridged") + require.NotNil(t, stored) + + assert.Equal(t, acceptCommunityDID, stored.CommunityDID, + "the community comes from the parent's outbound state, the same answer the "+ + "ap_objects path gives") + assert.Equal(t, acceptCommunityAPID, stored.CommunityAPID) + assert.Equal(t, 1, stored.Depth) + + calls := fixture.enqueuer.Calls() + require.Len(t, calls, 1) + assert.Equal(t, rootATURI, calls[0].ParentATURI) + intent, ok := calls[0].Intent.(CommentIntent) + require.True(t, ok) + assert.Equal(t, parent.APObjectID, intent.ParentAPID, + "and the parent's AP id comes from the row the engine wrote") +} + +// --------------------------------------------------------------------------- +// H5 — opting out does not trap content +// --------------------------------------------------------------------------- + +func TestCommentDelete_ProcessesEvenForAnOptedOutAuthor(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + ctx := context.Background() + + const rkey = "3lzcmntopt001" + atURI := commentATURIFor(dispatchNativeDID, rkey) + createComment(t, fixture, rkey, "already federated") + require.Len(t, fixture.enqueuer.Calls(), 1) + + // The author opts out AFTER the comment is already out on the fediverse. + _, err := store.NewFederationPrefs(database).Upsert(ctx, store.FederationPref{ + DID: dispatchNativeDID, + Source: store.FederationPrefSourceRecord, + }) + require.NoError(t, err) + + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRevHigher, rkey, "delete", "", ""))) + + dead, err := store.NewOutboundObjects(database).GetByATURI(ctx, atURI) + require.NoError(t, err) + require.NotNil(t, dead) + require.NotNil(t, dead.TombstonedAt) + + calls := fixture.enqueuer.Calls() + require.Len(t, calls, 2, + "the opt-out gate must NOT block a delete. Blocking it would leave the peer's "+ + "copy standing forever — the exact opposite of what a user asking to stop "+ + "federating means. A delete only ever REMOVES content, so it is always "+ + "safe, and it is the only way an opted-out user can retract what is "+ + "already out there") + intent, ok := calls[1].Intent.(CommentIntent) + require.True(t, ok) + assert.Equal(t, "delete", intent.Op) +} + +func TestCommentUpdate_IsBlockedForAnOptedOutAuthor(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + seedThreadRoot(t, database) + fixture := newDispatchFixture(t, database) + ctx := context.Background() + + const rkey = "3lzcmntopt002" + atURI := commentATURIFor(dispatchNativeDID, rkey) + createComment(t, fixture, rkey, "original") + + _, err := store.NewFederationPrefs(database).Upsert(ctx, store.FederationPref{ + DID: dispatchNativeDID, + Source: store.FederationPrefSourceRecord, + }) + require.NoError(t, err) + + require.NoError(t, fixture.handle(t, commentFrameFull( + dispatchNativeDID, dispatchRevHigher, rkey, "update", "new text", acceptRootATURI))) + + assert.Len(t, fixture.enqueuer.Calls(), 1, + "an UPDATE pushes new content outward, so the opt-out gate does block it — the "+ + "asymmetry with delete is the whole point: stop sending, but never trap "+ + "what is already sent") + + stored, err := store.NewOutboundObjects(database).GetByATURI(ctx, atURI) + require.NoError(t, err) + require.NotNil(t, stored) + assert.NotContains(t, string(stored.TranslatedSnapshot), "new text", + "and the state keeps the last version that actually federated") +} diff --git a/internal/consume/dispatch.go b/internal/consume/dispatch.go index 1c0fc85..c2f4cb3 100644 --- a/internal/consume/dispatch.go +++ b/internal/consume/dispatch.go @@ -302,16 +302,60 @@ func (d *Dispatcher) handleCommit(ctx context.Context, event *JetstreamEvent) er // handlePostV2 hands a native post to the task 16 acceptance engine. Admission, // the acceptance write and the outbound enqueue must ride ONE commit, which // only the engine can do — so this consumer hands over the whole commit and -// owns none of it, not even the outbound_objects row. +// owns none of it, not even the outbound_objects row. Keeping no post state +// here is also what makes the lexicon's community-immutability rule +// enforceable in ONE place: there is no second copy of the answer to disagree +// with the engine's. func (d *Dispatcher) handlePostV2(ctx context.Context, did string, commit *CommitEvent) error { if d.engine == nil { d.logger.Debug("no acceptance engine wired; skipping postv2", slog.String("did", did), slog.String("rkey", commit.RKey)) return nil } + + // A DELETE passes through ungated. It carries no record, so there is no + // community field to check — and gating on one it cannot see would drop + // every author delete and strand the acceptance records those deletes + // exist to take down. The engine already knows which posts it accepted + // and can no-op the rest. + if commit.Operation != operationDelete { + communityDID := stringField(commit.Record, "community") + if communityDID == "" { + // The lexicon REQUIRES community. A post without one is malformed + // and belongs in the DLQ, where a lexicon rollout mistake stays + // visible instead of becoming a silent drop. + return fmt.Errorf("%w: postv2 %s names no community", ErrPermanentEvent, commit.RKey) + } + bridged, err := d.isBridgedCommunity(ctx, communityDID) + if err != nil { + return err + } + if !bridged { + // Most Coves posts are exactly this. Admitting one would write an + // acceptance record into a community repo that has no business + // existing. + d.logger.Debug("skipping postv2 for a non-bridged community", + slog.String("did", did), slog.String("community", communityDID)) + return nil + } + } + return d.engine.AdmitPost(ctx, did, commit) } +// isBridgedCommunity reports whether Tidepool federates the community. The +// communities table is the authority: a row is what makes a community bridged. +func (d *Dispatcher) isBridgedCommunity(ctx context.Context, communityDID string) (bool, error) { + _, err := d.communities.GetByDID(ctx, communityDID) + if errors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("look up community %s: %w", communityDID, err) + } + return true, nil +} + // handleAccount applies a #account status change (decision 19). Status is what // distinguishes the transient states from deletion: active=false alone means // deactivated, suspended, takendown or throttled, none of which is a deletion, diff --git a/internal/consume/identity_fake_test.go b/internal/consume/identity_fake_test.go index 82a41a5..c0a33ee 100644 --- a/internal/consume/identity_fake_test.go +++ b/internal/consume/identity_fake_test.go @@ -3,6 +3,7 @@ package consume import ( "context" "fmt" + "net" "net/http" "net/http/httptest" "strings" @@ -35,9 +36,14 @@ type fakeIdentity struct { plcBody map[string]string // wellKnownStatus forces a status for one handle's endpoint. wellKnownStatus map[string]int + // txt is the DID each handle publishes over DNS, and txtRaw overrides the + // raw record set. Absent from both = NXDOMAIN. + txt map[string]string + txtRaw map[string][]string plcHits int wellKnownHits int + txtHits int } const wellKnownATProtoDIDPath = "/.well-known/atproto-did" @@ -51,6 +57,8 @@ func newFakeIdentity(t *testing.T) *fakeIdentity { plcStatus: map[string]int{}, plcBody: map[string]string{}, wellKnownStatus: map[string]int{}, + txt: map[string]string{}, + txtRaw: map[string][]string{}, } mux := http.NewServeMux() @@ -164,6 +172,46 @@ func (f *fakeIdentity) wellKnownFails(handle string, status int) { f.wellKnownStatus[strings.ToLower(handle)] = status } +// txtClaims publishes a handle's DID over DNS, the way a self-hosted handle +// does. This is the FIRST direction the resolver tries. +func (f *fakeIdentity) txtClaims(handle, did string) { + f.mu.Lock() + defer f.mu.Unlock() + f.txt[strings.ToLower(handle)] = did +} + +// txtRecords sets a handle's raw TXT record set — for the case where DNS +// answers but says nothing about atproto. +func (f *fakeIdentity) txtRecords(handle string, records ...string) { + f.mu.Lock() + defer f.mu.Unlock() + f.txtRaw[strings.ToLower(handle)] = records +} + +// lookupTXT is the LookupTXTFunc the resolver is wired with. It is injected +// rather than hitting the system resolver, which is what keeps these tests +// from issuing real DNS queries for the handles in their fixtures. +func (f *fakeIdentity) lookupTXT(_ context.Context, name string) ([]string, error) { + handle := strings.ToLower(strings.TrimPrefix(name, atprotoTXTPrefix)) + + f.mu.Lock() + defer f.mu.Unlock() + f.txtHits++ + if records, ok := f.txtRaw[handle]; ok { + return records, nil + } + if did, ok := f.txt[handle]; ok { + return []string{"did=" + did}, nil + } + return nil, &net.DNSError{Err: "no such host", Name: name, IsNotFound: true} +} + +func (f *fakeIdentity) TXTHits() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.txtHits +} + func (f *fakeIdentity) PLCHits() int { f.mu.Lock() defer f.mu.Unlock() @@ -188,6 +236,7 @@ func (f *fakeIdentity) resolver(t *testing.T) *HandleResolver { Transport: identityRewriteTransport{target: f.server.Listener.Addr().String()}, }, UserAgent: "tidepool-test/0.1", + LookupTXT: f.lookupTXT, }) require.NoError(t, err, "build handle resolver") require.NotNil(t, resolver) diff --git a/internal/consume/postv2_test.go b/internal/consume/postv2_test.go new file mode 100644 index 0000000..6a8106c --- /dev/null +++ b/internal/consume/postv2_test.go @@ -0,0 +1,191 @@ +package consume + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Task 14 cycle G: community.postv2 → the task 16 acceptance engine. +// +// This consumer owns none of a post. Admission, the acceptance record in the +// community's repo, and the outbound enqueue must all ride ONE commit, which +// only the engine can do — so the handler's entire job is deciding WHICH +// events reach the seam, and writing nothing itself. + +func postV2DeleteFrame(did, rev, rkey string) []byte { + return []byte(fmt.Sprintf( + `{"did":%q,"time_us":8100,"kind":"commit","commit":{"rev":%q,"operation":"delete",`+ + `"collection":"social.coves.community.postv2","rkey":%q}}`, + did, rev, rkey)) +} + +func postV2FrameNoCommunity(did, rev, rkey string) []byte { + return []byte(fmt.Sprintf( + `{"did":%q,"time_us":8200,"kind":"commit","commit":{"rev":%q,"operation":"create",`+ + `"collection":"social.coves.community.postv2","rkey":%q,`+ + `"cid":"bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4",`+ + `"record":{"$type":"social.coves.community.postv2","title":"orphan",`+ + `"createdAt":"2026-08-13T10:00:00.000Z"}}}`, + did, rev, rkey)) +} + +// --------------------------------------------------------------------------- +// G1 — gating on the target community +// --------------------------------------------------------------------------- + +// The happy path and the replay proof (same rev → the engine is invoked once, +// not twice) are pinned in +// TestDispatcher_RoutesPostV2ToTheAcceptanceEngine and +// TestDispatcher_ReplayedCommitReachesNoHandlerSideEffects/acceptance_engine. +// What follows is the gating those two do not cover. + +func TestPostV2_NonBridgedCommunityNeverReachesTheEngine(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) // a DIFFERENT community is bridged + fixture := newDispatchFixture(t, database) + + // A post to a community Tidepool does not federate. Most Coves posts are + // exactly this. + require.NoError(t, fixture.handle(t, postV2Frame(dispatchNativeDID, dispatchRev, + "3lzpostaaa111", "did:plc:someothercommunity000")), + "a post to a non-bridged community is a SKIP at debug, not a failure: it is the "+ + "common case, and dead-lettering it would bury the queue") + + assert.Zero(t, fixture.engine.Calls(), + "the acceptance engine must not be asked to admit a post into a community this "+ + "bridge does not federate — admission would write an acceptance record into "+ + "a repo that has no business existing") +} + +func TestPostV2_BridgedCommunityIsDecidedByTheCommunitiesTable(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + require.NoError(t, fixture.handle(t, + postV2Frame(dispatchNativeDID, dispatchRev, "3lzpostbbb222", acceptCommunityDID))) + + require.Equal(t, 1, fixture.engine.Calls(), + "a post whose community has a communities row reaches the engine") + assert.Equal(t, acceptCommunityDID, + fixture.engine.Commits()[0].Record["community"], + "the engine receives the community the record names, unaltered") +} + +func TestPostV2_MissingCommunityIsPermanent(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + err := fixture.handle(t, postV2FrameNoCommunity(dispatchNativeDID, dispatchRev, "3lzpostccc333")) + + require.Error(t, err, + "the lexicon REQUIRES community; a postv2 without one is malformed and belongs "+ + "in the DLQ where a lexicon rollout mistake stays visible") + assert.ErrorIs(t, err, ErrPermanentEvent, + "no amount of retrying puts a community field into a record that has none") + assert.Zero(t, fixture.engine.Calls()) +} + +func TestPostV2_NilEngineSkipsWithoutFailing(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + // A deployment where task 16 has not landed yet. + fixture := newDispatchFixture(t, database, func(opts *Options) { opts.Engine = nil }) + + require.NoError(t, fixture.handle(t, + postV2Frame(dispatchNativeDID, dispatchRev, "3lzpostddd444", acceptCommunityDID)), + "a missing engine must not panic and must not fail the event") + + assert.Zero(t, countRows(t, database, "outbound_objects")) + assert.Empty(t, fixture.enqueuer.Calls()) +} + +// --------------------------------------------------------------------------- +// G2 — the immutability fence +// --------------------------------------------------------------------------- + +// The lexicon makes a postv2's `community` immutable, and an update that +// changes it must be discarded WHOLE — consumers may retain neither half. +// +// Enforcement belongs to the ACCEPTANCE ENGINE (task 16), not here. The engine +// is the only party that knows a post's prior community: it wrote the +// acceptance record into that community's repo, so the previous value is its +// own state. This consumer has no memory of the create at all — it keeps no +// row for a postv2, by design (see the fence below). +// +// See the cycle G report for the one consumer-side memory that WILL exist once +// task 16 lands, and why it is worth revisiting then. +func TestPostV2_HandlerRetainsNoStateOfItsOwn(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + require.NoError(t, fixture.handle(t, + postV2Frame(dispatchNativeDID, dispatchRev, "3lzposteee555", acceptCommunityDID))) + require.Equal(t, 1, fixture.engine.Calls()) + + // An update that MOVES the post to another community — the hijack the + // immutability rule exists to stop. + require.NoError(t, fixture.handle(t, postV2Frame(dispatchNativeDID, dispatchRevHigher, + "3lzposteee555", acceptCommunityDID))) + + assert.Zero(t, countRows(t, database, "outbound_objects"), + "the consumer writes NO outbound state for a post. That is what makes the "+ + "immutability rule enforceable in one place: there is no consumer-side row "+ + "for a community change to corrupt, and no second copy of the answer for "+ + "the engine's to disagree with") + assert.Empty(t, fixture.enqueuer.Calls(), + "and it enqueues nothing: the post's outbound rides the engine's acceptance "+ + "commit, so an enqueue here would be a duplicate delivery") +} + +// --------------------------------------------------------------------------- +// G3 — deletes +// --------------------------------------------------------------------------- + +func TestPostV2_DeleteReachesTheEngine(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + require.NoError(t, fixture.handle(t, + postV2Frame(dispatchNativeDID, dispatchRev, "3lzpostfff666", acceptCommunityDID))) + require.Equal(t, 1, fixture.engine.Calls()) + + require.NoError(t, fixture.handle(t, + postV2DeleteFrame(dispatchNativeDID, dispatchRevHigher, "3lzpostfff666"))) + + require.Equal(t, 2, fixture.engine.Calls(), + "an author deleting their post must reach the engine too: the acceptance record "+ + "in the community repo has to come down with it, and only the engine can "+ + "take it down") + + deleted := fixture.engine.Commits()[1] + assert.Equal(t, "delete", deleted.Operation, + "the seam already carries the operation, so a delete needs no separate method") + assert.Equal(t, "3lzpostfff666", deleted.RKey) + assert.Nil(t, deleted.Record, + "a delete commit carries NO record body — the engine identifies the post by "+ + "repo + collection + rkey, exactly as this consumer does") +} + +func TestPostV2_DeleteIsNotGatedOnACommunityItCannotSee(t *testing.T) { + database := dispatchTestDB(t) + seedBridgedCommunity(t, database) + fixture := newDispatchFixture(t, database) + + // No create was ever seen, and the delete frame carries no record — so + // there is no community field to check against the communities table. + require.NoError(t, fixture.handle(t, + postV2DeleteFrame(dispatchNativeDID, dispatchRev, "3lzpostggg777"))) + + assert.Equal(t, 1, fixture.engine.Calls(), + "the bridged-community gate applies to records that HAVE a community field. A "+ + "delete has none, so gating it would silently drop every author delete and "+ + "strand the acceptance records they were supposed to remove; the engine "+ + "already knows which posts it accepted and can no-op the rest") +} diff --git a/internal/consume/resolver.go b/internal/consume/resolver.go index b9046a5..eb5f0ed 100644 --- a/internal/consume/resolver.go +++ b/internal/consume/resolver.go @@ -226,10 +226,22 @@ func handleFromAlsoKnownAs(alsoKnownAs []string) (string, error) { } return candidate, nil } - // RULED PERMANENT: the document is a complete answer, and it says this DID - // claims no handle. Nothing about retrying changes what the document says, - // so it is dead-lettered exhausted rather than redriven — the recovery - // path is manual, by design. + if len(alsoKnownAs) == 0 { + // TRANSIENT, deliberately. An empty alsoKnownAs is a complete answer + // today, but it is an answer about a mutable external world: a user + // who publishes their handle minutes after their first comment must + // still get that comment federated. Under this consumer's semantics a + // permanent failure is dead-lettered with its redrive budget already + // spent and never retried, so "permanent" here would mean "lost until + // a human intervenes", while transient costs ten cheap retries and + // reaches the same terminal state if the handle never appears. + return "", fmt.Errorf("DID document publishes no alsoKnownAs yet") + } + // PERMANENT. The DID has published identity claims and not one of them is + // an atproto handle — https:// profile links and mailto: addresses are + // perfectly valid alsoKnownAs entries and are simply not handles. Retrying + // re-reads the same list; the recovery path is the user publishing a + // handle, which arrives as a new event rather than a redrive of this one. return "", fmt.Errorf("%w: DID document claims no atproto handle", ErrPermanentEvent) } diff --git a/internal/consume/resolver_test.go b/internal/consume/resolver_test.go index d982757..fb3ab62 100644 --- a/internal/consume/resolver_test.go +++ b/internal/consume/resolver_test.go @@ -140,7 +140,7 @@ func TestHandleResolver_TransientFailuresStayRedrivable(t *testing.T) { } } -func TestHandleResolver_MissingHandleIsPermanent(t *testing.T) { +func TestHandleResolver_MissingHandleIsTransient(t *testing.T) { fake := newFakeIdentity(t) // A valid DID document with no alsoKnownAs at all. fake.plcServes(resolveDID, `{"@context":["https://www.w3.org/ns/did/v1"], @@ -150,13 +150,19 @@ func TestHandleResolver_MissingHandleIsPermanent(t *testing.T) { require.Error(t, err, "a DID with no handle cannot be given a frozen local part") assert.Empty(t, handle) - assert.ErrorIs(t, err, ErrPermanentEvent, - "RULED PERMANENT: the document states the DID has no handle, which is an answer "+ - "rather than a failure. See the cycle F report — permanent events are "+ - "excluded from redrive by design, so the recovery path is manual") + assert.NotErrorIs(t, err, ErrPermanentEvent, + "RE-RULED TRANSIENT. The document is a complete answer TODAY, but it is an "+ + "answer about a mutable external world: a user who publishes their handle "+ + "minutes after their first comment must get that comment federated. Under "+ + "our shipped semantics a permanent failure is dead-lettered with its "+ + "budget already spent and the redriver never touches it again, so "+ + "'permanent' here would mean 'lost until a human intervenes' — while "+ + "transient costs ten cheap retries and reaches the same terminal state "+ + "if the handle never appears") assert.Zero(t, fake.WellKnownHits(), "and nothing is fetched from the network on a claim that does not exist") + assert.Zero(t, fake.TXTHits(), "nor from DNS") } func TestHandleResolver_NonATProtoAlsoKnownAsIsPermanent(t *testing.T) { @@ -214,3 +220,86 @@ func TestNewHandleResolver_RequiresGuardedEgress(t *testing.T) { require.Error(t, err, "the directory URL must be an absolute http(s) URL") assert.True(t, errors.IsValidation(err), "want validation error, got %v", err) } + +// --------------------------------------------------------------------------- +// FF2 — the DNS half of handle verification +// --------------------------------------------------------------------------- +// +// The atproto convention offers two ways for a handle to claim a DID back: +// a _atproto.{handle} TXT record, and the HTTPS well-known. DNS is tried +// first, which is both the spec's order and the safer one — see the +// contradiction test below. + +func TestHandleResolver_DNSAloneVerifiesAndSkipsTheWellKnown(t *testing.T) { + fake := newFakeIdentity(t) + // The document claims the handle; the handle claims the DID back over DNS + // ONLY — it serves no well-known at all, like most self-hosted handles. + fake.claimOneWay(resolveDID, resolveHandle) + fake.txtClaims(resolveHandle, resolveDID) + + handle, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveDID) + require.NoError(t, err, + "a TXT-verified handle must resolve: DNS-only handles are the majority on the "+ + "network, and refusing them would silently drop those users' comments") + assert.Equal(t, resolveHandle, handle) + + assert.Equal(t, 1, fake.TXTHits()) + assert.Zero(t, fake.WellKnownHits(), + "DNS answered, so the well-known is never fetched — one round-trip, not two, "+ + "on the path in front of every first-time commenter") +} + +func TestHandleResolver_ContradictingTXTIsPermanentAndTheWellKnownCannotOverrideIt(t *testing.T) { + fake := newFakeIdentity(t) + // Mallory's document claims alice's handle. + fake.claimOneWay(resolveOtherDID, resolveHandle) + // DNS — which alice controls — says the handle is alice's. + fake.txtClaims(resolveHandle, resolveDID) + // The well-known says otherwise. An attacker who can serve HTTP for the + // handle's host, but cannot change its DNS, would win if a contradicting + // TXT fell through to the well-known. + fake.wellKnownReturns(resolveHandle, resolveOtherDID) + + handle, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveOtherDID) + + require.Error(t, err) + assert.Empty(t, handle) + assert.ErrorIs(t, err, ErrPermanentEvent, + "DNS answered and the answer names a different DID: that is a refusal, not a "+ + "missing record") + assert.Zero(t, fake.WellKnownHits(), + "and the well-known must NOT be consulted after a contradicting TXT — falling "+ + "through would let whoever controls the handle's web server overrule the "+ + "DNS its real owner published") +} + +func TestHandleResolver_MissingTXTFallsThroughToTheWellKnown(t *testing.T) { + fake := newFakeIdentity(t) + fake.claim(resolveDID, resolveHandle) // well-known only; no TXT registered + + handle, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveDID) + require.NoError(t, err) + assert.Equal(t, resolveHandle, handle) + + assert.Equal(t, 1, fake.TXTHits(), "DNS is asked first") + assert.Equal(t, 1, fake.WellKnownHits(), + "NXDOMAIN is 'this handle does not use DNS', not 'this handle disowns the DID', "+ + "so the well-known is the answer") +} + +func TestHandleResolver_TXTWithoutAnATProtoRecordFallsThrough(t *testing.T) { + fake := newFakeIdentity(t) + fake.claim(resolveDID, resolveHandle) + // DNS answers, but says nothing about atproto. Almost every domain has + // TXT records; only the did= one is a claim. + fake.txtRecords(resolveHandle, "v=spf1 -all", "google-site-verification=abc123") + + handle, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveDID) + require.NoError(t, err) + assert.Equal(t, resolveHandle, handle) + + assert.Equal(t, 1, fake.TXTHits()) + assert.Equal(t, 1, fake.WellKnownHits(), + "an unrelated TXT record set is not an atproto answer, so verification "+ + "continues rather than failing") +}