From 45f1afd753a64456a25fb0d078b097fbb4d48c7c Mon Sep 17 00:00:00 2001 From: Bretton Date: Fri, 14 Aug 2026 12:31:23 -0700 Subject: [PATCH] feat(moderation): thread locks and native comment removal (17c-2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The moderation state the bridge owns, backed by one table. LOCK. An announced Lock from the community that owns a post's mapping is recorded in object_moderation; a native comment in that thread is REFUSED. The refusal is permanent (ErrPermanentEvent carrying parent-locked) so the event dead-letters and the cursor advances — a transient error would park every other native user behind one locked thread, retrying a decision only a moderator can change. It runs after the depth cap and before the actor mint, so nothing is created for a comment that will never federate, and it leaves the rev gate un-advanced so the byte-identical retry after an Undo{Lock} is admitted rather than swallowed as a stale replay. The lock is per-THREAD, not per-parent. Checking only the direct parent stops nothing, because replying under an existing comment is how conversations continue. The thread root is carried on resolvedThread and on the stored snapshot (the edit path rebuilds from state), and both parent and root are asked in one statement. Roots for fediverse content come from a new ap_objects.thread_root_at_uri (migration 026), recorded where materialize.resolveReplyRefs already resolves reply.root at commit time — zero extra reads on the consumer's most common write, versus a repo read (a REPEATABLE READ tx, head read and MST walk) per comment. It belongs on ap_objects, unlike the lock, precisely because the materializer re-supplies it on every re-put and a repin therefore cannot lose it. This also closes structurally, rather than by policy, the risk that an unresolvable root parks a whole community: an empty root now requires a dead-ended walk, which requires depth > 0, which requires an outbound row, which only native content has. COMMENT REMOVAL. An announced Delete WITH a summary records removed_at, code and reason bound to the announcing community; Undo clears them. Nothing Coves-visible is written — the removal lexicon is post-scoped and the comment-subject extension is Coves-owned and unlanded — and the test asserting that says inline that it is the assertion which must change when the extension lands, so the gap is a contract with an expiry rather than an absence nobody checks. Post removal state stays in the community repo record: duplicating it would make two sources of truth for what acceptrec already makes atomic. A summary-LESS announced delete records nothing. Lemmy spells a moderator removal with a summary and a self-delete without one, so writing moderator-discretion for it would assert a moderator acted when none did. It is still TAKEN rather than declined, because the v1 path it would otherwise fall into soft-deletes our own mapping and tombstones our own AP id, leaving the object permanently unmoderatable. The two shapes now fork with distinct reasons and counters: previously one reason and one counter meant an operator asking "did a moderator remove this?" read an answer that could not distinguish yes from no. Every cycle bite-proven, including the ruling-C scope control (routing a post removal through the comment writer) and the ownership conjunct. Co-Authored-By: Claude Opus 5 (1M context) --- FOLLOWUPS.md | 25 + internal/ap/vocab.go | 12 + internal/consume/comments.go | 157 +++ internal/consume/subjects.go | 125 +- .../db/migrations/025_object_moderation.sql | 53 + .../migrations/026_ap_object_thread_root.sql | 38 + internal/ingest/consent.go | 72 +- internal/ingest/handler.go | 10 +- internal/ingest/ingest_test.go | 14 +- internal/ingest/moderation.go | 179 +++ internal/ingest/moderation_lock_test.go | 1166 +++++++++++++++++ internal/ingest/moderation_terminal_test.go | 43 +- internal/materialize/acceptance.go | 13 +- internal/materialize/community.go | 24 + internal/materialize/materializer.go | 12 + internal/store/ap_objects.go | 21 +- internal/store/interfaces.go | 56 + internal/store/models.go | 36 +- internal/store/object_moderation.go | 179 +++ 19 files changed, 2175 insertions(+), 60 deletions(-) create mode 100644 internal/db/migrations/025_object_moderation.sql create mode 100644 internal/db/migrations/026_ap_object_thread_root.sql create mode 100644 internal/ingest/moderation.go create mode 100644 internal/ingest/moderation_lock_test.go create mode 100644 internal/store/object_moderation.go diff --git a/FOLLOWUPS.md b/FOLLOWUPS.md index 218909a..4b7765c 100644 --- a/FOLLOWUPS.md +++ b/FOLLOWUPS.md @@ -119,6 +119,31 @@ task documents and git history rather than this list. event. The decision-19 reconciliation job (task 18) is the natural home for a periodic acceptance-vs-record pin audit. +## Thread locks (task 17c-2) + +- **Comments materialized before migration 026 carry no + `thread_root_at_uri`**, so they resolve to themselves and a lock on the + post above them will not refuse native replies hanging under them. It + degrades to the pre-lock behaviour rather than to the community-wide + park, and it heals the moment each comment is next materialized — an + admin backfill of the affected communities is the deliberate fix. A + record-read fallback (reading `reply.root` back through `RecordGetter` + when the column is empty) would close it for the existing corpus, but + the consumer holds no repo manager and wiring one to read a fact the + materializer already computes is the wrong trade on the hot path. +- **The lock read is per-object, keyed by at-uri.** A lock recorded on a + Lemmy post reaches native replies anywhere in that thread; it does NOT + reach replies whose chain the bridge never materialized. That is the + boundary of what the bridge can know, not a policy choice. +- **`walkThreadRoot`'s dead-end branch remains reachable for native + content** whose outbound state predates the root column (legacy rows + only, and self-healing on the next successful edit). When it fires and + the community holds a standing lock, the event is refused retryably and + the error names the at-uri the chain stopped at. If dead ends ever + become common, that branch parks traffic — the two fediverse controls in + `moderation_lock_test.go` go red the moment it becomes reachable from + fediverse resolution, which is what keeps it honest. + ## Vote accounting (task 17b) - **The re-seed baseline clamp is a FLOOR BREACH, not a discard diff --git a/internal/ap/vocab.go b/internal/ap/vocab.go index 0d50022..db87b8a 100644 --- a/internal/ap/vocab.go +++ b/internal/ap/vocab.go @@ -51,6 +51,18 @@ const ( TypeLike = "Like" TypeDislike = "Dislike" + // TypeLock is Lemmy's thread lock (activities/community/lock_page.rs), + // announced by the community as Announce{Lock} and lifted with + // Announce{Undo{Lock}}. + // + // Its `object` is the TARGET — the post being locked — never a payload the + // activity carries. That is why it must stay OUT of echo.carriesPayload: + // the target of every inbound moderation action against native content is, + // by definition, one of OUR ids, so descending into it would classify each + // one as our own echo and drop inbound moderation entirely while the drop + // counter reported it working. + TypeLock = "Lock" + TypeTombstone = "Tombstone" TypeImage = "Image" TypeLink = "Link" diff --git a/internal/consume/comments.go b/internal/consume/comments.go index a01f836..4c37cc3 100644 --- a/internal/consume/comments.go +++ b/internal/consume/comments.go @@ -98,6 +98,10 @@ func (d *Dispatcher) applyCommentWrite(ctx context.Context, tx *sql.Tx, did stri ErrPermanentEvent, atURI, thread.Depth, maxCommentDepth) } + if err := d.refuseInLockedThread(ctx, atURI, thread); err != nil { + return err + } + if err := d.ensureActor(ctx, did); err != nil { return err } @@ -127,6 +131,84 @@ func (d *Dispatcher) applyCommentWrite(ctx context.Context, tx *sql.Tx, did stri return d.enqueueComment(ctx, tx, did, commit.Operation, stored, thread.ParentATURI, thread.ParentAPID) } +// refuseInLockedThread refuses a comment in a thread a community has LOCKED +// (task 17c-2). It is the read half of the lock: recording one and then +// federating a reply under it is worse than not recording it at all, because +// Lemmy rejects comments on locked posts server-side — the reply buys a failed +// delivery, a retry loop and finally a poisoned row whose cause is a moderator +// decision three tables away. +// +// It asks about the PARENT AND THE THREAD ROOT, because Lemmy locks threads and +// a check on the parent alone stops only direct replies: anyone can keep talking +// by hitting reply one level down, which is not an edge case but the ordinary +// shape of a conversation. Both are asked in ONE statement, and the scope stays +// per-object — a lock on one post says nothing about the community's other +// threads. +// +// The refusal is PERMANENT, and that is the whole design: +// +// - it DEAD-LETTERS and the cursor moves on. A transient error would block +// every other native user's traffic behind one locked thread, retrying a +// decision only a moderator can change. +// - it carries the REASON, "parent-locked" — the same code the admissions +// ledger already spells for a post refused under a locked parent, so an +// operator meets one vocabulary rather than two. The connector stores +// err.Error() as the dead letter's last_error, and that string is the only +// surface anyone triaging the queue — or answering the author asking where +// their comment went — has to go on. +// - it is a GATE, not a verdict on the record. It runs before the mint and +// before any outbound state, and it leaves the rev gate un-advanced (the +// gate transaction rolls back on any error), so the IDENTICAL comment — +// same rkey, same rev, same cid — is admitted once the lock is lifted. A +// refusal that advanced the gate would swallow that retry as a stale replay +// and lose the comment for good. +// +// A store failure PROPAGATES as an ordinary (retryable) error: "we could not +// read the lock" must never be answered with "there is no lock", which is +// exactly the reply the lock exists to stop. +func (d *Dispatcher) refuseInLockedThread(ctx context.Context, atURI string, thread *resolvedThread) error { + locked, err := d.objectMappings.LockedAmong(ctx, thread.ParentATURI, thread.RootATURI) + if err != nil { + return fmt.Errorf("read lock state of the thread above %s: %w", atURI, err) + } + if locked != "" { + return fmt.Errorf("%w: parent-locked: comment %s is in a thread its community locked (%s)", + ErrPermanentEvent, atURI, locked) + } + if thread.RootATURI != "" { + return nil + } + + // The thread could not be established (walkThreadRoot dead-ended on state + // this consumer never wrote — every comment snapshot it has ever written + // names its parent, and a post is at depth 0). "We do not know which thread + // this is" must not be answered with "the thread is not locked" — that is + // the same fail-open the empty root exists to prevent — but neither may it + // strand replies forever under content nobody has moderated. So the question + // narrows to the only one that can still matter: does this community hold + // ANY lock the unknown root might be? If it holds none, there is provably + // nothing to miss. If it holds one, we cannot tell, and a retryable failure + // is the honest answer — the operator sees it, and a lifted lock resolves it. + // + // This is NOT a community-scoped refusal, and it is not reachable from + // fediverse content: a mapped subject's thread is answered from state the + // materializer recorded, so it resolves or it is a thread root, never an + // empty answer. A read that could FAIL here would turn one locked post into + // a community-wide park of ordinary replies, which is why there is no read + // on that path at all. + held, err := d.objectMappings.CommunityHoldsAnyLock(ctx, thread.CommunityDID) + if err != nil { + return fmt.Errorf("read standing locks of %s: %w", thread.CommunityDID, err) + } + if !held { + return nil + } + return fmt.Errorf( + "cannot tell whether comment %s is in a locked thread: the chain above it cannot be followed past %s, "+ + "whose outbound state names no parent, and %s holds standing locks", + atURI, thread.RootDeadEnd, thread.CommunityDID) +} + // applyCommentDelete withdraws a comment, using ONLY state. // // The opt-out gate is deliberately absent. A delete only ever REMOVES content, @@ -194,6 +276,22 @@ type resolvedThread struct { CommunityAPID string // Depth is THIS comment's depth: the parent's recorded depth plus one. Depth int + // RootATURI is the at-uri of the thread this comment hangs in — the post at + // the top of it, whoever the immediate parent is. A community locks a + // THREAD, so this is what the lock check asks about; the parent alone would + // stop only direct replies, and the reply button under every existing + // comment is the normal way a conversation continues. + // + // It is derived from the bridge's own state (the parent's recorded root, or + // the parent itself when the parent IS a root), never from the record's + // reply.root, which the author writes and could point anywhere. + RootATURI string + // RootDeadEnd names the object the thread resolution stopped at when + // RootATURI could not be established. It is DIAGNOSTIC ONLY — never written + // to the snapshot, never inherited — and exists so the one error that holds + // a comment for an undeterminable thread names something an operator can + // open, instead of a thread nobody can look up. + RootDeadEnd string } // commentThread resolves the thread context for a create or an update. @@ -203,6 +301,13 @@ type resolvedThread struct { // 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. +// +// The thread ROOT is read back the same way, and RESOLVED when the stored +// snapshot predates it (walkThreadRoot). It cannot be defaulted away: an update +// never re-resolves its thread, so a root the create path knows and the update +// path does not is a lock the create is refused by and the edit sails through — +// leaving the author of an existing reply a live, editable surface inside a +// closed thread. func (d *Dispatcher) commentThread(ctx context.Context, atURI string, commit *CommitEvent) (*resolvedThread, error) { if commit.Operation == operationUpdate { stored, err := d.objects.GetByATURI(ctx, atURI) @@ -213,12 +318,23 @@ func (d *Dispatcher) commentThread(ctx context.Context, atURI string, commit *Co return nil, fmt.Errorf("read outbound state for %s: %w", atURI, err) } parent := d.parentFromSnapshot(stored.TranslatedSnapshot) + rootATURI, deadEnd := rootFromSnapshot(stored.TranslatedSnapshot), "" + if rootATURI == "" { + // Written before the root was recorded: climb this comment's own + // parent chain rather than assuming anything. The successful edit + // re-writes the snapshot below, so the walk happens once per row. + if rootATURI, deadEnd, err = d.walkThreadRoot(ctx, atURI); err != nil { + return nil, err + } + } return &resolvedThread{ ParentATURI: parent.ATURI, ParentAPID: parent.APID, CommunityDID: stored.CommunityDID, CommunityAPID: stored.CommunityAPID, Depth: stored.Depth, + RootATURI: rootATURI, + RootDeadEnd: deadEnd, }, nil } return d.resolveParent(ctx, commit) @@ -241,12 +357,22 @@ func (d *Dispatcher) resolveParent(ctx context.Context, commit *CommitEvent) (*r if err != nil || parent == nil { return nil, err } + // The thread is the parent's thread — its recorded root, or the parent + // itself when the parent is a root. Same shape as the depth above: inherited + // from the parent's own state, which is what keeps both O(1) instead of + // walking the thread on every comment. + rootATURI, deadEnd, err := d.threadRootOf(ctx, parent) + if err != nil { + return nil, err + } return &resolvedThread{ ParentATURI: parent.ATURI, ParentAPID: parent.APID, CommunityDID: parent.CommunityDID, CommunityAPID: parent.CommunityAPID, Depth: parent.Depth + 1, + RootATURI: rootATURI, + RootDeadEnd: deadEnd, }, nil } @@ -294,6 +420,12 @@ func (d *Dispatcher) apObjectID(did string, commit *CommitEvent) string { // 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. +// +// The thread ROOT is here for a different consumer: the update path, which +// never re-resolves its thread and would otherwise have no way to know which +// conversation an edit belongs to. It is also what every LATER comment inherits +// its own root from, so recording it once at create time is what keeps the +// answer O(1) forever after. func commentSnapshot(atURI string, commit *CommitEvent, thread *resolvedThread) ([]byte, error) { snapshot, err := json.Marshal(map[string]any{ "atUri": atURI, @@ -303,6 +435,7 @@ func commentSnapshot(atURI string, commit *CommitEvent, thread *resolvedThread) "record": commit.Record, "parentAtUri": thread.ParentATURI, "parentApId": thread.ParentAPID, + "rootAtUri": thread.RootATURI, "communityApId": thread.CommunityAPID, }) if err != nil { @@ -335,6 +468,30 @@ func (d *Dispatcher) parentFromSnapshot(snapshot []byte) snapshotParent { return parent } +// rootFromSnapshot reads the thread root back out of stored state, or "" when +// the snapshot does not carry one (a row written before the root was recorded, +// or one that will not parse). +// +// The empty answer is NOT "no thread": every caller resolves it (threadRootOf, +// commentThread), because defaulting it to the object itself would quietly +// exempt every pre-existing nested comment from its thread's lock — and that +// does not self-heal, since the snapshot only advances on a successful edit and +// the edit is what would be wrongly allowed. +// +// A parse failure is deliberately indistinguishable from an absent key here: +// both mean "this snapshot cannot tell us", parentFromSnapshot already logs the +// unmarshal error for the same bytes, and the resolution the caller falls into +// is the correct response to either. +func rootFromSnapshot(snapshot []byte) string { + var thread struct { + RootATURI string `json:"rootAtUri"` + } + if err := json.Unmarshal(snapshot, &thread); err != nil { + return "" + } + return thread.RootATURI +} + // 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) diff --git a/internal/consume/subjects.go b/internal/consume/subjects.go index 175948d..92d91dc 100644 --- a/internal/consume/subjects.go +++ b/internal/consume/subjects.go @@ -3,6 +3,7 @@ package consume import ( "context" "fmt" + "log/slog" "tidepool/internal/errors" "tidepool/internal/materialize" @@ -31,6 +32,16 @@ type resolvedSubject struct { // subject the bridge does not track depth for). Callers that nest below it // add one. Depth int + // RootATURI is the at-uri of the THREAD this subject hangs in, as the + // bridge's own outbound state records it. It is empty for a subject that IS + // a thread root (Depth 0) and for a comment written before the root was + // recorded — threadRootOf tells those two apart, because they are opposite + // answers, not one missing value. + // + // It comes from state, never from the record: reply.root is written by the + // author, so trusting it would let anyone reopen a locked thread by naming a + // different root. + RootATURI string } // resolveSubject looks a subject up in the two places one can live, in order. @@ -69,7 +80,7 @@ func (d *Dispatcher) resolveSubject(ctx context.Context, atURI string) (*resolve if err != nil { return nil, fmt.Errorf("resolve community %s: %w", communityDID, err) } - depth, tombstoned, err := d.recordedState(ctx, atURI) + depth, rootATURI, tombstoned, err := d.recordedState(ctx, atURI) if err != nil { return nil, err } @@ -83,12 +94,23 @@ func (d *Dispatcher) resolveSubject(ctx context.Context, atURI string) (*resolve // author-DELETED native post would start federating again. return nil, nil } + if rootATURI == "" { + // No outbound row said which thread this is, so it is not one + // the bridge federated OUTWARD — it is fediverse content the + // bridge materialized IN, and the materializer recorded the + // thread on the mapping (migration 026). Without this a native + // reply to a LEMMY COMMENT resolves its thread to that comment, + // and the lock on the post above it never reaches the reply — + // which is most of Lemmy, since most replies go under comments. + rootATURI = mapping.ThreadRootATURI + } return &resolvedSubject{ ATURI: atURI, APID: mapping.APID, CommunityDID: communityDID, CommunityAPID: community.APGroupID, Depth: depth, + RootATURI: rootATURI, }, nil } } @@ -111,9 +133,96 @@ func (d *Dispatcher) resolveSubject(ctx context.Context, atURI string) (*resolve CommunityDID: state.CommunityDID, CommunityAPID: state.CommunityAPID, Depth: state.Depth, + RootATURI: rootFromSnapshot(state.TranslatedSnapshot), }, nil } +// threadRootOf answers which THREAD a new child of subject belongs to — the +// at-uri a lock on the whole conversation is recorded against. +// +// Three answers, and the order matters: +// +// 1. a recorded root is the answer, and the cheapest one; +// 2. a subject at depth 0 IS a thread root — a post, or an object the bridge +// tracks no thread state for, which is the same boundary recordedState +// already draws for depth; +// 3. anything else is a comment whose stored state predates the recorded root, +// and it must be RESOLVED rather than assumed. Answering "itself" there +// would silently make every pre-existing nested comment its own thread, so +// a lock on the real root would not reach it — the bypass surviving the fix +// for exactly the comments most likely to be in an old thread. +// +// The second result names the object the answer stopped at when the root could +// not be established — the only thing an operator can look up when a comment is +// held for an undeterminable thread. +func (d *Dispatcher) threadRootOf(ctx context.Context, subject *resolvedSubject) (root, deadEnd string, err error) { + if subject.RootATURI != "" { + return subject.RootATURI, "", nil + } + if subject.Depth == 0 { + return subject.ATURI, "", nil + } + return d.walkThreadRoot(ctx, subject.ATURI) +} + +// walkThreadRoot climbs the recorded parent chain of a comment whose stored +// state does not name its thread root, and returns the root it reaches. +// +// Cold by construction: every comment written since the root was recorded +// carries it, so this runs only for rows that predate it — and the update path +// re-writes the snapshot on the next successful edit, so a row that walks once +// stops walking. +// +// Reaching content the bridge holds no outbound state for is NOT a failure: that +// is the edge of what this bridge federated, and the last object on the chain is +// the top of the thread as far as any of our state goes — the same boundary +// recordedState draws for depth, and the same answer a direct reply to that +// object would have got. +// +// A chain that DEAD-ENDS — a nested row naming no parent, or one longer than a +// conversation can be — returns an empty root and NAMES the object it stopped +// at. Three things it deliberately does not do. It does not guess the deepest +// object reached, because that answer would be written into the child's snapshot +// and every later comment would inherit the guess. It does not decide the +// outcome, because "we could not determine the thread" is not "the thread is not +// locked" — refuseInLockedThread weighs the empty answer against the locks that +// actually exist. And it does not report the dead end as a bare failure: the +// object it names is the one an operator has to open to see why. A store failure +// IS an error: retrying is the only honest response to a question that was never +// answered. +func (d *Dispatcher) walkThreadRoot(ctx context.Context, atURI string) (root, deadEnd string, err error) { + climbed := atURI + // Bounded by Lemmy's nesting cap plus the root itself: every hop is one + // level up, so a chain longer than that is a cycle, not a conversation. + for hop := 0; hop <= maxCommentDepth+1; hop++ { + state, err := d.objects.GetByATURI(ctx, climbed) + if errors.IsNotFound(err) { + return climbed, "", nil + } + if err != nil { + return "", "", fmt.Errorf("read thread state for %s: %w", climbed, err) + } + if root := rootFromSnapshot(state.TranslatedSnapshot); root != "" { + return root, "", nil + } + if state.Depth == 0 { + return climbed, "", nil + } + parent := d.parentFromSnapshot(state.TranslatedSnapshot) + if parent.ATURI == "" { + // State this consumer never wrote: every comment snapshot it has + // ever written names its parent, and a post is at depth 0. + d.logger.Warn("thread root is undeterminable: nested outbound state names no parent", + slog.String("at_uri", atURI), slog.String("dead_end", climbed)) + return "", climbed, nil + } + climbed = parent.ATURI + } + d.logger.Warn("thread root is undeterminable: the parent chain is longer than a thread can be", + slog.String("at_uri", atURI), slog.String("dead_end", climbed)) + return "", climbed, nil +} + // subjectCommunityDID answers which community a mapped record belongs to. // // materialize.CommunityDIDOf is THE answer to that question bridge-wide — @@ -140,12 +249,14 @@ func (d *Dispatcher) subjectCommunityDID(ctx context.Context, mapping *store.APO } // recordedState reads what the bridge's OWN outbound row says about a mapped -// subject: its reply depth, and whether it has been withdrawn. +// subject: its reply depth, the thread it hangs in, and whether it has been +// withdrawn. // // depth exists only if the bridge federated the subject outward too. A mapped // subject with no outbound row is a post, or a Lemmy object whose depth this // bridge does not track, so it counts as the top: replies to it are depth 1 -// (this returns 0). +// (this returns 0), and it is its own thread root (this returns no root, which +// threadRootOf reads together with the depth). // // tombstoned is the same fact the outbound_objects branch of resolveSubject // checks, read HERE because both facts come off one row and the caller needs @@ -157,13 +268,13 @@ func (d *Dispatcher) subjectCommunityDID(ctx context.Context, mapping *store.APO // a deeply nested comment at the wrong nesting and, past Lemmy's cap, keep // federating ones it will reject, silently and with no retry; and reading // "not tombstoned" off one would resurrect deleted content. -func (d *Dispatcher) recordedState(ctx context.Context, atURI string) (depth int, tombstoned bool, err error) { +func (d *Dispatcher) recordedState(ctx context.Context, atURI string) (depth int, rootATURI string, tombstoned bool, err error) { state, err := d.objects.GetByATURI(ctx, atURI) if errors.IsNotFound(err) { - return 0, false, nil + return 0, "", false, nil } if err != nil { - return 0, false, fmt.Errorf("read recorded state for %s: %w", atURI, err) + return 0, "", false, fmt.Errorf("read recorded state for %s: %w", atURI, err) } - return state.Depth, state.IsTombstoned(), nil + return state.Depth, rootFromSnapshot(state.TranslatedSnapshot), state.IsTombstoned(), nil } diff --git a/internal/db/migrations/025_object_moderation.sql b/internal/db/migrations/025_object_moderation.sql new file mode 100644 index 0000000..8c15e0b --- /dev/null +++ b/internal/db/migrations/025_object_moderation.sql @@ -0,0 +1,53 @@ +-- +goose Up +-- Task 17c-2: the moderation state THE BRIDGE OWNS. +-- +-- A lock is the first moderation decision with no home in either repo. A +-- community's removal of a post is a record in that community's own repo +-- (social.coves.community.removal, written in ONE commit with the acceptance it +-- replaces); the post itself is the author's and is never touched; the +-- acceptance says only that the post was admitted. Lemmy keeps `locked` on the +-- post row, and atproto has no record whose meaning is "this thread is closed". +-- So the bridge holds it, here — and holding it is worth nothing unless +-- something READS it, which is the point: the next comment does not go out. +-- +-- KEYED BY AT-URI, AND DELIBERATELY NOT A COLUMN ON ap_objects. putMapping's +-- ON CONFLICT DO UPDATE rewrites the whole mapping row, so any re-pin — a +-- re-materialization, a restore, an author edit — would silently CLEAR a lock, +-- and the symptom (comments start federating again) is indistinguishable from +-- the moderators having lifted it. A separate row survives everything that +-- rewrites a mapping. +-- +-- The AT-URI is the key because it is the handle BOTH readers hold: the comment +-- consumer resolves a parent at-uri and never sees an AP id, and the announced +-- moderation path holds a mapping, which carries both. Locks apply to +-- FEDIVERSE-origin posts too — a Lemmy post a native user replies to can be +-- locked exactly like a native one — and those have an at-uri from the moment +-- they are materialized. +CREATE TABLE object_moderation ( + at_uri TEXT PRIMARY KEY, + -- ap_id is the same object's fediverse id, denormalized so the row can be + -- read back from either side of the bridge without a join through + -- ap_objects (whose row is the very thing this table must not depend on). + ap_id TEXT NOT NULL, + -- community_did BINDS the decision to the community that made it. Unbound, + -- any co-hosted community's Undo{Lock} could clear a decision it did not + -- make — and Lemmy hosts many communities per instance by design, so that + -- is ordinary traffic rather than a threat model. + community_did TEXT NOT NULL, + -- locked_at IS the lock: NULL means open. Undo{Lock} clears it instead of + -- deleting the row, because the removal state below shares the row. + locked_at TIMESTAMPTZ, + -- COMMENTS ONLY. A post's removal state lives in the community repo's + -- removal record, which acceptrec writes atomically with the withdrawal of + -- the acceptance; duplicating it here would create two sources of truth for + -- one decision, and they would disagree the first time the commit succeeded + -- and this write did not. Comments have no such record yet (the vendored + -- removal lexicon is post-scoped), which is why the columns exist at all. + removed_at TIMESTAMPTZ, + removal_code TEXT NOT NULL DEFAULT '', + removal_reason TEXT NOT NULL DEFAULT '', + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- +goose Down +DROP TABLE IF EXISTS object_moderation; diff --git a/internal/db/migrations/026_ap_object_thread_root.sql b/internal/db/migrations/026_ap_object_thread_root.sql new file mode 100644 index 0000000..06450a7 --- /dev/null +++ b/internal/db/migrations/026_ap_object_thread_root.sql @@ -0,0 +1,38 @@ +-- +goose Up +-- Task 17c-2 cycle 2: which THREAD a materialized comment hangs in. +-- +-- A community locks a THREAD, and it can lock a Lemmy post exactly as it locks +-- a native one. But the bridge holds no outbound state for fediverse content, +-- and "no outbound state" is read as "the top of the thread" everywhere else +-- (the same boundary recordedState draws for depth) — so a native reply to a +-- LEMMY COMMENT resolved its thread to that comment, and the lock recorded on +-- the post above it never reached the reply. Lemmy threads are mostly Lemmy +-- comments, so that is the ordinary shape, not the corner: reply under any +-- existing comment in a closed thread and the bridge federates something Lemmy +-- rejects server-side. +-- +-- The answer already exists at materialization time and nowhere else afterwards. +-- resolveReplyRefs computes each comment's reply.root as it commits the record +-- (one parent-record read, paid once per comment); recovering it later would +-- mean re-reading that record on the hot path of every native reply. So it is +-- recorded here, on the mapping every reader already holds — the same treatment, +-- for the same reason, as community_did in migration 016. +-- +-- NOT moderation state, and deliberately not on object_moderation: this is +-- thread STRUCTURE, immutable for the life of the record, and re-supplied by the +-- materializer on every re-put — so unlike a lock it cannot be lost to a repin. +-- putMapping still COALESCEs it, because a write path that simply omits the +-- field must not blank a good one. +-- +-- Nullable, and NOT backfilled: a comment's root lives in its record's reply.root +-- (or, for a legacy row, in the thread above it), where no UPDATE can reach. Rows +-- written before this migration read as NULL, and the reader treats that exactly +-- as it did before this column existed — the reply resolves to its own parent as +-- the top of the thread. Those rows heal when the comment is next materialized. +-- +-- No index: every reader arrives holding the mapping it already fetched by ap_id +-- or at_uri and reads this column in memory. Nothing queries BY it. +ALTER TABLE ap_objects ADD COLUMN thread_root_at_uri TEXT; + +-- +goose Down +ALTER TABLE ap_objects DROP COLUMN IF EXISTS thread_root_at_uri; diff --git a/internal/ingest/consent.go b/internal/ingest/consent.go index 2b8fa17..d027528 100644 --- a/internal/ingest/consent.go +++ b/internal/ingest/consent.go @@ -107,7 +107,7 @@ func (h *Handler) handleDelete(ctx context.Context, del *ap.Object, signer strin // the post's own later Creates and Updates, and deleting the record would // hand one community the power to destroy content in all the others. if announcer != nil { - handled, err := h.moderateAnnouncedDelete(ctx, del, targetID) + handled, err := h.moderateAnnouncedDelete(ctx, del, targetID, announcer) if err != nil || handled { return err } @@ -171,17 +171,30 @@ func (h *Handler) handleDelete(ctx context.Context, del *ap.Object, signer strin // visibility mechanism Coves does not consult for that collection, so those // keep the v1 behaviour exactly: delete the record, tombstone the mapping. // -// A NATIVE (bridge-origin) comment is the one case it TAKES without acting: -// declining would run that v1 behaviour against a record in the author's own -// repo. See the branch below. -// NativeCommentModerationDeferred counts announced deletes of NATIVE comments -// that were taken and deliberately not acted on, pending 17c-2's comment -// removal record. It is a DECIDED non-action, so it is counted: the alternative -// reading of a flat zero is "no community has ever tried", and the two must not -// look the same when the feature lands. -var NativeCommentModerationDeferred = expvar.NewInt("tidepool_moderation_native_comment_deferred") +// A NATIVE (bridge-origin) comment is TAKEN here and decided by +// moderateNativeComment: declining would run that v1 behaviour against a record +// in the author's own repo. +// +// The three counters below are the operator's answer to "what has this bridge +// done about native comments?", and they are SEPARATE because the questions are. +// A moderator's removal and an author's own delete arrive on the same activity, +// distinguished only by `summary`, so counting them together (as 17c-1's single +// deferred counter did) reports a number that cannot tell a moderated comment +// from an unmoderated one. +var ( + // NativeCommentRemoved counts announced moderator removals of native + // comments that were RECORDED bridge-side. + NativeCommentRemoved = expvar.NewInt("tidepool_moderation_native_comment_removed") + // NativeCommentRemovalLifted counts the Undos that cleared one. + NativeCommentRemovalLifted = expvar.NewInt("tidepool_moderation_native_comment_removal_lifted") + // NativeCommentSelfDeleted counts summary-less announced deletes of native + // comments: the author's own delete coming home, taken and deliberately + // recorded nowhere. A DECIDED non-action, so it is counted — a flat zero + // must not be readable as "this never happens". + NativeCommentSelfDeleted = expvar.NewInt("tidepool_moderation_native_comment_self_deleted") +) -func (h *Handler) moderateAnnouncedDelete(ctx context.Context, del *ap.Object, targetID string) (bool, error) { +func (h *Handler) moderateAnnouncedDelete(ctx context.Context, del *ap.Object, targetID string, announcer *store.Community) (bool, error) { mapping, err := h.objects.GetByAPID(ctx, targetID) if errors.IsNotFound(err) { if del.HasSummary() { @@ -201,8 +214,7 @@ func (h *Handler) moderateAnnouncedDelete(ctx context.Context, del *ap.Object, t return false, nil } if mapping.Collection != materialize.CollectionPostV2 { - // A NATIVE comment: moderation of it is TAKEN here and deferred, never - // declined into the path below. + // A NATIVE comment: TAKEN here, never declined into the path below. // // Declining used to be safe by accident — a bridge-origin mapping had no // community_did, so authorization refused before reaching this function @@ -215,19 +227,13 @@ func (h *Handler) moderateAnnouncedDelete(ctx context.Context, del *ap.Object, t // resolveSubject reads the still-live outbound row, so replies to the // "removed" comment keep federating regardless. // - // The removal RECORD for comments needs the lexicon work in 17c-2; until - // then the honest outcome is a visible non-action. - // // EVERY announced delete of a native comment is taken, not only a // summary-bearing one: the author's own deletes arrive through the // consumer (their repo), never announced back at us, so an announced // one is either a moderation action or an echo — and neither may reach // a path that deletes the author's record. if mapping.Origin == store.OriginBridge { - NativeCommentModerationDeferred.Add(1) - return true, skip(targetID, - "announced delete of a native comment: moderation of native comments is not "+ - "implemented yet (needs the 17c-2 removal record), taking no action") + return true, h.moderateNativeComment(ctx, del, mapping, announcer) } return false, nil } @@ -303,6 +309,13 @@ func (h *Handler) handleUndo(ctx context.Context, undo *ap.Object, signer string return h.votes.RetractVote(ctx, inner, announcerID) case ap.TypeDelete: return h.handleUndoDelete(ctx, undo, inner, signer, announcer) + case ap.TypeLock: + // Lemmy carries the Lock INLINE inside the Undo rather than referencing + // its id, so the same handler that applied it lifts it. A lock the + // moderators lifted that still refuses comments is moderation state + // nobody can reach: no later activity clears it, because this is the + // only one Lemmy will ever send about it. + return h.handleLock(ctx, inner, announcer, false) case ap.TypeFollow: // A remote undoing a follow of us — the bridge has no followers in // v1 (read-only), nothing to do. @@ -531,12 +544,10 @@ func (h *Handler) restoreNativeContent(ctx context.Context, undo *ap.Object, return skip(mapping.APID, "bare undo of a delete cannot restore the bridge's own content") } if mapping.Collection != materialize.CollectionPostV2 { - // Same deferral as the removal side: comment-level moderation state - // needs 17c-2's record. Taken and counted, never fallen through. - NativeCommentModerationDeferred.Add(1) - return skip(mapping.APID, - "announced restore of a native comment: moderation of native comments is not "+ - "implemented yet (needs the 17c-2 removal record), taking no action") + // A COMMENT: its removal lives in the bridge's own moderation state, not + // in the community repo, so lifting it is a state clear rather than the + // acceptance transition below. + return h.liftNativeCommentRemoval(ctx, undo, mapping, announcer) } if err := h.tombstones.Remove(ctx, mapping.APID, scope); err != nil { return fmt.Errorf("ingest: clear tombstone for %s: %w", mapping.APID, err) @@ -667,6 +678,11 @@ func (h *Handler) authorizeDelete(ctx context.Context, activityID, targetID, sig // predating that column). An answer that cannot be determined REFUSES: an // announced delete is a moderation action by a community over its own // content, and content whose community we cannot name is not that. +// +// Announced Lock/Undo{Lock} (17c-2) ask THIS function, not a copy of it: the +// conjunction is one rule for every announced moderation verb, and a second +// implementation of it is a second place for the two conjuncts to drift apart. +// The messages therefore speak of moderation generally. func (h *Handler) authorizeAnnouncedContentDelete(ctx context.Context, activityID string, mapping *store.APObjectMapping, announcer *store.Community) error { communityDID, err := materialize.CommunityDIDOf(ctx, h.records, mapping) if err != nil { @@ -690,13 +706,13 @@ func (h *Handler) authorizeAnnouncedContentDelete(ctx context.Context, activityI // A live mapping we cannot bind is a permanent inconsistency (a missing // record, a comment with no reply.root). Retrying would re-read the same // hole forever and wedge the ordering key behind it: log and drop. - h.logger.Warn("announced delete: cannot bind target to a community", + h.logger.Warn("announced moderation: cannot bind target to a community", "ap_id", mapping.APID, "at_uri", mapping.ATURI, "collection", mapping.Collection) return skip(activityID, mapping.ATURI+" cannot be bound to a community to authorize against") } if communityDID != announcer.DID { return skip(activityID, fmt.Sprintf( - "announced delete of %s targets content outside %s", mapping.APID, announcer.APGroupID)) + "announced moderation of %s targets content outside %s", mapping.APID, announcer.APGroupID)) } return nil } diff --git a/internal/ingest/handler.go b/internal/ingest/handler.go index b03266d..8645de7 100644 --- a/internal/ingest/handler.go +++ b/internal/ingest/handler.go @@ -326,9 +326,15 @@ func (h *Handler) handleAnnounce(ctx context.Context, announce *ap.Object, signe return h.handleDelete(ctx, inner, signer, community) case ap.TypeUndo: return h.handleUndo(ctx, inner, signer, community) + case ap.TypeLock: + // A community closing one of its own threads. The Undo arrives on the + // TypeUndo branch above and lands in the same handler with locked=false. + return h.handleLock(ctx, inner, community, true) default: - // Lock, Add, Remove, Block, ... — moderation activities the bridge - // does not translate in v1. + // Add, Remove, Block, ... — moderation activities the bridge does not + // translate yet. Remove in particular is NOT content removal in Lemmy + // (it is un-pin / demote-moderator, dispatched by `target`), so it must + // never be folded in beside Lock on the assumption that it is. return skip(announce.ID, "unsupported announced activity type "+inner.Type) } } diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 5f64f7b..e6a3835 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -258,7 +258,13 @@ func newHarness(t *testing.T) *harness { // failure surfaces as a missing acceptance record rather than as // anything about revs. "jetstream_record_revs", "jetstream_dead_letters", "consumer_cursors", - "admissions", "federation_prefs") + "admissions", "federation_prefs", + // The moderation state the bridge owns (migration 025). It MUST be + // cleared: the moderation fixtures' at-uris are package-level constants, + // so a lock left by one run refuses the next run's comment before the + // test that locks it has run — green first, red second, which a single CI + // run never sees. + "object_moderation") custodian, err := identity.NewCustodian(testKEK) require.NoError(t, err) @@ -377,6 +383,12 @@ func newHarness(t *testing.T) *harness { Workers: 1, MaxAttempts: 3, Lease: time.Minute, + // Captured for the same reason the handler's is: a SKIP reason is not + // stored on the event row (the queue logs it and marks the event + // processed), so this log line is the only place the bridge says WHY it + // decided to do nothing — and "did nothing for reason X" versus "did + // nothing for reason Y" is a distinction some behaviours are made of. + Logger: slog.New(slog.NewTextHandler(h.logs, &slog.HandlerOptions{Level: slog.LevelDebug})), }) require.NoError(t, err) diff --git a/internal/ingest/moderation.go b/internal/ingest/moderation.go new file mode 100644 index 0000000..ef922de --- /dev/null +++ b/internal/ingest/moderation.go @@ -0,0 +1,179 @@ +package ingest + +import ( + "context" + "fmt" + + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/materialize" + "tidepool/internal/store" +) + +// moderateNativeComment decides an announced Delete of a NATIVE comment — a +// record in the AUTHOR's own repo that this bridge federated on their behalf. +// +// TWO SHAPES ARRIVE ON ONE ACTIVITY, and only `summary` tells them apart +// (PRESENCE, never emptiness — a moderator who typed no reason sends `""`). The +// post path already turns on exactly this, and the two outcomes here are as far +// apart as they are there: +// +// - WITH a summary: a moderator removed the comment. The decision is recorded +// bridge-side, bound to the community that made it, under the SAME code a +// post removal writes. +// - WITHOUT one: the author deleted their own comment. NOTHING is recorded. +// Writing moderator-discretion here would assert that a moderator acted when +// none did — a removal naming a moderator team that took no action — and it +// would stand until somebody sent an Undo for something that never happened. +// +// Both are TAKEN, and that is the point of the branch: falling through runs the +// v1 destructive path against the author's record, soft-deleting our own mapping +// and tombstoning our own AP id, after which moderateAnnouncedDelete declines +// forever on IsDeleted() and every Lemmy reply beneath the comment is dropped. +// So they share the taking and differ in everything else — including the skip +// REASON and the COUNTER, because "a moderator removed this" and "the author +// deleted it themselves" are opposite answers to the only question an operator +// ever asks here, and one line for both answers neither. +// +// Ownership is NOT re-checked here. authorizeDelete already established that the +// announcing community owns this mapping (the same conjunction the lock path +// uses), and a second copy of that rule is a second place for it to drift. +// +// NOTHING COVES-VISIBLE IS WRITTEN, deliberately: the removal lexicon is +// post-scoped, the comment-subject extension is Coves-owned and has not landed, +// and inventing a record shape here would publish a vocabulary the read path +// does not consult — a moderation decision that looks honored and hides nothing. +// The skip reason says so, because that gap is the whole of what an operator +// needs to know about this path today. +func (h *Handler) moderateNativeComment(ctx context.Context, del *ap.Object, + mapping *store.APObjectMapping, announcer *store.Community) error { + + if !del.HasSummary() { + NativeCommentSelfDeleted.Add(1) + return skip(mapping.APID, + "announced delete of a native comment carries no summary: the author's own delete, "+ + "not a moderator's removal — taken so it cannot destroy their record, and "+ + "recorded nowhere because nobody moderated anything") + } + + // announcer.DID is the community authorizeDelete just proved owns this + // mapping, so the row is bound to the community that made the decision. + if err := h.objects.SetRemoval(ctx, store.ModeratedObject{ + ATURI: mapping.ATURI, + APID: mapping.APID, + CommunityDID: announcer.DID, + }, materialize.RemovalCodeModeratorDiscretion, del.Summary); err != nil { + return fmt.Errorf("ingest: record removal of %s: %w", mapping.ATURI, err) + } + NativeCommentRemoved.Add(1) + h.logger.Info("community removed a native comment; recorded bridge-side", + "ap_id", mapping.APID, "at_uri", mapping.ATURI, + "community", announcer.APGroupID, "activity", del.ID) + return skip(mapping.APID, + "announced moderator removal of a native comment: recorded bridge-side under "+ + materialize.RemovalCodeModeratorDiscretion+"; nothing is published to the community "+ + "repo until the comment-subject removal lexicon lands") +} + +// liftNativeCommentRemoval is the Undo of the above: the moderators reversed +// their decision, so the bridge-side removal is cleared. +// +// It runs for EVERY announced Undo{Delete} of a native comment, whether or not +// the undone Delete carried a summary — the same reasoning restoreNativeContent +// applies to posts. On the delete side the key separates two opposite actions, +// so presence has to decide; here both readings converge on the same +// non-destructive outcome (a removal that no longer stands), and requiring the +// key would only create a way for a real restore to be dropped, leaving a +// removal the moderators lifted standing forever. +// +// Clearing an object nobody removed is a no-op success, which is what a +// re-delivered Undo is. +func (h *Handler) liftNativeCommentRemoval(ctx context.Context, undo *ap.Object, + mapping *store.APObjectMapping, announcer *store.Community) error { + + if err := h.objects.ClearRemoval(ctx, mapping.ATURI, announcer.DID); err != nil { + return fmt.Errorf("ingest: clear removal of %s: %w", mapping.ATURI, err) + } + NativeCommentRemovalLifted.Add(1) + h.logger.Info("community lifted its removal of a native comment", + "ap_id", mapping.APID, "at_uri", mapping.ATURI, + "community", announcer.APGroupID, "activity", undo.ID) + return skip(mapping.APID, + "announced restore of a native comment: the bridge-side removal is cleared; nothing "+ + "is published to the community repo until the comment-subject removal lexicon lands") +} + +// handleLock applies an announced Lock — a community closing one of its threads +// — and, with locked=false, the Undo that lifts it. +// +// A lock is state NOBODY ELSE CAN HOLD. The post is the author's record, the +// acceptance says only that it was admitted, and Lemmy keeps the flag on its own +// post row; so the bridge records it (store.ObjectModeration) and the native +// comment consumer reads it back before federating a reply. Recording it and +// then federating a reply under it would be worse than not recording it at all: +// Lemmy rejects comments on locked posts server-side, so the reply buys a failed +// delivery, a retry loop and finally a poisoned row whose cause is a moderator +// decision nothing in the delivery names. +// +// AUTHORIZATION IS THE SAME CONJUNCTION AS EVERY OTHER ANNOUNCED MODERATION +// ACTION (decision 18), asked of the same function: the announcing community +// must OWN the target's mapping. Not authority equality — Lemmy co-hosts many +// communities per instance and SameAuthority is true across all of them, so +// authority alone would let one moderator team freeze every thread in every +// community beside theirs. +// +// Nothing here enqueues, and nothing can: the write is a bridge-side state +// write, and an inbound moderation action echoed back is an activity aimed at +// the moderators who just sent it. +func (h *Handler) handleLock(ctx context.Context, lock *ap.Object, announcer *store.Community, locked bool) error { + if announcer == nil { + // A bare Lock has no community behind it. The verb IS a community + // decision — Lemmy announces every one of them through the group — so a + // direct one is either a mistake or somebody claiming an authority the + // signature does not carry. + return skip(lock.ID, "bare lock is not a community decision; only an announced lock is") + } + targetID := refID(lock.Object) + if targetID == "" { + return errors.NewValidationError("lock", "lock carries no object id") + } + + mapping, err := h.objects.GetByAPID(ctx, targetID) + if errors.IsNotFound(err) { + // Nothing was ever bridged under this id, so there is nothing to refuse + // comments on. Unlike a Delete there is no marker worth laying: a lock + // for content that does not exist here suppresses nothing, and a future + // Create would carry no reply the lock could apply to anyway. + return skip(lock.ID, "lock of an object the bridge has no mapping for: "+targetID) + } + if err != nil { + return fmt.Errorf("ingest: look up mapping for lock of %s: %w", targetID, err) + } + if mapping.IsDeleted() { + // A soft-deleted mapping is content already withdrawn: replies to it are + // refused by subject resolution long before a lock could matter. It is + // also the one state in which the authorization below cannot bind the + // target to a community (the record its fallback reads is gone), so + // declining here keeps an unbindable target from being recorded against + // whichever community happened to announce. + return skip(lock.ID, "lock of an already-deleted object: "+targetID) + } + if err := h.authorizeAnnouncedContentDelete(ctx, lock.ID, mapping, announcer); err != nil { + return err + } + + // announcer.DID is the community the authorization above just proved OWNS + // this mapping — the same DID CommunityDIDOf answered with — so the row is + // bound to the community that made the decision, not to whoever announced. + if err := h.objects.SetLock(ctx, store.ModeratedObject{ + ATURI: mapping.ATURI, + APID: mapping.APID, + CommunityDID: announcer.DID, + }, locked); err != nil { + return fmt.Errorf("ingest: record lock state for %s: %w", mapping.ATURI, err) + } + h.logger.Info("community lock state applied", + "ap_id", mapping.APID, "at_uri", mapping.ATURI, + "community", announcer.APGroupID, "locked", locked, "activity", lock.ID) + return nil +} diff --git a/internal/ingest/moderation_lock_test.go b/internal/ingest/moderation_lock_test.go new file mode 100644 index 0000000..0b9ee52 --- /dev/null +++ b/internal/ingest/moderation_lock_test.go @@ -0,0 +1,1166 @@ +package ingest + +import ( + "context" + "database/sql" + "encoding/json" + stderrors "errors" + "expvar" + "fmt" + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/ap" + "tidepool/internal/consume" + "tidepool/internal/echo" + "tidepool/internal/errors" + "tidepool/internal/materialize" + "tidepool/internal/store" +) + +// TASK 17c-2 — A LOCK IS MODERATION STATE THE BRIDGE OWNS. +// +// A lock is the first moderation decision with no home in either repo. A +// removal lives in the community's repo as a record; a lock has nowhere to go — +// the post is the author's, the acceptance says only that it was admitted, and +// Lemmy's own model keeps the flag on the post row. So the bridge has to hold +// it, and holding it is worth nothing unless something READS it: the whole +// point of a lock is that the next comment does not go out. +// +// That last half is what this file exists for. Recording a lock and then +// federating a reply under it is worse than not recording it at all: Lemmy +// rejects comments on locked posts server-side, so the delivery fails, retries, +// and eventually poisons — the author sees their comment sitting in their own +// repo forever with no explanation anywhere, while the operator sees a poisoned +// delivery whose cause is a moderator decision three tables away. +// +// THE FIXTURE HAS TWO COMMUNITIES, CO-HOSTED, AND TWO NATIVE ACTORS. +// Decision 18's rule is a CONJUNCTION — the signer must BE the community AND +// the target must belong to it — and in a one-community world those are the +// same fact, so an implementation checking either conjunct passes every test. +// The second native actor is here for the same reason one step ahead: a ban is +// (community, actor, content), and a one-actor fixture cannot tell "cancel that +// actor's deliveries to that community" from "cancel everything". +const ( + mtLockActivity = "https://lemmy.world/activities/announce/lock/mt-lock" + mtUnlockActivity = "https://lemmy.world/activities/announce/undo/mt-lock" + mtCrossLock = "https://lemmy.world/activities/announce/lock/mt-cross-lock" + mtOtherLock = "https://lemmy.world/activities/announce/lock/mt-other-lock" + + // The fediverse half of the fixture: a Lemmy human, and their comment on the + // Lemmy post the standard page fixture materializes. + mlReplier = "https://lemmy.world/u/replier" + mlLemmyComment = "https://lemmy.world/comment/9101" + + // The SECOND native post in community A: another thread, in the same + // community, that no moderator has touched. It is the control for the + // coarsest wrong fix — refusing everything once anything is locked — which a + // one-thread fixture cannot see, exactly as a one-community fixture cannot + // see a per-community over-reach. + mtOtherPostRKey = "3lzmtpost00002" + mtOtherPostRev = "3lzmtrev000020" + mtOtherPostATURI = "at://" + mtAuthorDID + "/social.coves.community.postv2/" + mtOtherPostRKey + mtOtherPostTime = int64(1_775_000_000_000_200) +) + +// The comment fixtures. Their revs and CIDs are reused VERBATIM on every retry: +// "an identical retry is admitted" is the contract, and a retry that changed +// the rev would be admitted by the rev gate for reasons that have nothing to do +// with the lock being lifted. +var ( + // mtDirectReply hangs DIRECTLY under the post. Its parent IS the locked + // object, so the parent alone answers the question. + mtDirectReply = nativeComment{ + did: mtCommenterDID, rkey: "3lzmtcomment01", + root: nativeRef{mtPostATURI, mtPostCID}, + parent: nativeRef{mtPostATURI, mtPostCID}, + createRev: "3lzmtrev000010", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6i", + editRev: "3lzmtrev000011", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6j", + timeUS: 1_775_000_000_000_100, + } + + // mtNestedReply hangs under mtDirectReply — one level down, which is where + // an ordinary conversation goes. Its PARENT is a comment no moderator has + // touched; its thread ROOT is the post. + mtNestedReply = nativeComment{ + did: mtAuthorDID, rkey: "3lzmtcomment02", + root: nativeRef{mtPostATURI, mtPostCID}, + parent: nativeRef{"at://" + mtCommenterDID + "/social.coves.community.comment/3lzmtcomment01", "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6i"}, + createRev: "3lzmtrev000012", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6k", + editRev: "3lzmtrev000013", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6m", + timeUS: 1_775_000_000_000_120, + } + + // mtLateNestedReply is the same shape, arriving AFTER the lock: the reply + // somebody writes to a conversation the moderators have just closed. + mtLateNestedReply = nativeComment{ + did: mtCommenterDID, rkey: "3lzmtcomment03", + root: nativeRef{mtPostATURI, mtPostCID}, + parent: nativeRef{"at://" + mtCommenterDID + "/social.coves.community.comment/3lzmtcomment01", "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6i"}, + createRev: "3lzmtrev000014", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6n", + editRev: "3lzmtrev000015", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6p", + timeUS: 1_775_000_000_000_140, + } + + // The untouched thread: a direct reply to the OTHER post, and a nested reply + // beneath it. Same author, same community, same shape — different thread. + mtOtherDirectReply = nativeComment{ + did: mtCommenterDID, rkey: "3lzmtcomment04", + root: nativeRef{mtOtherPostATURI, mtPostCID}, + parent: nativeRef{mtOtherPostATURI, mtPostCID}, + createRev: "3lzmtrev000016", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6q", + editRev: "3lzmtrev000017", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6r", + timeUS: 1_775_000_000_000_160, + } + mtOtherNestedReply = nativeComment{ + did: mtAuthorDID, rkey: "3lzmtcomment05", + root: nativeRef{mtOtherPostATURI, mtPostCID}, + parent: nativeRef{"at://" + mtCommenterDID + "/social.coves.community.comment/3lzmtcomment04", "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6q"}, + createRev: "3lzmtrev000018", createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6s", + editRev: "3lzmtrev000019", editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6t", + timeUS: 1_775_000_000_000_180, + } +) + +// TestALockedPostRefusesNativeCommentsUntilItIsLifted is the OUTER CONTRACT for +// the lock vertical. +// +// GIVEN a native post accepted into community A, WHEN A announces a Lock for +// it, THEN the lock is recorded, a native comment on that post is REFUSED with +// a parent-locked reason an operator can read, and WHEN A announces Undo{Lock} +// the identical comment is admitted and federates. +// +// Every step runs through the real path: the signed inbox, the real queue and +// handler, the real dispatcher, engine and enqueuer. The lock and the comment +// arrive on OPPOSITE SIDES of the bridge — one over HTTP from Lemmy, one over +// Jetstream from the author's PDS — and the only thing that can join them is +// state the bridge durably owns. A test that reached into a store to set the +// flag would prove the reader works while leaving the writer, and the seam +// between them, entirely untested. +func TestALockedPostRefusesNativeCommentsUntilItIsLifted(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + + // Snapshotted BEFORE the lock, so an enqueue caused by the LOCK ITSELF + // cannot be folded into the baseline. Boomerang suppression for moderation + // is structural today (the materializer's paths take no side effect and so + // cannot enqueue), and a lock that echoes back is an activity aimed at the + // moderators who just sent it. + activitiesBefore := rowCount(t, h.db, "outbound_activities") + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + dropsBefore := dropSnapshot() + + // --- WHEN: community A locks its own post, honestly signed. + h.announceLock(world.groupA, mtLockActivity, mtPostAPID) + + // The echo classifier must NOT have taken this. carriesPayload is an + // ALLOWLIST (Announce|Create|Update|Undo) and Lock is deliberately outside + // it, because a Lock's `object` is its TARGET — and the target of every + // inbound moderation action against native content is, by definition, one of + // OUR ids. Adding the new verbs to that allowlist "for completeness" would + // drop every moderation action the bridge receives AND score each one as a + // successful suppression, which is 17a's HIGH re-committed in new vocabulary. + assert.Equal(t, dropsBefore, dropSnapshot(), + "an announced Lock of our own post is GENUINE remote traffic: the id it names is "+ + "ours precisely because the community is moderating our content, and reading that "+ + "as an echo silently disables inbound moderation while the drop counter reports "+ + "it as working") + + event, err := h.events.GetEvent(ctx, mtLockActivity) + require.NoError(t, err) + assert.NotNil(t, event.ProcessedAt, "the lock is DECIDED, not left retrying: %s", event.Error) + assert.Nil(t, event.FailedAt, "nor poisoned") + + // --- THEN: a native comment on the locked post is REFUSED. + // + // require, not assert: everything after this point is about what a refusal + // looks like, and if the comment federated instead there is nothing left to + // characterise — the harm (a Note delivered to a community that rejects + // comments on that post) has already happened. + err = world.dispatcher.HandleEvent(ctx, mtCommentEvent(t)) + require.Error(t, err, + "a comment under a locked post must be REFUSED: Lemmy rejects it server-side, so "+ + "federating it buys a failed delivery, a retry loop and finally a poisoned row "+ + "whose cause is a moderator decision nothing in the delivery names") + + assert.True(t, stderrors.Is(err, consume.ErrPermanentEvent), + "and refused PERMANENTLY, so the event dead-letters and the cursor moves on: a "+ + "transient error would block every other native user's traffic behind one locked "+ + "thread, retrying a decision only a moderator can change (err=%v)", err) + assert.Contains(t, err.Error(), "parent-locked", + "with the REASON in the message: the connector stores err.Error() as the dead "+ + "letter's last_error, and that string is the only surface an operator triaging "+ + "the queue — or answering the author asking where their comment went — has to "+ + "go on. A silent skip returning nil is the failure mode this asserts against") + + assert.Zero(t, outboundRowsFor(t, h.db, mtDirectReply.atURI()), + "and NOTHING is written for the refused comment: an outbound_objects row is what a "+ + "later delete is rebuilt from, so a row here means the bridge believes it "+ + "federated a comment it never sent") + assert.Equal(t, activitiesBefore, rowCount(t, h.db, "outbound_activities"), + "nor is anything enqueued — not by the lock, and not by the comment it refused") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "...and no delivery") + + // --- AND: the lock is RECORDED, bound to the community that made it. + // + // The binding is not bookkeeping. The reader that lifts a lock has to know + // whose lock it is, and an unbound row lets any co-hosted community's + // Undo{Lock} clear a decision it did not make. + communityDID, locked, found := lockStateFor(t, h.db, mtPostATURI) + require.True(t, found, + "the bridge must own this state: neither repo can hold it — the post is the "+ + "author's and the acceptance says only that it was admitted — so a lock that is "+ + "not recorded here survives nothing, not a restart and not the next comment") + assert.True(t, locked, "and it must read as locked") + assert.Equal(t, world.communityADID, communityDID, + "bound to community A, the community that locked it") + + // --- WHEN: the moderators lift it. + h.announceUndoLock(world.groupA, mtUnlockActivity, mtLockActivity+"/lock", mtPostAPID) + + _, stillLocked, _ := lockStateFor(t, h.db, mtPostATURI) + assert.False(t, stillLocked, + "Undo{Lock} clears the lock: a lock the moderators lifted that still refuses "+ + "comments is moderation state nobody can reach — no later activity clears it, "+ + "because Lemmy has already sent the only one it will ever send") + + // --- THEN: the IDENTICAL comment — same rkey, same rev, same cid — is + // admitted and federates. + // + // Identical on purpose: the refusal must have left NO trace that makes a + // retry a no-op. The rev gate is claimed inside the same transaction the + // handler runs in, so a refusal that advanced the gate would swallow the + // retry as a stale replay and the comment would be lost for good — a lock + // lifted, an author retrying, and silence. + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtCommentEvent(t)), + "the identical comment must be admitted once the lock is lifted: the refusal is a "+ + "gate, not a verdict on the record") + + assert.Equal(t, 1, outboundRowsFor(t, h.db, mtDirectReply.atURI()), + "the comment now has outbound state") + assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore, + "and a delivery to the community: 'admitted' means it went out, not merely that "+ + "the handler stopped returning an error") +} + +// TestAReplyBeneathALockedThreadIsRefused closes the bypass. +// +// The refusal asks about the resolved PARENT. A reply to the post has the post +// as its parent, so it is caught; a reply to a COMMENT that federated before +// the lock has an unlocked parent and goes out anyway. That is not an edge +// case, it is the ordinary shape of a conversation — Lemmy locks THREADS, and +// anyone can keep talking simply by hitting reply one level down. +// +// What comes back is the exact harm the lock exists to prevent, with an extra +// step: Lemmy rejects the comment server-side, the delivery retries and +// poisons, and the operator now has a poisoned row for a thread the moderators +// closed, on a post whose lock the bridge did record correctly. +// +// The answer has to come from STATE, not from the record: reply.root is written +// by the author, so trusting it would let anyone reopen a locked thread by +// naming a different root. +func TestAReplyBeneathALockedThreadIsRefused(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNestedThread(t, world) + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtLockActivity, mtPostAPID) + + // A reply to the FIRST comment. Its parent is a comment nobody moderated; + // its thread root is the post the moderators just closed. + err := world.dispatcher.HandleEvent(ctx, mtLateNestedReply.create(t)) + require.Error(t, err, + "a reply one level down is still a reply in a locked THREAD: a lock that only "+ + "stops direct replies stops nothing, because the reply button under every "+ + "existing comment is the normal way a conversation continues") + + assert.True(t, stderrors.Is(err, consume.ErrPermanentEvent), + "refused with the same permanence as a direct reply: an author must not learn "+ + "that where they clicked reply decides whether they get an answer (err=%v)", err) + assert.Contains(t, err.Error(), "parent-locked", + "and with the same reason, so the DLQ shows one cause for one moderator decision "+ + "rather than a locked thread's replies landing under two different stories") + + assert.Zero(t, outboundRowsFor(t, h.db, mtLateNestedReply.atURI()), + "nothing is written for the refused reply") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "and nothing is sent: the delivery is the harm — Lemmy rejects it, the row "+ + "retries, and it poisons with a cause three tables away") + + // --- And it comes back when the moderators reopen the thread. + h.announceUndoLock(world.groupA, mtUnlockActivity, mtLockActivity+"/lock", mtPostAPID) + + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtLateNestedReply.create(t)), + "the identical reply is admitted once the lock is lifted") + assert.Equal(t, 1, outboundRowsFor(t, h.db, mtLateNestedReply.atURI())) + assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore, + "and it federates: a thread that never reopens for nested replies is a lock "+ + "nobody can lift") +} + +// TestAnEditBeneathALockedThreadIsRefused is the same bypass on the OTHER path, +// and it is the one a create-side fix leaves behind. +// +// An update never re-resolves its thread — it reads back the state its create +// wrote, deliberately, so an edit cannot move a comment between communities or +// up the thread. So whatever the create path learns about the thread ROOT has +// to be on that stored state, or the edit sails through a lock the create is +// refused by: the author of an existing reply keeps a live, editable surface +// inside a closed thread, and every edit is delivered to a community that has +// stopped accepting comments on it. +func TestAnEditBeneathALockedThreadIsRefused(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNestedThread(t, world) + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtLockActivity, mtPostAPID) + + err := world.dispatcher.HandleEvent(ctx, mtNestedReply.edit(t)) + require.Error(t, err, + "an edit to a reply in a locked thread must be refused too: an Update{Note} is a "+ + "delivery like any other, and the community that closed the thread has to accept "+ + "it for the edit to mean anything") + + assert.True(t, stderrors.Is(err, consume.ErrPermanentEvent), + "with the same permanence (err=%v)", err) + assert.Contains(t, err.Error(), "parent-locked", "and the same reason") + + assert.Equal(t, mtNestedReply.createCID, outboundCIDFor(t, h.db, mtNestedReply.atURI()), + "and the refused edit wrote NOTHING: outbound state that advanced to the edited "+ + "version is state claiming the bridge federated an edit it never sent, and every "+ + "later delete would be rebuilt from it") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "nor was anything delivered") + + // --- And the edit lands once the thread reopens. + h.announceUndoLock(world.groupA, mtUnlockActivity, mtLockActivity+"/lock", mtPostAPID) + + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtNestedReply.edit(t)), + "the identical edit is admitted after the unlock: the refusal is a gate, not a "+ + "verdict on the record") + assert.Equal(t, mtNestedReply.editCID, outboundCIDFor(t, h.db, mtNestedReply.atURI()), + "and now the outbound state names the edited version") + assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore, + "and the edit went out") +} + +// TestALockReachesOnlyItsOwnThread is the CONTROL, and it PASSES TODAY — nothing +// refuses a nested reply at all, so it can only fail once a root-aware refusal +// exists to over-reach. It is here as the negative half of the two tests above, +// and it has been tooth-checked (an unconditional refusal in +// refuseUnderLockedParent turns it red). +// +// The scope it pins is per-OBJECT. A second thread in a DIFFERENT community +// could not pin it: an implementation that refused every comment in a community +// holding any locked post would pass that test and fail this one. Same +// community, same author, same nesting — the only difference is which post the +// thread hangs from, which is exactly the difference the lock is keyed on. +func TestALockReachesOnlyItsOwnThread(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + + // A second native post in community A, admitted through the real engine. + require.NoError(t, world.dispatcher.HandleEvent(ctx, + mtPostEventFor(t, mtOtherPostRKey, "create", mtOtherPostRev, mtPostCID, mtOtherPostTime)), + "precondition: a second thread exists in the same community") + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtOtherDirectReply.create(t)), + "precondition: it has a reply to nest under") + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtOtherLock, mtPostAPID) + + // Everything about the untouched thread keeps working: a nested reply, + // whose ROOT is the other post... + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtOtherNestedReply.create(t)), + "a reply in an UNLOCKED thread must still federate: a lock is per-object, and one "+ + "closed thread that silences a community is a moderator action nobody asked for "+ + "and nobody can see — the replies simply stop arriving") + assert.Equal(t, 1, outboundRowsFor(t, h.db, mtOtherNestedReply.atURI())) + + // ...and an edit to the reply already standing in it. + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtOtherDirectReply.edit(t)), + "and an edit in that thread lands too") + assert.Equal(t, mtOtherDirectReply.editCID, outboundCIDFor(t, h.db, mtOtherDirectReply.atURI())) + + assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore, + "both went out") + + // The locked thread is genuinely locked, so this is a scope test and not an + // accident of the lock never having been recorded. + _, locked, found := lockStateFor(t, h.db, mtPostATURI) + require.True(t, found) + require.True(t, locked, "precondition: the OTHER post really is locked") +} + +// seedNestedThread federates the two comments a lock bypass needs to exist +// before the lock: a direct reply to the post, and a reply to THAT. Both are +// admitted through the real path while the thread is open — which is what makes +// them the shape a later lock has to reach, rather than fixture rows asserting +// their own conclusion. +func seedNestedThread(t *testing.T, world moderationWorld) { + t.Helper() + ctx := context.Background() + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtDirectReply.create(t)), + "precondition: a direct reply federates while the thread is open") + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtNestedReply.create(t)), + "precondition: a reply to that reply federates too") +} + +// outboundCIDFor reads the version the bridge believes it last federated. +func outboundCIDFor(t *testing.T, db *sql.DB, atURI string) string { + t.Helper() + var cid string + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT last_cid FROM outbound_objects WHERE at_uri = $1`, atURI).Scan(&cid), + "read outbound state for %s", atURI) + return cid +} + +// TestCrossCommunityLockIsRefused is the RELATIONAL case: community B, on the +// same instance as A, announces a Lock for A's post. +// +// Decision 18's conjunction collapses in a one-community fixture, so this is the +// only shape that can tell "the signer IS the community" from "the target is IN +// the community". Nothing about this delivery is malformed — B is followed, B +// signs as itself, and SameAuthority is true across every community lemmy.world +// hosts. It is simply not B's post to lock. +// +// The consequence of getting it wrong is quiet and total: one moderator team +// could freeze every thread in every community co-hosted with theirs, and the +// only visible symptom would be authors' comments dead-lettering with a reason +// that names a lock nobody in their community made. +func TestCrossCommunityLockIsRefused(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + + h.announceLock(world.groupB, mtCrossLock, mtPostAPID) + + _, locked, found := lockStateFor(t, h.db, mtPostATURI) + assert.False(t, found && locked, + "community B may not lock community A's post: the target's community mapping is "+ + "the authorization input, and a signer that merely shares an instance with it "+ + "has no claim on it") + + // A's post is untouched, and the strongest evidence of that is not the + // absence of a row — it is that the thread still works. + _, _, err := h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, world.digestRKey) + assert.NoError(t, err, "A's acceptance stands: a refused lock changes nothing about it") + + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtCommentEvent(t)), + "and a native comment still federates: a refusal that half-applied — recorded "+ + "nowhere but read somewhere — would be indistinguishable from a legitimate lock "+ + "to everyone except the community that never made it") + assert.Equal(t, 1, outboundRowsFor(t, h.db, mtDirectReply.atURI()), + "the comment has outbound state, so it really did go out") +} + +// announceLock delivers Lemmy's lock shape: Announce{Lock} from the community, +// whose INNER Lock is attributed to the acting MODERATOR (a /u/ actor) and whose +// `object` is the post being locked — the target, not a payload. +func (h *harness) announceLock(group *remoteActor, activityID, targetID string) { + h.t.Helper() + require.Equal(h.t, http.StatusAccepted, h.deliver(group, map[string]any{ + "id": activityID, + "type": "Announce", + "actor": group.id, + "audience": group.id, + "cc": []any{group.id + "/followers"}, + "object": lockActivity(group, activityID+"/lock", targetID), + })) + h.drain() +} + +// announceUndoLock delivers the unlock shape: Announce{Undo{Lock}} with the Lock +// carried INLINE (Lemmy embeds it rather than referencing its id). +func (h *harness) announceUndoLock(group *remoteActor, activityID, lockActivityID, targetID string) { + h.t.Helper() + require.Equal(h.t, http.StatusAccepted, h.deliver(group, map[string]any{ + "id": activityID, + "type": "Announce", + "actor": group.id, + "audience": group.id, + "cc": []any{group.id + "/followers"}, + "object": map[string]any{ + "id": activityID + "/undo", + "type": "Undo", + "actor": modActorID, + "audience": group.id, + "cc": []any{group.id}, + "object": lockActivity(group, lockActivityID, targetID), + }, + })) + h.drain() +} + +func lockActivity(group *remoteActor, activityID, targetID string) map[string]any { + return map[string]any{ + "id": activityID, + "type": "Lock", + "actor": modActorID, + "object": targetID, + "audience": group.id, + "to": []any{ap.PublicAudience}, + "cc": []any{group.id}, + } +} + +// nativeComment is one native reply, described by everything a commit frame +// needs and nothing else. Building the create and the edit from ONE fixture is +// what makes "the identical retry is admitted" expressible: every call returns +// the same bytes, so a retry differs from the refused attempt in nothing at all +// — not the rev, which the gate would otherwise admit for its own reasons. +type nativeComment struct { + did string + rkey string + // root and parent are the reply refs as the RECORD asserts them. The record + // is the author's, so these are claims; what the bridge does with them is + // the point of the tests below. + root, parent nativeRef + // The create and the edit carry different revs and CIDs, because they are + // different commits on one record and the rev gate orders them. + createRev, createCID string + editRev, editCID string + timeUS int64 +} + +type nativeRef struct{ uri, cid string } + +func (c nativeComment) atURI() string { + return "at://" + c.did + "/" + materialize.CollectionComment + "/" + c.rkey +} + +// apID is the AP object id this comment federates under — the id a community +// names when it announces a moderation action against it. +func (c nativeComment) apID() string { + return mtUserOrigin + "/ap/object/" + c.did + "/" + materialize.CollectionComment + "/" + c.rkey +} + +func (c nativeComment) create(t *testing.T) *consume.JetstreamEvent { + t.Helper() + return c.commit(t, "create", c.createRev, c.createCID, c.timeUS) +} + +func (c nativeComment) edit(t *testing.T) *consume.JetstreamEvent { + t.Helper() + return c.commit(t, "update", c.editRev, c.editCID, c.timeUS+1) +} + +func (c nativeComment) commit(t *testing.T, operation, rev, cid string, timeUS int64) *consume.JetstreamEvent { + t.Helper() + frame := fmt.Sprintf(`{ + "did": %q, "time_us": %d, "kind": "commit", + "commit": { + "rev": %q, "operation": %q, + "collection": %q, + "rkey": %q, "cid": %q, + "record": { + "$type": %q, + "reply": { + "root": {"uri": %q, "cid": %q}, + "parent": {"uri": %q, "cid": %q} + }, + "content": "a reply whose fate the community's lock decides", + "createdAt": "2026-08-13T11:00:00.000Z" + } + } +}`, c.did, timeUS, rev, operation, + materialize.CollectionComment, c.rkey, cid, + materialize.CollectionComment, + c.root.uri, c.root.cid, c.parent.uri, c.parent.cid) + var event consume.JetstreamEvent + require.NoError(t, json.Unmarshal([]byte(frame), &event), "the frame must be valid wire JSON") + return &event +} + +// mtCommentEvent is the direct reply the outer contract uses. +func mtCommentEvent(t *testing.T) *consume.JetstreamEvent { + t.Helper() + return mtDirectReply.create(t) +} + +// lockStateFor reads the bridge's own moderation state for one object. +// +// Read as SQL rather than through a store: this tier is asserting that the +// state is DURABLE and BOUND to a community, and a test that went through the +// same accessor the implementation writes with could not tell a persisted lock +// from one held in memory. +func lockStateFor(t *testing.T, db *sql.DB, atURI string) (communityDID string, locked, found bool) { + t.Helper() + err := db.QueryRowContext(context.Background(), ` + SELECT community_did, locked_at IS NOT NULL + FROM object_moderation + WHERE at_uri = $1`, atURI).Scan(&communityDID, &locked) + if err == sql.ErrNoRows { + return "", false, false + } + require.NoError(t, err, "read the bridge-owned moderation state for %s", atURI) + return communityDID, locked, true +} + +// outboundRowsFor counts the outbound state rows for one at-uri. +func outboundRowsFor(t *testing.T, db *sql.DB, atURI string) int { + t.Helper() + var n int + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM outbound_objects WHERE at_uri = $1`, atURI).Scan(&n)) + return n +} + +// TASK 17c-2, CYCLE 2 — THE THREAD ABOVE A FEDIVERSE COMMENT. +// +// A native reply to a LEMMY comment resolves its thread root to that comment: +// the bridge holds no outbound state for fediverse content, and "no state" is +// read as "the top of the thread" everywhere else (it is the same boundary +// recordedState draws for depth). So a lock recorded on the Lemmy POST above it +// never reaches the reply. +// +// Lemmy threads are mostly Lemmy comments, so this is not the exotic corner — +// it is the ordinary one. A moderator closes a busy thread, a native user hits +// reply under any existing comment in it, and the bridge federates a comment +// Lemmy rejects server-side: failed delivery, retry loop, poisoned row. Exactly +// the noise the lock exists to prevent, on the shape it will meet most often. +// +// The answer has to come from the materialized RECORD's reply.root — a read of +// a repo the bridge hosts (the bridged author's), not a network call. + +// fediverseThread is a Lemmy post with a Lemmy comment under it, both +// materialized into community A, plus a native reply hanging off the comment. +type fediverseThread struct { + postAPID string + postATURI string + commentATURI string + reply nativeComment +} + +// seedFediverseThread materializes the Lemmy half through the REAL announce +// path: the post, then a Lemmy human's comment on it. Both land in bridged +// authors' repos, which is where the reply.root a lock has to be read from +// lives. +func seedFediverseThread(t *testing.T, h *harness, world moderationWorld) fediverseThread { + t.Helper() + ctx := context.Background() + + require.Equal(t, http.StatusAccepted, + h.deliver(world.groupA, loadFixture(t, "announce_create_page_lemmy_world.json"))) + h.drain() + post, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err, "precondition: the Lemmy post materialized into community A") + + h.serveObject("/u/replier", person(mlReplier, "replier", nil)) + comment := note(mlLemmyComment, mlReplier, pageID, "a Lemmy comment in the thread", + "2026-08-13T10:00:00.000000Z") + h.serveObject("/comment/9101", comment) + require.Equal(t, http.StatusAccepted, h.deliver(world.groupA, + announceCreateNote("https://lemmy.world/activities/announce/create/ml-9101", mlReplier, comment))) + h.drain() + lemmyComment, err := h.objects.GetByAPID(ctx, mlLemmyComment) + require.NoError(t, err, "precondition: the Lemmy comment materialized under it") + + return fediverseThread{ + postAPID: pageID, + postATURI: post.ATURI, + commentATURI: lemmyComment.ATURI, + reply: nativeComment{ + did: mtCommenterDID, rkey: "3lzmtcomment06", + // The record's own claim about its thread. It is the AUTHOR's claim, + // so nothing may be decided on it — but it is what a real client + // writes, so the fixture writes it too. + root: nativeRef{post.ATURI, post.CID}, + parent: nativeRef{lemmyComment.ATURI, lemmyComment.CID}, + createRev: "3lzmtrev000030", + createCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6u", + editRev: "3lzmtrev000031", + editCID: "bafyreih5xbmigkq5ikyhqiqhqzbwuqjxeitgtzwyxvjhfsfvswsxmnnf6v", + timeUS: 1_775_000_000_000_300, + }, + } +} + +// TestAReplyBeneathALockedFediverseThreadIsRefused is the common case. +func TestAReplyBeneathALockedFediverseThreadIsRefused(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + thread := seedFediverseThread(t, h, world) + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + h.announceLock(world.groupA, mtLockActivity, thread.postAPID) + + _, locked, found := lockStateFor(t, h.db, thread.postATURI) + require.True(t, found && locked, + "precondition: a community can lock a LEMMY post it hosts — locks are not a "+ + "native-content feature, and most locked threads will be this shape") + + err := world.dispatcher.HandleEvent(ctx, thread.reply.create(t)) + require.Error(t, err, + "a native reply under a Lemmy comment is a reply in the Lemmy POST's thread: our "+ + "state stops at the comment because we hold no outbound row for fediverse "+ + "content, but Lemmy's lock is on the post, and Lemmy is what rejects the reply") + + assert.True(t, stderrors.Is(err, consume.ErrPermanentEvent), + "refused with the same permanence as a native thread: the author who replied under "+ + "a Lemmy comment did the same thing as the author who replied under a native one "+ + "(err=%v)", err) + assert.Contains(t, err.Error(), "parent-locked", + "and with the same reason — one moderator decision, one story in the DLQ") + + assert.Zero(t, outboundRowsFor(t, h.db, thread.reply.atURI()), + "nothing is written for the refused reply") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "and nothing is sent to a community that will reject it") + + // --- And it comes back when the moderators reopen the thread. + h.announceUndoLock(world.groupA, mtUnlockActivity, mtLockActivity+"/lock", thread.postAPID) + + require.NoError(t, world.dispatcher.HandleEvent(ctx, thread.reply.create(t)), + "the identical reply is admitted once the lock is lifted") + assert.Equal(t, 1, outboundRowsFor(t, h.db, thread.reply.atURI())) + assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore, + "and it federates") +} + +// TestAReplyBeneathAnUnlockedFediverseThreadFederates is the CONTROL, and it +// PASSES TODAY — nothing refuses these replies at all. +// +// It is the half that costs something to get wrong in the other direction: +// reading a thread root out of a materialized record is a read that can fail, +// be absent, or dead-end, and every one of those must resolve toward federating. +// A native reply into an ordinary open Lemmy thread is the single most common +// write this consumer handles. +func TestAReplyBeneathAnUnlockedFediverseThreadFederates(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + thread := seedFediverseThread(t, h, world) + + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + require.NoError(t, world.dispatcher.HandleEvent(ctx, thread.reply.create(t)), + "a reply in an open Lemmy thread must federate: this is the ordinary case, and a "+ + "lock check that refuses when it cannot read the thread takes the whole comment "+ + "path down for every community that has ever locked anything") + assert.Equal(t, 1, outboundRowsFor(t, h.db, thread.reply.atURI())) + + require.NoError(t, world.dispatcher.HandleEvent(ctx, thread.reply.edit(t)), + "and an edit to it lands too") + assert.Greater(t, rowCount(t, h.db, "outbound_deliveries"), deliveriesBefore) +} + +// TestAFediverseThreadIsUnaffectedByALockElsewhere is the SCOPE control for the +// same read, and it also PASSES TODAY. +// +// The community holds a real, standing lock — on the NATIVE post — while the +// Lemmy thread beside it is open. This is the case a conservative fallback +// gets wrong: answering "we could not establish this thread, and this community +// holds locks, so refuse" would park every reply to every fediverse comment in +// any community that has ever locked one post. The refusal must be about THIS +// thread or it is not about a thread at all. +func TestAFediverseThreadIsUnaffectedByALockElsewhere(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + thread := seedFediverseThread(t, h, world) + + h.announceLock(world.groupA, mtOtherLock, mtPostAPID) + _, locked, found := lockStateFor(t, h.db, mtPostATURI) + require.True(t, found && locked, "precondition: the community holds a standing lock") + + require.NoError(t, world.dispatcher.HandleEvent(ctx, thread.reply.create(t)), + "a reply in a DIFFERENT, open thread must still federate: a lock is per-object, "+ + "and one locked post that silences every fediverse thread in the community is a "+ + "moderator action nobody took and nobody can see") + assert.Equal(t, 1, outboundRowsFor(t, h.db, thread.reply.atURI())) +} + +// TASK 17c-2, CYCLE 3 — A COMMUNITY'S REMOVAL OF A NATIVE COMMENT IS RECORDED. +// +// 17c-1 left this a DECIDED non-action: an announced delete of a native comment +// was TAKEN and counted, never declined, because declining would have run the v1 +// destructive path against a record in the author's own repo — soft-deleting our +// own mapping and tombstoning our own AP id, after which the comment is +// permanently unmoderatable and every reply beneath it silently disappears. +// +// The state goes in the one place that can hold it. There is nothing to write +// Coves-side: the removal lexicon is POST-scoped, the comment-subject extension +// is Coves-owned and has not landed, and inventing a record shape here would +// publish a vocabulary the read path does not consult — a moderation decision +// that looks acted upon and is not. That gap is asserted below as the CURRENT +// CONTRACT, so the day the extension lands, a test fails and says so. + +// nativeCommentRemoval is the fixture: one native comment, federated into +// community A through the real path, that a moderator then removes. +func seedNativeComment(t *testing.T, world moderationWorld) { + t.Helper() + require.NoError(t, world.dispatcher.HandleEvent(context.Background(), mtDirectReply.create(t)), + "precondition: the native comment federated into community A") +} + +// TestACommunitysRemovalOfANativeCommentIsRecorded is cycle 3's contract. +func TestACommunitysRemovalOfANativeCommentIsRecorded(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNativeComment(t, world) + + commentATURI, commentAPID := mtDirectReply.atURI(), mtDirectReply.apID() + mappingBefore, err := h.objects.GetByAPID(ctx, commentAPID) + require.NoError(t, err, "precondition: the enqueuer mapped the federated comment") + require.Equal(t, world.communityADID, mappingBefore.CommunityDID, + "precondition: bound to the community it federated into — the authorization input") + + activitiesBefore := rowCount(t, h.db, "outbound_activities") + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + communityFramesBefore := len(communityEvents(t, h, world.communityADID)) + require.NotZero(t, communityFramesBefore, + "precondition: the community repo HAS a firehose history (the post's acceptance), so "+ + "'no new frame' below is a measurement and not an empty counter agreeing with itself") + deferredBefore := deferredCommentModerations() + + // --- A moderator of community A removes the comment: Delete WITH summary, + // announced by the community that owns its mapping. + reason := "rule 3: no personal attacks" + h.announceDeleteWithSummary(world.groupA, + "https://lemmy.world/activities/announce/delete/mc-removal", commentAPID, &reason) + + // --- THEN: the bridge records the decision. + state, found := removalStateFor(t, h.db, commentATURI) + require.True(t, found, + "the removal must be RECORDED: a moderator decision the bridge takes and then "+ + "holds nowhere is one no admin surface can answer for and no later Undo can "+ + "reverse — and the alternative was destroying the author's own record") + assert.True(t, state.removed, "removed_at stamps WHEN, so the decision has a time") + assert.Equal(t, world.communityADID, state.communityDID, + "bound to the community that made it: unbound, any co-hosted community's Undo "+ + "could lift a decision it had no part in") + assert.Equal(t, "moderator-discretion", state.code, + "with the SAME code a post removal writes: one moderator action must not read as "+ + "two different things depending on whether they removed a post or a comment") + assert.Equal(t, reason, state.reason, "and the moderator's own reason, intact") + + assert.Equal(t, deferredBefore, deferredCommentModerations(), + "and the placeholder is RETIRED: a removal that is both recorded and counted as "+ + "deferred reports a feature as missing while it works, which is how the counter "+ + "stops meaning anything") + + // --- AND: nothing Coves-visible was written. This is the documented gap, + // asserted as a contract rather than left as an absence nobody checks. + assert.Equal(t, communityFramesBefore, len(communityEvents(t, h, world.communityADID)), + "NO commit frame in the community repo: the removal lexicon is post-scoped and the "+ + "comment-subject extension is Coves-owned and unlanded, so any record written "+ + "here would publish a vocabulary Coves' read path does not consult — a removal "+ + "that looks honored and hides nothing. When that extension lands, this assertion "+ + "is the one that must change") + _, _, err = h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionRemoval, testDigestRKey(commentATURI)) + assert.True(t, errors.IsNotFound(err), + "and specifically no removal record for the comment (err=%v)", err) + _, _, err = h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, world.digestRKey) + assert.NoError(t, err, + "the POST's acceptance is untouched: removing a comment says nothing about the "+ + "post it hangs under") + + // --- AND: the v1 destructive path was not entered. + mapping, err := h.objects.GetByAPID(ctx, commentAPID) + require.NoError(t, err) + assert.False(t, mapping.IsDeleted(), + "our own mapping stays live: soft-deleting it makes moderateAnnouncedDelete decline "+ + "forever on IsDeleted(), so the removal could never be undone") + assert.Equal(t, store.OriginBridge, mapping.Origin, "and it is still ours") + + tombstoned, err := h.tombstones.ExistsFor(ctx, commentAPID, groupID) + require.NoError(t, err) + assert.False(t, tombstoned, + "no tombstone against our own AP id: it suppresses this comment's own later "+ + "activities and drops every Lemmy reply beneath it") + assert.False(t, outboundTombstoned(t, h.db, commentATURI), + "and the bridge's outbound state is not withdrawn: the comment still exists in the "+ + "AUTHOR's repo, and tombstoning our copy would make their own later delete a no-op") + + assert.Equal(t, activitiesBefore, rowCount(t, h.db, "outbound_activities"), + "NOTHING is enqueued: an inbound moderation action is the community telling US what "+ + "it did, and echoing it back is a Delete aimed at the moderators who sent it") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), "...and no delivery") + + event, err := h.events.GetEvent(ctx, "https://lemmy.world/activities/announce/delete/mc-removal") + require.NoError(t, err) + assert.NotNil(t, event.ProcessedAt, "the removal is DECIDED, not left retrying: %s", event.Error) + assert.Nil(t, event.FailedAt, "nor poisoned") + + // --- AND: the moderators can lift it. + h.announceUndoDelete(world.groupA, + "https://lemmy.world/activities/announce/undo/mc-removal", + "https://lemmy.world/activities/announce/delete/mc-removal/delete", + commentAPID, &reason) + + after, found := removalStateFor(t, h.db, commentATURI) + assert.False(t, found && after.removed, + "Undo clears the removal: a decision the moderators reversed that still stands is "+ + "moderation nobody can reach — Lemmy sends no second activity to clear it") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "and the restore enqueues nothing either") +} + +// TestCrossCommunityRemovalOfANativeCommentIsRefused is the RELATIONAL control, +// and it PASSES TODAY — no removal is recorded for anyone yet. +func TestCrossCommunityRemovalOfANativeCommentIsRefused(t *testing.T) { + h := newHarness(t) + world := newModerationWorld(t, h) + seedNativeComment(t, world) + + reason := "not your community's comment" + h.announceDeleteWithSummary(world.groupB, + "https://lemmy.world/activities/announce/delete/mc-cross", mtDirectReply.apID(), &reason) + + state, found := removalStateFor(t, h.db, mtDirectReply.atURI()) + assert.False(t, found && state.removed, + "community B may not remove community A's comment: one moderator team would "+ + "otherwise be able to withdraw content from every community co-hosted with it") + + mapping, err := h.objects.GetByAPID(context.Background(), mtDirectReply.apID()) + require.NoError(t, err) + assert.False(t, mapping.IsDeleted(), + "and the refusal leaves the comment exactly as A admitted it") +} + +// TestAPostRemovalWritesNoBridgeSideRemovalState is the SCOPE control for ruling +// C, and it PASSES TODAY. +// +// removed_at is COMMENTS ONLY. A post's removal is a record in the community's +// own repo, written by acceptrec in ONE commit with the withdrawal of the +// acceptance it replaces. A second copy here would be a second source of truth +// for one decision, and they would disagree the first time the commit succeeded +// and this write did not — leaving a post that reads as removed to the bridge +// and accepted to Coves, or the reverse. +func TestAPostRemovalWritesNoBridgeSideRemovalState(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + + reason := "off topic for this community" + h.announceDeleteWithSummary(world.groupA, + "https://lemmy.world/activities/announce/delete/mc-post", mtPostAPID, &reason) + + _, _, err := h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionRemoval, world.digestRKey) + require.NoError(t, err, + "precondition: the post's removal really did happen, in the community's repo where "+ + "it belongs") + + state, found := removalStateFor(t, h.db, mtPostATURI) + assert.False(t, found && state.removed, + "and it wrote NO bridge-side removal state: the community repo record is the single "+ + "source of truth for a post, and a duplicate here can only ever disagree with it") +} + +// removalStateFor reads the bridge-owned removal state for one object. +type bridgeRemoval struct { + communityDID string + code string + reason string + removed bool +} + +func removalStateFor(t *testing.T, db *sql.DB, atURI string) (bridgeRemoval, bool) { + t.Helper() + var state bridgeRemoval + err := db.QueryRowContext(context.Background(), ` + SELECT community_did, removal_code, removal_reason, removed_at IS NOT NULL + FROM object_moderation + WHERE at_uri = $1`, atURI).Scan(&state.communityDID, &state.code, &state.reason, &state.removed) + if err == sql.ErrNoRows { + return bridgeRemoval{}, false + } + require.NoError(t, err, "read the bridge-owned moderation state for %s", atURI) + return state, true +} + +// outboundTombstoned reports whether the bridge withdrew its own outbound state. +func outboundTombstoned(t *testing.T, db *sql.DB, atURI string) bool { + t.Helper() + var tombstoned bool + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT tombstoned_at IS NOT NULL FROM outbound_objects WHERE at_uri = $1`, + atURI).Scan(&tombstoned), "read outbound state for %s", atURI) + return tombstoned +} + +// deferredCommentModerations reads the 17c-1 placeholder counter, tolerating its +// RETIREMENT: once the removal is real the counter has no reason to exist, and a +// test that could not survive its deletion would force it to be kept. +func deferredCommentModerations() int64 { + counter, _ := expvar.Get("tidepool_moderation_native_comment_deferred").(*expvar.Int) + if counter == nil { + return 0 + } + return counter.Value() +} + +// TestASummarylessDeleteOfANativeCommentRecordsNothing pins the OTHER half of +// the discriminator, and it is the half that writes nothing. +// +// Lemmy's convention — the same one the post path already turns on, by PRESENCE +// and not emptiness — is that `summary` present means a moderator removed it and +// `summary` absent means the author deleted it themselves. So a summary-less +// announced delete of native content is an author's own delete coming home, not +// a moderator's decision, and recording moderator-discretion for it would assert +// that somebody moderated when nobody did: a removal record naming a moderator +// team that took no action, against an author who moderated nobody. +// +// It must still not fall into the v1 destructive path. That path is what the +// 17c-1 branch exists to keep native comments away from: it soft-deletes our own +// mapping and tombstones our own AP id, after which the comment can never be +// moderated again and every Lemmy reply beneath it is dropped. So the branch +// keeps TAKING this activity. It simply records nothing, and says so +// distinguishably — because "we removed it because a moderator said so" and "we +// did nothing because the author deleted their own comment" are different +// answers to the only question an operator ever asks here. +func TestASummarylessDeleteOfANativeCommentRecordsNothing(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNativeComment(t, world) + require.NoError(t, world.dispatcher.HandleEvent(ctx, mtNestedReply.create(t)), + "precondition: a second native comment to aim the other shape at") + + activitiesBefore := rowCount(t, h.db, "outbound_activities") + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + + // The CONTRAST, delivered first: a moderator removal of the other comment. + // Its outcome is the contract test's; what matters here is that the bridge + // does not tell the same story about both. + removalReason := "rule 3: no personal attacks" + const removalActivity = "https://lemmy.world/activities/announce/delete/ms-removal" + h.announceDeleteWithSummary(world.groupA, removalActivity, mtDirectReply.apID(), &removalReason) + + // The shape under test: NO summary key at all, attributed to a Lemmy + // moderator — a foreign attribution, so the echo classifier cannot be what + // decides this (see the sibling test below for the case where it is). + const selfDeleteActivity = "https://lemmy.world/activities/announce/delete/ms-selfdelete" + h.announceDeleteBy(world.groupA, selfDeleteActivity, mtNestedReply.apID(), modActorID, nil) + + // --- THEN: nothing is recorded about it. + _, found := removalStateFor(t, h.db, mtNestedReply.atURI()) + assert.False(t, found, + "a summary-less delete writes NO moderation state: `summary` present is what marks a "+ + "moderator removal, so recording one here asserts a decision no moderator made — "+ + "and it would stand until somebody sent an Undo for an action that never happened") + + // --- AND: the v1 destructive path is still not entered. This is what the + // branch exists for, and it must survive the branch learning to write. + mapping, err := h.objects.GetByAPID(ctx, mtNestedReply.apID()) + require.NoError(t, err) + assert.False(t, mapping.IsDeleted(), + "our own mapping stays live: the v1 path soft-deletes it, and moderateAnnouncedDelete "+ + "then declines forever on IsDeleted() — the comment becomes unmoderatable by a "+ + "delete that was never a moderation action") + assert.Equal(t, store.OriginBridge, mapping.Origin, "and it is still ours") + tombstoned, err := h.tombstones.ExistsFor(ctx, mtNestedReply.apID(), groupID) + require.NoError(t, err) + assert.False(t, tombstoned, + "no tombstone against our own AP id: it drops every Lemmy reply beneath the comment") + assert.False(t, outboundTombstoned(t, h.db, mtNestedReply.atURI()), + "and the bridge does not withdraw its own outbound state on a claim it cannot verify") + + assert.Equal(t, activitiesBefore, rowCount(t, h.db, "outbound_activities"), + "nothing is enqueued by either shape") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), "...and no delivery") + + // --- AND: it is DECIDED, once, with its own reason. + event, err := h.events.GetEvent(ctx, selfDeleteActivity) + require.NoError(t, err) + assert.NotNil(t, event.ProcessedAt, + "a non-action is a decision and must be marked processed, not retried: %s", event.Error) + assert.Nil(t, event.FailedAt, "nor poisoned") + + assert.NotEqual(t, + skipReasonFor(t, h, removalActivity, mtDirectReply.apID()), + skipReasonFor(t, h, selfDeleteActivity, mtNestedReply.apID()), + "and the bridge must not tell the SAME story about both: today a moderator's removal "+ + "of a comment and an author's own delete are logged with one reason and counted in "+ + "one counter, so the operator asking 'did a moderator remove this?' reads an answer "+ + "that cannot distinguish yes from no") +} + +// TestASummarylessDeleteOfANativeCommentByOurPersonaIsDroppedAsAnEcho records +// which mechanism is actually load-bearing when the attribution is TRUTHFUL. +// +// CHARACTERIZATION: this passes today, and it is the comment-shaped twin of the +// post case 17c-1 pinned. A native comment's author IS one of our personas, so a +// truthful self-delete announced back at us is indistinguishable from our own +// Delete coming home — and the echo classifier takes it by the inner ACTOR, +// before any authorization or moderation branch runs. +// +// It is worth pinning because it means the summary-less branch's unverified +// attribution is unreachable for native comments from either direction: a +// persona attribution is dropped here, and a foreign attribution is bounded by +// the ownership conjunct. Take this guard away — by descending into a Delete's +// target, or by probing the wrong table for "ours" — and an author's own delete +// starts arriving at a branch that is about to learn how to write moderation +// state. +func TestASummarylessDeleteOfANativeCommentByOurPersonaIsDroppedAsAnEcho(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + world := newModerationWorld(t, h) + seedNativeComment(t, world) + before := dropSnapshot() + + // Community A — the OWNER, so ownership cannot be what refuses this — and the + // inner Delete attributed to the comment's real author, one of our personas. + h.announceDeleteBy(world.groupA, + "https://lemmy.world/activities/announce/delete/ms-persona", + mtDirectReply.apID(), mtUserOrigin+"/ap/actor/"+mtCommenterDID, nil) + + assert.Equal(t, before[echo.ClassLocalActor]+1, echo.Drops(echo.ClassLocalActor), + "it is dropped as an ECHO, by the inner actor: our own Delete of a native comment "+ + "comes back announced, and the only thing that can tell it from a moderator's is "+ + "whose activity it is") + + _, found := removalStateFor(t, h.db, mtDirectReply.atURI()) + assert.False(t, found, + "so no moderation state is recorded for it: an echo of the author's own delete must "+ + "never become a community-signed removal — the exact failure 17a's M1 was fixed to "+ + "prevent, one collection over") + + mapping, err := h.objects.GetByAPID(ctx, mtDirectReply.apID()) + require.NoError(t, err) + assert.False(t, mapping.IsDeleted(), "and the mapping is untouched") +} + +// skipReasonFor returns the reason the queue logged when it skipped an activity, +// or "" if it logged no skip for it. The reason is not stored on the event row — +// the queue logs it and marks the event processed — so this line is the only +// place the bridge says WHY it decided to do nothing. +// +// target is REDACTED out of the answer, and that is not cosmetic: a skip reason +// carries the id it is about, so two reasons about two different objects differ +// as strings no matter how identical their wording. Comparing them without +// redacting reads "the bridge distinguished these two cases" off nothing but the +// two objects having different names — a test that passes before the behaviour +// it describes exists. +func skipReasonFor(t *testing.T, h *harness, activityID, target string) string { + t.Helper() + for _, line := range strings.Split(h.logs.String(), "\n") { + if !strings.Contains(line, "inbox event skipped") || !strings.Contains(line, activityID) { + continue + } + _, reason, ok := strings.Cut(line, "reason=") + if !ok { + return "" + } + return strings.ReplaceAll(reason, target, "") + } + return "" +} diff --git a/internal/ingest/moderation_terminal_test.go b/internal/ingest/moderation_terminal_test.go index e82e79a..1ac05f4 100644 --- a/internal/ingest/moderation_terminal_test.go +++ b/internal/ingest/moderation_terminal_test.go @@ -60,9 +60,17 @@ const ( mtAuthorDID = "did:plc:mtnativeauthor0001" mtAuthorHandle = "mtauthor.coves.social" - mtPostRKey = "3lzmtpost00001" - mtPostATURI = "at://" + mtAuthorDID + "/social.coves.community.postv2/" + mtPostRKey - mtPostAPID = mtUserOrigin + "/ap/object/" + mtAuthorDID + + // A SECOND native actor, who never authored the post. Bans are + // three-dimensional — (community, actor, content) — so a fixture with one + // actor cannot tell "cancel that actor's deliveries to that community" from + // "cancel every delivery", and one with one community cannot tell "that + // community" from "everywhere". Both are in the world from the start. + mtCommenterDID = "did:plc:mtnativecommnter1" + mtCommenterHandle = "mtcommenter.coves.social" + + mtPostRKey = "3lzmtpost00001" + mtPostATURI = "at://" + mtAuthorDID + "/social.coves.community.postv2/" + mtPostRKey + mtPostAPID = mtUserOrigin + "/ap/object/" + mtAuthorDID + "/social.coves.community.postv2/" + mtPostRKey mtPostCID = "bafyreievgu2ty7qbiaaom5zhmkznsnajuzideek3lo7e65dwqlrvrxnmo4" mtEditCID = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" @@ -182,6 +190,14 @@ func newModerationWorld(t *testing.T, h *harness) moderationWorld { // mtPostEvent builds a postv2 commit frame the consumer accepts. func mtPostEvent(t *testing.T, operation, rev, cid string, timeUS int64) *consume.JetstreamEvent { + t.Helper() + return mtPostEventFor(t, mtPostRKey, operation, rev, cid, timeUS) +} + +// mtPostEventFor is mtPostEvent for an arbitrary record key: a lock is +// per-OBJECT, so telling that apart from per-community needs a second thread in +// the SAME community — which needs a second post to root it. +func mtPostEventFor(t *testing.T, rkey, operation, rev, cid string, timeUS int64) *consume.JetstreamEvent { t.Helper() frame := fmt.Sprintf(`{ "did": %q, "time_us": %d, "kind": "commit", @@ -197,7 +213,7 @@ func mtPostEvent(t *testing.T, operation, rev, cid string, timeUS int64) *consum "createdAt": "2026-08-13T10:00:00.000Z" } } -}`, mtAuthorDID, timeUS, rev, operation, mtPostRKey, cid, testDIDFor(mtCommunityAName, "lemmy.world")) +}`, mtAuthorDID, timeUS, rev, operation, rkey, cid, testDIDFor(mtCommunityAName, "lemmy.world")) var event consume.JetstreamEvent require.NoError(t, json.Unmarshal([]byte(frame), &event), "the frame must be valid wire JSON") return &event @@ -205,8 +221,23 @@ func mtPostEvent(t *testing.T, operation, rev, cid string, timeUS int64) *consum type mtResolver struct{} -func (mtResolver) ResolveDIDHandle(context.Context, string) (string, error) { - return mtAuthorHandle, nil +// mtHandles is per-DID rather than one answer for every DID: the persona's +// local part is DERIVED from the handle and frozen at mint, so a resolver that +// answered the same handle for two native actors would mint two personas +// fighting over one name — and the collision search would quietly hand the +// second one a suffixed identity that no assertion about "that actor" matches. +var mtHandles = map[string]string{ + mtAuthorDID: mtAuthorHandle, + mtCommenterDID: mtCommenterHandle, +} + +func (mtResolver) ResolveDIDHandle(_ context.Context, did string) (string, error) { + handle, ok := mtHandles[did] + if !ok { + // Loud, not a fallback: minting on a guessed handle freezes the guess. + return "", fmt.Errorf("no test handle registered for %s", did) + } + return handle, nil } // admissionFor reads the ledger row for one (community, post). diff --git a/internal/materialize/acceptance.go b/internal/materialize/acceptance.go index 62293e4..699f8e2 100644 --- a/internal/materialize/acceptance.go +++ b/internal/materialize/acceptance.go @@ -99,10 +99,15 @@ func (m *Materializer) removalStands(ctx context.Context, communityDID, rkey str } } -// removalCodeModeratorDiscretion is the removal lexicon's catch-all: Lemmy +// RemovalCodeModeratorDiscretion is the removal lexicon's catch-all: Lemmy // sends no machine-readable code, so anything narrower would be the bridge // asserting a reason the moderator never gave. -const removalCodeModeratorDiscretion = "moderator-discretion" +// +// Exported because the BRIDGE-SIDE removal of a native comment (ingest, task +// 17c-2) records the same decision in a different place: one moderator action +// must not read as two different things depending on whether they removed a +// post or a comment, and two copies of the literal is exactly how that drifts. +const RemovalCodeModeratorDiscretion = "moderator-discretion" // RemovePost records a community's moderator removal of a post: the acceptance // is deleted and a removal written IN ONE COMMIT, at the same digest rkey. @@ -150,7 +155,7 @@ func (m *Materializer) RemovePost(ctx context.Context, mapping *store.APObjectMa // Lemmy sends no machine-readable code, so the open knownValues set's // catch-all applies. Inventing a narrower code (spam, rule-violation) // would be the bridge asserting a reason the moderator never gave. - "code": removalCodeModeratorDiscretion, + "code": RemovalCodeModeratorDiscretion, "createdAt": recordDatetime(m.moderationStamp(ctx, communityDID, CollectionRemoval, rkey)), } // Omitted rather than written blank: Lemmy spells "no reason given" as an @@ -172,7 +177,7 @@ func (m *Materializer) RemovePost(ctx context.Context, mapping *store.APObjectMa m.logger.Info("post removed from community by moderator", "community_did", communityDID, "post", postURI, "ap_id", mapping.APID) m.recordModeration(ctx, mapping, func() error { - return m.ledger.RecordRemoval(ctx, communityDID, postURI, mapping.DID, removalCodeModeratorDiscretion) + return m.ledger.RecordRemoval(ctx, communityDID, postURI, mapping.DID, RemovalCodeModeratorDiscretion) }) return nil } diff --git a/internal/materialize/community.go b/internal/materialize/community.go index 244db69..db46b94 100644 --- a/internal/materialize/community.go +++ b/internal/materialize/community.go @@ -108,6 +108,30 @@ func mappingCommunityDID(collection, did string, record map[string]any, fallback } } +// mappingThreadRootATURI is the thread a record being committed belongs to, for +// its mapping's thread_root_at_uri column. +// +// Only a COMMENT has one to record. A post IS the top of its thread, and every +// reader already treats it that way, so writing its own at-uri back at it would +// add a second spelling of a fact the collection alone already answers. +// +// The value comes from the record's own reply.root — the strongRef the +// materializer just resolved through the thread (resolveReplyRefs), never from +// anything a delivery asserted about itself. +func mappingThreadRootATURI(collection string, record map[string]any, stored string) string { + if stored != "" { + return stored + } + if collection != CollectionComment { + return "" + } + did, rootCollection, rkey := replyRootRef(record) + if did == "" || rootCollection == "" || rkey == "" { + return "" + } + return "at://" + did + "/" + rootCollection + "/" + rkey +} + // commentThreadCommunityDID recovers a comment's community from its thread // root. Which era the root belongs to decides how: a legacy root's repo IS the // community, a postv2 root only NAMES one, so the root's own record has to be diff --git a/internal/materialize/materializer.go b/internal/materialize/materializer.go index 8da73e9..b49e678 100644 --- a/internal/materialize/materializer.go +++ b/internal/materialize/materializer.go @@ -341,6 +341,12 @@ func (m *Materializer) commitRecord(ctx context.Context, did, collection, rkey s // inReplyTo would hand another community moderation authority over content // posted somewhere else. var storedCommunityDID string + // storedThreadRoot is the thread a previous materialization recorded. A + // comment cannot change threads, so a binding already made wins — exactly + // like the community above, and for the same reason: it is read back on the + // moderation path, and re-deriving it from an edited delivery would let an + // edit move a comment out from under its thread's lock. + var storedThreadRoot string if existing, err := m.objects.GetByAPID(ctx, obj.ID); err == nil { if existing.IsDeleted() { return nil, skip(obj.ID, "object was deleted upstream; not resurrecting") @@ -349,6 +355,7 @@ func (m *Materializer) commitRecord(ctx context.Context, did, collection, rkey s collection == CollectionPostV2 || collection == CollectionComment storedCommunityDID = existing.CommunityDID + storedThreadRoot = existing.ThreadRootATURI } else if !errors.IsNotFound(err) { return nil, fmt.Errorf("materialize: check mapping for %s: %w", obj.ID, err) } @@ -377,6 +384,11 @@ func (m *Materializer) commitRecord(ctx context.Context, did, collection, rkey s // restored by then, and the mapping must agree with the record it // maps or the two would authorize different communities. mapping.CommunityDID = mappingCommunityDID(collection, did, record, communityDID, storedCommunityDID) + // Same rule, same moment, for the same reason: the thread a comment + // hangs in is decided once and read off the RECORD being committed, so + // an update whose reply refs were carried forward maps the thread the + // record actually names rather than one this delivery asserted. + mapping.ThreadRootATURI = mappingThreadRootATURI(collection, record, storedThreadRoot) var mapErr error stored, mapErr = m.objects.PutMappingTx(ctx, tx, mapping) if mapErr != nil { diff --git a/internal/store/ap_objects.go b/internal/store/ap_objects.go index e2fa8aa..523786f 100644 --- a/internal/store/ap_objects.go +++ b/internal/store/ap_objects.go @@ -23,7 +23,8 @@ func NewAPObjects(db *sql.DB) APObjects { const apObjectColumns = ` id, ap_id, ap_type, origin_instance, origin, did, author_did, community_did, - collection, rkey, at_uri, cid, ap_published_at, indexed_at, deleted_at` + thread_root_at_uri, collection, rkey, at_uri, cid, ap_published_at, indexed_at, + deleted_at` func (r *postgresAPObjects) PutMapping(ctx context.Context, mapping APObjectMapping) (*APObjectMapping, error) { return r.putMapping(ctx, r.db, mapping) @@ -49,8 +50,9 @@ func (r *postgresAPObjects) putMapping(ctx context.Context, q queryRower, mappin query := ` INSERT INTO ap_objects ( ap_id, ap_type, origin_instance, origin, did, author_did, - community_did, collection, rkey, at_uri, cid, ap_published_at - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + community_did, thread_root_at_uri, collection, rkey, at_uri, cid, + ap_published_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT (ap_id) DO UPDATE SET ap_type = EXCLUDED.ap_type, origin = EXCLUDED.origin, @@ -67,6 +69,13 @@ func (r *postgresAPObjects) putMapping(ctx context.Context, q queryRower, mappin -- write path) would NULL a good binding and make moderation refuse -- forever, silently. A write that HAS the value still wins. community_did = COALESCE(EXCLUDED.community_did, ap_objects.community_did), + -- COALESCE for the same reason, with one difference worth stating: + -- the materializer re-derives this from the record it is committing, + -- so a re-put normally re-supplies it and a row written before + -- migration 026 heals itself. The COALESCE is what stops a write path + -- that does NOT know the thread (any non-materializer mapping write) + -- from blanking one that does. + thread_root_at_uri = COALESCE(EXCLUDED.thread_root_at_uri, ap_objects.thread_root_at_uri), collection = EXCLUDED.collection, rkey = EXCLUDED.rkey, at_uri = EXCLUDED.at_uri, @@ -79,6 +88,7 @@ func (r *postgresAPObjects) putMapping(ctx context.Context, q queryRower, mappin row := q.QueryRowContext(ctx, query, mapping.APID, mapping.APType, mapping.OriginInstance, string(mapping.Origin), mapping.DID, nullIfEmpty(mapping.AuthorDID), nullIfEmpty(mapping.CommunityDID), + nullIfEmpty(mapping.ThreadRootATURI), mapping.Collection, mapping.RKey, mapping.ATURI, mapping.CID, mapping.PublishedAt, ) stored, err := scanAPObject(row) @@ -283,10 +293,10 @@ type rowScanner interface { func scanAPObject(row rowScanner) (*APObjectMapping, error) { var mapping APObjectMapping var origin string - var authorDID, communityDID sql.NullString + var authorDID, communityDID, threadRootATURI sql.NullString err := row.Scan( &mapping.ID, &mapping.APID, &mapping.APType, &mapping.OriginInstance, - &origin, &mapping.DID, &authorDID, &communityDID, + &origin, &mapping.DID, &authorDID, &communityDID, &threadRootATURI, &mapping.Collection, &mapping.RKey, &mapping.ATURI, &mapping.CID, &mapping.PublishedAt, &mapping.IndexedAt, &mapping.DeletedAt, ) @@ -296,6 +306,7 @@ func scanAPObject(row rowScanner) (*APObjectMapping, error) { mapping.Origin = Origin(origin) mapping.AuthorDID = authorDID.String mapping.CommunityDID = communityDID.String + mapping.ThreadRootATURI = threadRootATURI.String return &mapping, nil } diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go index e9e8954..d3cc91a 100644 --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -14,10 +14,66 @@ import ( "time" ) +// ObjectModeration is the moderation state the BRIDGE owns for one bridged +// object — today a community's thread LOCK, which has no home in either repo +// (see migration 025). +// +// It rides the APObjects interface, and the two are deliberately different +// things at different levels: the TABLE is separate, because putMapping +// rewrites a whole ap_objects row and a re-pin would clear a lock; the +// ACCESSOR sits here because every holder of a mapping is exactly the caller +// that needs to ask, and both readers (the announced-moderation dispatch and +// the native comment consumer) already hold one. +type ObjectModeration interface { + // SetLock records or clears a community's lock on an object. Locking an + // already-locked object preserves the ORIGINAL locked_at — a re-announced + // Lock is the same decision, not a new one. Unlocking is scoped to the + // community that holds the lock: clearing is a no-op for anyone else, and a + // no-op success for an object that was never locked. + SetLock(ctx context.Context, object ModeratedObject, locked bool) error + + // LockedAmong returns the first of the given at-uris that currently carries + // a lock, or "" when none does. It takes a SET because the question is + // always asked of a thread — a comment is refused by a lock on its parent + // OR on its thread root — and one statement keeps that one round trip + // however many ancestors it names. Empty at-uris are ignored; an object no + // community has ever moderated simply has no row, which is the answer + // "open" rather than an error. + LockedAmong(ctx context.Context, atURIs ...string) (string, error) + + // SetRemoval records a community's removal of a COMMENT: when it happened, + // under which code, and with the moderator's own reason. Re-recording an + // existing removal keeps the ORIGINAL removed_at — a re-delivered Delete is + // the same decision arriving twice. + // + // COMMENTS ONLY, deliberately. A post's removal is a record in the + // community's own repo, written atomically with the withdrawal of the + // acceptance it replaces; a second copy here would be a second source of + // truth for one decision, and the two would disagree the first time the + // commit succeeded and this write did not. + SetRemoval(ctx context.Context, object ModeratedObject, code, reason string) error + + // ClearRemoval lifts a removal (Undo{Delete}), scoped to the community that + // made it: clearing is a no-op for anyone else, and a no-op success for an + // object nobody removed. The row survives — a lock on the same object is a + // separate decision and is not lifted with it. + ClearRemoval(ctx context.Context, atURI, communityDID string) error + + // CommunityHoldsAnyLock reports whether a community currently holds a lock + // on anything at all. It answers the ONE question left when a comment's + // thread cannot be determined: a community holding no lock cannot have + // locked the thread we failed to name, so there is provably nothing to miss. + // It is never the refusal rule itself — a lock is per-object, and a + // community holding one says nothing about its other threads. + CommunityHoldsAnyLock(ctx context.Context, communityDID string) (bool, error) +} + // APObjects maps AP object ids to the atproto records they materialized // as, and back. Every materialization writes a mapping; every strongRef // resolution reads one. type APObjects interface { + ObjectModeration + // PutMapping idempotently upserts a mapping keyed on APID. It validates // DID, Collection, RKey, and CID, derives ATURI from the first three, // and returns the stored row. An empty Origin defaults to diff --git a/internal/store/models.go b/internal/store/models.go index cc25483..c9fe407 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -106,18 +106,40 @@ type APObjectMapping struct { // through materialize.CommunityDIDOf, never compared directly — a direct // comparison silently treats every pre-016 row as belonging to nobody. CommunityDID string - Collection string // record NSID, e.g. social.coves.community.post - RKey string // deterministic TID rkey - ATURI string // at://did/collection/rkey (derived; set by PutMapping) - CID string // CID of the current record version - PublishedAt *time.Time // AP `published` time (may be absent upstream) - IndexedAt time.Time - DeletedAt *time.Time + // ThreadRootATURI is the at-uri of the thread a materialized COMMENT hangs + // in — its record's reply.root, recorded at materialization time because + // that is the only moment the bridge knows it without re-reading the + // record. Empty for posts (a post IS its own thread root) and for comments + // materialized before migration 026. + // + // It is thread STRUCTURE, not moderation state: immutable for the life of + // the record, and the answer to "which thread is this in?" that a lock on + // the post above a Lemmy comment is read against. + ThreadRootATURI string + Collection string // record NSID, e.g. social.coves.community.post + RKey string // deterministic TID rkey + ATURI string // at://did/collection/rkey (derived; set by PutMapping) + CID string // CID of the current record version + PublishedAt *time.Time // AP `published` time (may be absent upstream) + IndexedAt time.Time + DeletedAt *time.Time } // IsDeleted reports whether the mapping has been soft-deleted. func (m *APObjectMapping) IsDeleted() bool { return m.DeletedAt != nil } +// ModeratedObject identifies the object a moderation decision applies to and +// the community that made it. All three fields travel together because none of +// them is derivable from another here: the at-uri is what the comment consumer +// reads back, the AP id is what the announcing community named, and the +// community DID is the binding without which any co-hosted community could +// lift the decision. +type ModeratedObject struct { + ATURI string + APID string + CommunityDID string +} + // BridgedActor is a fediverse actor (person or group) that Tidepool has // minted an atproto identity for. type BridgedActor struct { diff --git a/internal/store/object_moderation.go b/internal/store/object_moderation.go new file mode 100644 index 0000000..664890b --- /dev/null +++ b/internal/store/object_moderation.go @@ -0,0 +1,179 @@ +package store + +import ( + "context" + "database/sql" + stderrors "errors" + "fmt" + + "github.com/lib/pq" + + "tidepool/internal/errors" +) + +// object_moderation is the bridge-owned moderation state described on the +// ObjectModeration interface and in migration 025. The methods hang off the +// ap_objects repository — the callers that need them all hold one — but the +// ROW is separate, because putMapping rewrites a mapping wholesale and a +// re-pin must never clear a lock. + +func (r *postgresAPObjects) SetLock(ctx context.Context, object ModeratedObject, locked bool) error { + if object.ATURI == "" { + return errors.NewValidationError("at_uri", "must not be empty") + } + if object.CommunityDID == "" { + // The binding is not bookkeeping: an unbound row is one any co-hosted + // community's Undo{Lock} could clear, so a write that cannot name the + // deciding community is refused rather than stored unbound. + return errors.NewValidationError("community_did", "must not be empty") + } + + if !locked { + // Scoped to the holder. Authorization upstream already established that + // this community owns the target's mapping, so this is the same rule + // stated where the row lives — and a lift for an object nobody locked + // updates nothing, which is the right outcome for a re-delivered Undo. + if _, err := r.db.ExecContext(ctx, ` + UPDATE object_moderation + SET locked_at = NULL, updated_at = now() + WHERE at_uri = $1 AND community_did = $2`, + object.ATURI, object.CommunityDID); err != nil { + return fmt.Errorf("clear lock on %q: %w", object.ATURI, err) + } + return nil + } + + if object.APID == "" { + return errors.NewValidationError("ap_id", "must not be empty") + } + if _, err := r.db.ExecContext(ctx, ` + INSERT INTO object_moderation (at_uri, ap_id, community_did, locked_at) + VALUES ($1, $2, $3, now()) + ON CONFLICT (at_uri) DO UPDATE SET + ap_id = EXCLUDED.ap_id, + community_did = EXCLUDED.community_did, + -- COALESCE, never EXCLUDED outright: a re-delivered Announce{Lock} + -- is the SAME decision arriving twice, and re-stamping it would walk + -- the moderators' timestamp forward every time Lemmy retried. + locked_at = COALESCE(object_moderation.locked_at, EXCLUDED.locked_at), + updated_at = now()`, + object.ATURI, object.APID, object.CommunityDID); err != nil { + return fmt.Errorf("record lock on %q: %w", object.ATURI, err) + } + return nil +} + +func (r *postgresAPObjects) SetRemoval(ctx context.Context, object ModeratedObject, code, reason string) error { + if object.ATURI == "" { + return errors.NewValidationError("at_uri", "must not be empty") + } + if object.APID == "" { + return errors.NewValidationError("ap_id", "must not be empty") + } + if object.CommunityDID == "" { + // Same rule as a lock: an unbound decision is one any co-hosted + // community's Undo could lift. + return errors.NewValidationError("community_did", "must not be empty") + } + if code == "" { + // A removal with no code is a decision with no machine-readable why, and + // the admin surface reading this row has nothing else to go on. + return errors.NewValidationError("removal_code", "must not be empty") + } + if _, err := r.db.ExecContext(ctx, ` + INSERT INTO object_moderation ( + at_uri, ap_id, community_did, removed_at, removal_code, removal_reason) + VALUES ($1, $2, $3, now(), $4, $5) + ON CONFLICT (at_uri) DO UPDATE SET + ap_id = EXCLUDED.ap_id, + community_did = EXCLUDED.community_did, + -- COALESCE, as for a lock: a re-delivered Delete is the same removal + -- arriving twice, and re-stamping would walk the moderators' timestamp + -- forward every time Lemmy retried. + removed_at = COALESCE(object_moderation.removed_at, EXCLUDED.removed_at), + removal_code = EXCLUDED.removal_code, + removal_reason = EXCLUDED.removal_reason, + updated_at = now()`, + object.ATURI, object.APID, object.CommunityDID, code, reason); err != nil { + return fmt.Errorf("record removal of %q: %w", object.ATURI, err) + } + return nil +} + +func (r *postgresAPObjects) ClearRemoval(ctx context.Context, atURI, communityDID string) error { + if atURI == "" { + return errors.NewValidationError("at_uri", "must not be empty") + } + if communityDID == "" { + return errors.NewValidationError("community_did", "must not be empty") + } + // The code and the reason go with it: they describe a decision that no + // longer stands, and leaving them behind would let an admin surface read a + // live reason off a lifted removal. + if _, err := r.db.ExecContext(ctx, ` + UPDATE object_moderation + SET removed_at = NULL, removal_code = '', removal_reason = '', updated_at = now() + WHERE at_uri = $1 AND community_did = $2`, + atURI, communityDID); err != nil { + return fmt.Errorf("clear removal of %q: %w", atURI, err) + } + return nil +} + +func (r *postgresAPObjects) LockedAmong(ctx context.Context, atURIs ...string) (string, error) { + // The caller names a thread — a parent and a root, sometimes the same object + // twice — so the empties and duplicates it may hold are filtered here rather + // than at every call site. + candidates := make([]string, 0, len(atURIs)) + for _, atURI := range atURIs { + if atURI != "" && !containsString(candidates, atURI) { + candidates = append(candidates, atURI) + } + } + if len(candidates) == 0 { + return "", nil + } + var locked string + err := r.db.QueryRowContext(ctx, ` + SELECT at_uri FROM object_moderation + WHERE at_uri = ANY($1) AND locked_at IS NOT NULL + LIMIT 1`, pq.Array(candidates)).Scan(&locked) + if stderrors.Is(err, sql.ErrNoRows) { + // No row at all: nobody has ever moderated any of these objects, which is + // true of almost every object. Not a NotFound error — the question asked + // is "is anything here locked", and "no" is a complete answer to it. + return "", nil + } + if err != nil { + return "", fmt.Errorf("read lock state for %v: %w", candidates, err) + } + return locked, nil +} + +func (r *postgresAPObjects) CommunityHoldsAnyLock(ctx context.Context, communityDID string) (bool, error) { + if communityDID == "" { + return false, errors.NewValidationError("community_did", "must not be empty") + } + var held bool + // EXISTS, not a count: the caller only asks whether there is anything at all + // that could have locked a thread it could not name. + if err := r.db.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM object_moderation + WHERE community_did = $1 AND locked_at IS NOT NULL)`, + communityDID).Scan(&held); err != nil { + return false, fmt.Errorf("read standing locks of %q: %w", communityDID, err) + } + return held, nil +} + +// containsString reports whether the slice already holds s. The candidate sets +// here are two or three elements, so a scan beats building a map. +func containsString(values []string, s string) bool { + for _, v := range values { + if v == s { + return true + } + } + return false +} -- 2.51.2