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 +}