diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -671,16 +671,18 @@ // community is admitted here, the community-signed acceptance is written, and // the Create{Page} enqueued atomically with it. Wired whenever the consumer // runs, so postv2 events are admitted rather than skipped at debug. engine, err := accept.NewEngine(accept.Options{ - Repos: repoManager, - Enqueuer: enqueuer, - Actors: minter, - Resolver: resolver, - Communities: store.NewCommunities(database), - Objects: store.NewOutboundObjects(database), - Prefs: store.NewFederationPrefs(database), - Admissions: accept.NewAdmissions(database), - UserOrigin: cfg.APUserOrigin, - Logger: logger, + Repos: repoManager, + Enqueuer: enqueuer, + Actors: minter, + Resolver: resolver, + Communities: store.NewCommunities(database), + Objects: store.NewOutboundObjects(database), + Prefs: store.NewFederationPrefs(database), + Admissions: accept.NewAdmissions(database), + APActors: store.NewAPActors(database), + MaxPerAuthorPerCommunity: cfg.AdmissionMaxPerAuthorPerCommunity, + UserOrigin: cfg.APUserOrigin, + Logger: logger, }) if err != nil { return nil, fmt.Errorf("consumer: acceptance engine: %w", err) diff --git a/internal/accept/admissions.go b/internal/accept/admissions.go --- a/internal/accept/admissions.go +++ b/internal/accept/admissions.go @@ -26,6 +26,7 @@ // firehose-visible acceptance/removal records). type Admission struct { CommunityDID string PostURI string + AuthorDID string Status string DecisionCode string EvaluatedCID string @@ -70,10 +71,11 @@ return errors.NewValidationError("admission.status", "must be set") } _, err := ex.ExecContext(ctx, ` INSERT INTO admissions - (community_did, post_uri, status, decision_code, evaluated_cid, + (community_did, post_uri, author_did, status, decision_code, evaluated_cid, acceptance_rkey, accepted_cid, redrivable) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (community_did, post_uri) DO UPDATE SET + author_did = EXCLUDED.author_did, status = EXCLUDED.status, decision_code = EXCLUDED.decision_code, evaluated_cid = EXCLUDED.evaluated_cid, @@ -81,7 +83,7 @@ acceptance_rkey = EXCLUDED.acceptance_rkey, accepted_cid = EXCLUDED.accepted_cid, redrivable = EXCLUDED.redrivable, updated_at = now()`, - adm.CommunityDID, adm.PostURI, adm.Status, adm.DecisionCode, adm.EvaluatedCID, + adm.CommunityDID, adm.PostURI, adm.AuthorDID, adm.Status, adm.DecisionCode, adm.EvaluatedCID, adm.AcceptanceRKey, adm.AcceptedCID, adm.Redrivable) if err != nil { return fmt.Errorf("accept: record admission %s/%s: %w", adm.CommunityDID, adm.PostURI, err) @@ -94,11 +96,11 @@ // errors.IsNotFound when the engine has never decided on it. func (a *Admissions) Get(ctx context.Context, communityDID, postURI string) (*Admission, error) { var adm Admission err := a.db.QueryRowContext(ctx, ` - SELECT community_did, post_uri, status, decision_code, evaluated_cid, + SELECT community_did, post_uri, author_did, status, decision_code, evaluated_cid, acceptance_rkey, accepted_cid, redrivable FROM admissions WHERE community_did = $1 AND post_uri = $2`, communityDID, postURI).Scan( - &adm.CommunityDID, &adm.PostURI, &adm.Status, &adm.DecisionCode, &adm.EvaluatedCID, + &adm.CommunityDID, &adm.PostURI, &adm.AuthorDID, &adm.Status, &adm.DecisionCode, &adm.EvaluatedCID, &adm.AcceptanceRKey, &adm.AcceptedCID, &adm.Redrivable) if stderrors.Is(err, sql.ErrNoRows) { return nil, errors.NewNotFoundError("admission", communityDID+"/"+postURI) @@ -108,3 +110,38 @@ return nil, fmt.Errorf("accept: get admission %s/%s: %w", communityDID, postURI, err) } return &adm, nil } + +// CountAccepted reports how many posts one author currently has ACCEPTED in one +// community, excluding one post_uri (the post being decided — a repin must not +// count against its own author). It backs the per-author-per-community flood +// cap. The (author_did, community_did, created_at) index serves it index-only. +func (a *Admissions) CountAccepted(ctx context.Context, authorDID, communityDID, excludePostURI string) (int, error) { + var n int + err := a.db.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM admissions + WHERE author_did = $1 AND community_did = $2 AND status = $3 AND post_uri <> $4`, + authorDID, communityDID, StatusAccepted, excludePostURI).Scan(&n) + if err != nil { + return 0, fmt.Errorf("accept: count accepted for %s in %s: %w", authorDID, communityDID, err) + } + return n, nil +} + +// DeleteTx removes the ledger row for a (community, post) on an existing +// transaction — the author-delete path, which rides the acceptance-delete commit +// so the ledger row and the acceptance record go away together. The post no +// longer exists to re-decide, and no removal record stands to explain a +// 'removed' status, so the row is dropped rather than left behind. Deleting a +// missing row is a no-op success (a redelivered delete). A nil tx is a +// validation error. +func (a *Admissions) DeleteTx(ctx context.Context, tx *sql.Tx, communityDID, postURI string) error { + if tx == nil { + return errors.NewValidationError("tx", "must not be nil") + } + if _, err := tx.ExecContext(ctx, + `DELETE FROM admissions WHERE community_did = $1 AND post_uri = $2`, + communityDID, postURI); err != nil { + return fmt.Errorf("accept: delete admission %s/%s: %w", communityDID, postURI, err) + } + return nil +} diff --git a/internal/accept/engine.go b/internal/accept/engine.go --- a/internal/accept/engine.go +++ b/internal/accept/engine.go @@ -21,11 +21,15 @@ "fmt" "log/slog" "time" + "github.com/bluesky-social/indigo/atproto/atdata" + "github.com/bluesky-social/indigo/atproto/lexicon" + "tidepool/internal/acceptrec" "tidepool/internal/consume" "tidepool/internal/errors" "tidepool/internal/repo" "tidepool/internal/store" + "tidepool/lexicons" ) // Decision codes recorded on a rejected/removed admission (migration 021's @@ -40,8 +44,39 @@ // rejects a titleless post, and no title-derivation product decision exists. DecisionTitleRequired = "title-required" // DecisionTitleTooLong: over Lemmy's 200-char title cap. DecisionTitleTooLong = "title-too-long" + // DecisionPaused: the author's account is #account-paused (decision 19) — + // deactivated/suspended/takendown/throttled. Delivery is halted, so a new + // post is not admitted while the identity is paused (it can be re-driven). + DecisionPaused = "paused" + // DecisionRateLimit: the author exceeded the per-author-per-community accept + // cap (Tidepool must not let one native account flood a Lemmy community it + // vouches for). + DecisionRateLimit = "rate-limit-exceeded" + // DecisionLexiconInvalid: the postv2 failed strict lexicon validation. WE + // sign the acceptance, so native input that does not validate is fail-closed. + DecisionLexiconInvalid = "lexicon-invalid" + // DecisionCommunityImmutable: an UPDATE tried to MOVE the post to a different + // community than the one it was accepted into. The lexicon makes `community` + // immutable; the whole event is discarded. (Recorded on the ORIGINAL + // community's admissions row as a no-op annotation, if at all — the engine + // writes nothing to the target community.) + DecisionCommunityImmutable = "community-immutable" ) +// RemovalCodeAdmissionRevoked is the removal `code` written when a post that WAS +// accepted fails RE-admission (an edit made it titleless or over the cap). +// +// PROPOSED — FLAGGED FOR COORDINATOR RULING. The removal lexicon's knownValues +// (rule-violation, spam, off-topic, illegal-content, author-banned, +// moderator-discretion) are ALL moderation reasons, and an admission revocation +// is not a moderator's decision — writing moderator-discretion would assert a +// moderator acted when none did. knownValues is explicitly an OPEN set (peers +// may send unseen codes and they must still validate), so a precise new code is +// legal. The SPECIFIC cause (title-required / title-too-long) is recorded in the +// admissions ledger's decision_code; this open-set code is the firehose-visible +// one. Alternative if the coordinator prefers a knownValue: "rule-violation". +const RemovalCodeAdmissionRevoked = "admission-revoked" + // lemmyTitleCap is Lemmy 0.19.20's post-title length limit. const lemmyTitleCap = 200 @@ -70,6 +105,15 @@ // and is RECORDED as a rejection rather than silently dropped upstream. Prefs store.FederationPrefs // Admissions is the decision ledger (migration 021). Admissions *Admissions + // APActors reads the author's AP actor row for the delivery-paused admission + // check (decision 19). OPTIONAL: nil skips the paused check (the seam is not + // wired yet — flagged for the paused-rejection lifecycle). + APActors store.APActors + // MaxPerAuthorPerCommunity caps accepted posts by one author in one community + // within the ledger (the per-author-per-community flood guard, + // ADMISSION_MAX_PER_AUTHOR_PER_COMMUNITY). 0 means UNLIMITED (the generous + // default); a positive value is the cap the rate check enforces. + MaxPerAuthorPerCommunity int // UserOrigin is AP_USER_ORIGIN: the origin every deterministic activity id // is minted under. UserOrigin string @@ -79,16 +123,19 @@ } // Engine admits native posts into bridged communities. type Engine struct { - repos acceptrec.RepoManager - enqueuer consume.OutboundEnqueuer - actors consume.ActorMinter - resolver consume.DIDResolver - communities store.Communities - objects store.OutboundObjects - prefs store.FederationPrefs - admissions *Admissions - userOrigin string - logger *slog.Logger + repos acceptrec.RepoManager + enqueuer consume.OutboundEnqueuer + actors consume.ActorMinter + resolver consume.DIDResolver + communities store.Communities + objects store.OutboundObjects + prefs store.FederationPrefs + admissions *Admissions + apActors store.APActors + maxPerCommunity int + catalog *lexicon.BaseCatalog + userOrigin string + logger *slog.Logger } // The engine is the task 16 acceptance seam the dispatcher hands postv2 commits @@ -121,17 +168,27 @@ logger := opts.Logger if logger == nil { logger = slog.Default() } + // The vendored lexicon catalog validates native postv2 input strictly: WE + // sign the acceptance, so input that does not validate is fail-closed. Loaded + // once at construction — a broken vendored file fails startup, not admission. + catalog, err := lexicons.Catalog() + if err != nil { + return nil, fmt.Errorf("accept: load lexicon catalog: %w", err) + } return &Engine{ - repos: opts.Repos, - enqueuer: opts.Enqueuer, - actors: opts.Actors, - resolver: opts.Resolver, - communities: opts.Communities, - objects: opts.Objects, - prefs: opts.Prefs, - admissions: opts.Admissions, - userOrigin: opts.UserOrigin, - logger: logger, + repos: opts.Repos, + enqueuer: opts.Enqueuer, + actors: opts.Actors, + resolver: opts.Resolver, + communities: opts.Communities, + objects: opts.Objects, + prefs: opts.Prefs, + admissions: opts.Admissions, + apActors: opts.APActors, + maxPerCommunity: opts.MaxPerAuthorPerCommunity, + catalog: catalog, + userOrigin: opts.UserOrigin, + logger: logger, }, nil } @@ -140,15 +197,15 @@ // community acceptance record while enqueueing the Page delivery atomically with // it. create/update run admission on the event's content; delete takes the // acceptance down and enqueues Delete{Page} from stored state. func (e *Engine) AdmitPost(ctx context.Context, did string, commit *consume.CommitEvent) error { - // This cycle handles create/update admission. A delete (author retraction) - // takes the acceptance down and enqueues Delete{Page} from stored state; - // that lifecycle lands next cycle. A delete carries no record body, so there - // is nothing to admit or reject here yet. + postURI := fmt.Sprintf("at://%s/%s/%s", did, commit.Collection, commit.RKey) + + // An author-delete carries no record body: the retraction is built entirely + // from stored outbound state, and it is NOT moderation, so it takes the + // acceptance down WITHOUT a removal record. if commit.Operation == operationDelete { - return nil + return e.authorDelete(ctx, did, postURI) } - postURI := fmt.Sprintf("at://%s/%s/%s", did, commit.Collection, commit.RKey) communityDID, _ := commit.Record["community"].(string) if communityDID == "" { // The consumer already refuses a postv2 with no community, but the engine @@ -156,26 +213,135 @@ // re-asserts it: WE sign the acceptance, so a missing target is fail-closed. return errors.NewValidationError("community", "postv2 "+commit.RKey+" names no community") } - // The opt-out check MOVED here from the consumer: an opted-out author's post - // REACHES the engine and is RECORDED as a rejection with a distinct - // machine-readable reason — no acceptance written, nothing enqueued. Content - // authored while opted out never federates (decision 11). - federating, err := e.mayFederate(ctx, did) + // The post's current binding: whether it was already accepted (so a now-failing + // re-admission is a REMOVAL, not a fresh rejection) and which community it is + // bound to (so a community-moving edit is discarded whole). The engine writes + // outbound_objects ONLY on accept, so a row here means the post federated. + prior, priorBound, err := e.priorBinding(ctx, postURI) if err != nil { return err } - if !federating { - e.logger.Debug("rejecting postv2 from an opted-out author", - slog.String("did", did), slog.String("post", postURI)) + + // Decide admission. The order is deliberate (fail closed first, cheap policy + // last): lexicon-validate → community-immutable → opt-out → paused → title → + // rate cap. A discard means the whole event is dropped (nothing written to + // either community); a non-empty code is a rejection/removal cause. + code, discard, err := e.decide(ctx, did, commit, communityDID, prior, priorBound) + if err != nil { + return err + } + if discard { + e.logger.Debug("discarding community-moving edit", + slog.String("did", did), slog.String("post", postURI), + slog.String("bound_community", prior.CommunityDID), slog.String("event_community", communityDID)) + return nil + } + + if code != "" { + // A post that WAS accepted and now fails re-admission is REMOVED (it + // federated once, so leaving it alone would strand it live on Lemmy); one + // that was never accepted is simply a recorded rejection. + priorAccepted := priorBound && !prior.IsTombstoned() + if priorAccepted { + return e.removeAccepted(ctx, did, communityDID, postURI, commit, prior, code) + } + e.logger.Debug("rejecting postv2", + slog.String("did", did), slog.String("post", postURI), slog.String("reason", code)) return e.admissions.Record(ctx, Admission{ + AuthorDID: did, CommunityDID: communityDID, PostURI: postURI, Status: StatusRejected, - DecisionCode: DecisionOptedOut, + DecisionCode: code, EvaluatedCID: commit.CID, }) } + return e.accept(ctx, did, communityDID, postURI, commit) +} + +// operationDelete is the Jetstream commit operation for a record deletion. +const operationDelete = "delete" + +// decide runs the admission checks in order and returns the rejection/removal +// code ("" = admit), or discard=true when the event must be dropped whole (a +// community-moving edit). The order fails closed first: garbage input never +// reaches a policy check, and a hijack (community move) is refused before the +// author's own preferences are consulted. +func (e *Engine) decide(ctx context.Context, did string, commit *consume.CommitEvent, communityDID string, + prior *store.OutboundObject, priorBound bool) (code string, discard bool, err error) { + + // 1. Strict lexicon validation of the native input — fail closed. + if !e.lexiconValid(commit.Record) { + return DecisionLexiconInvalid, false, nil + } + + // 2. Community immutability. The lexicon marks `community` immutable: an + // UPDATE that names a different community than the post was accepted into is + // a retarget, which means writing a NEW post — so the whole event is + // discarded, not partially applied. + if priorBound && prior.CommunityDID != communityDID { + return "", true, nil + } + + // 3. Opt-out (decision 11): content pushed outward is exactly what an + // opted-out author refused. + federating, err := e.mayFederate(ctx, did) + if err != nil { + return "", false, err + } + if !federating { + return DecisionOptedOut, false, nil + } + + // 4. Paused (#account, decision 19): delivery is halted while the identity is + // deactivated/suspended/takendown/throttled, so a new post is not admitted. + if e.apActors != nil { + actor, err := e.apActors.GetByDID(ctx, did) + switch { + case err == nil: + if actor.DeliveryPaused { + return DecisionPaused, false, nil + } + case errors.IsNotFound(err): + // No actor yet: an unseen author is not paused (it is minted on admit). + default: + return "", false, fmt.Errorf("accept: read actor for %s: %w", did, err) + } + } + + // 5. Title: required and within Lemmy's cap (postv2 title is OPTIONAL in the + // lexicon, so this is admission policy, not validation). + title, _ := commit.Record["title"].(string) + if title == "" { + return DecisionTitleRequired, false, nil + } + if len(title) > lemmyTitleCap { + return DecisionTitleTooLong, false, nil + } + + // 6. Rate cap: one author must not flood a community Tidepool vouches for. + // Counts the author's currently-accepted posts in this community, excluding + // this post so a repin never counts against itself. 0 means unlimited. + if e.maxPerCommunity > 0 { + postURI := fmt.Sprintf("at://%s/%s/%s", did, commit.Collection, commit.RKey) + n, err := e.admissions.CountAccepted(ctx, did, communityDID, postURI) + if err != nil { + return "", false, err + } + if n >= e.maxPerCommunity { + return DecisionRateLimit, false, nil + } + } + + return "", false, nil +} + +// accept writes the community-signed acceptance and enqueues the Create/Update +// {Page} atomically with it (both ride ONE acceptrec commit via its side +// effect). A repin (UPDATE with a new CID) re-pins the same digest rkey and +// enqueues Update{Page}; a fresh create enqueues Create{Page}. +func (e *Engine) accept(ctx context.Context, did, communityDID, postURI string, commit *consume.CommitEvent) error { // The community's AP Group id is the Page's addressing target; the lookup // also re-confirms the community is one we federate. community, err := e.communities.GetByDID(ctx, communityDID) @@ -239,6 +405,7 @@ if err := e.enqueuer.EnqueueActivity(sctx, tx, did, did, "", intent); err != nil { return err } return e.admissions.RecordTx(sctx, tx, Admission{ + AuthorDID: did, CommunityDID: communityDID, PostURI: postURI, Status: StatusAccepted, @@ -255,8 +422,129 @@ } return nil } -// operationDelete is the Jetstream commit operation for a record deletion. -const operationDelete = "delete" +// removeAccepted withdraws a post that WAS accepted and now fails re-admission: +// the acceptance is deleted and a removal (code admission-revoked) written in ONE +// commit, carrying the Delete{Page} enqueue as the side effect. The admissions +// ledger records the SPECIFIC cause (title-required, …) even though the +// firehose-visible removal code is the open-set admission-revoked one. The +// author's post record is untouched — a community removal says where the post may +// appear, not whether it exists. +func (e *Engine) removeAccepted(ctx context.Context, did, communityDID, postURI string, + commit *consume.CommitEvent, prior *store.OutboundObject, code string) error { + + sideEffect := func(sctx context.Context, tx *sql.Tx, _ *repo.CommitResult) error { + // Tombstone the outbound state (the post is out) and build the Delete + // {Page} from it — the retraction carries no body of its own. + dead, err := e.objects.TombstoneTx(sctx, tx, postURI) + if err != nil { + return fmt.Errorf("accept: tombstone outbound state for %s: %w", postURI, err) + } + intent := consume.PostIntent{ + Op: operationDelete, + ATURI: postURI, + ID: consume.ActivityID(e.userOrigin, postURI, operationDelete, dead.LastActivitySeq), + CommunityAPID: dead.CommunityAPID, + Snapshot: dead.TranslatedSnapshot, + } + if err := e.enqueuer.EnqueueActivity(sctx, tx, did, did, "", intent); err != nil { + return err + } + return e.admissions.RecordTx(sctx, tx, Admission{ + AuthorDID: did, + CommunityDID: communityDID, + PostURI: postURI, + Status: StatusRemoved, + DecisionCode: code, + EvaluatedCID: commit.CID, + }) + } + + // The removal pins the version that was accepted when it was removed (audit + // metadata); the code is the open-set admission-revoked, not a moderation + // reason — no moderator acted. + if _, err := acceptrec.Remove(ctx, e.repos, communityDID, postURI, prior.LastCID, + RemovalCodeAdmissionRevoked, "", publishedAtOf(commit.Record), sideEffect); err != nil { + return fmt.Errorf("accept: remove %s from %s: %w", postURI, communityDID, err) + } + return nil +} + +// authorDelete takes an accepted post's acceptance down when its author deletes +// the postv2. Author deletion is NOT moderation, so NO removal record is written +// — the acceptance just goes away — and the Delete{Page} is built from stored +// outbound state (the delete commit carries no body). The ledger row is DELETED: +// the decided post is gone and no removal record stands to explain a 'removed' +// status. All of it rides ONE acceptrec commit, so it is atomic and idempotent +// under replay. +func (e *Engine) authorDelete(ctx context.Context, did, postURI string) error { + stored, err := e.objects.GetByATURI(ctx, postURI) + if errors.IsNotFound(err) { + // A post this bridge never accepted: nothing to withdraw. + e.logger.Debug("author delete for a post with no outbound state", + slog.String("did", did), slog.String("post", postURI)) + return nil + } + if err != nil { + return fmt.Errorf("accept: read outbound state for %s: %w", postURI, err) + } + communityDID := stored.CommunityDID + + sideEffect := func(sctx context.Context, tx *sql.Tx, _ *repo.CommitResult) error { + dead, err := e.objects.TombstoneTx(sctx, tx, postURI) + if err != nil { + return fmt.Errorf("accept: tombstone outbound state for %s: %w", postURI, err) + } + intent := consume.PostIntent{ + Op: operationDelete, + ATURI: postURI, + ID: consume.ActivityID(e.userOrigin, postURI, operationDelete, dead.LastActivitySeq), + CommunityAPID: dead.CommunityAPID, + Snapshot: dead.TranslatedSnapshot, + } + if err := e.enqueuer.EnqueueActivity(sctx, tx, did, did, "", intent); err != nil { + return err + } + return e.admissions.DeleteTx(sctx, tx, communityDID, postURI) + } + + if _, err := acceptrec.DeleteAcceptance(ctx, e.repos, communityDID, postURI, sideEffect); err != nil { + return fmt.Errorf("accept: author-delete %s from %s: %w", postURI, communityDID, err) + } + return nil +} + +// priorBinding reads the post's outbound state — present only after an accept — +// so the engine knows whether it federated and which community it is bound to. +// A miss is reported as (nil, false, nil): a post the bridge has not accepted. +func (e *Engine) priorBinding(ctx context.Context, postURI string) (*store.OutboundObject, bool, error) { + stored, err := e.objects.GetByATURI(ctx, postURI) + if errors.IsNotFound(err) { + return nil, false, nil + } + if err != nil { + return nil, false, fmt.Errorf("accept: read outbound state for %s: %w", postURI, err) + } + return stored, true, nil +} + +// lexiconValid reports whether the postv2 record passes strict validation against +// the vendored lexicon catalog. Invalid input is fail-closed (rejected), never +// signed. A record with no $type, or one whose $type has no schema, is invalid. +func (e *Engine) lexiconValid(record map[string]any) bool { + recordType, _ := record["$type"].(string) + if recordType == "" { + return false + } + raw, err := json.Marshal(record) + if err != nil { + return false + } + data, err := atdata.UnmarshalJSON(raw) + if err != nil { + return false + } + return lexicon.ValidateRecord(e.catalog, data, recordType, lexicon.ValidateFlags(0)) == nil +} // mayFederate reports whether the author permits outbound federation. A missing // preference MEANS default-on (decision 11), not unknown. diff --git a/internal/accept/lifecycle_test.go b/internal/accept/lifecycle_test.go new file mode 100644 --- /dev/null +++ b/internal/accept/lifecycle_test.go @@ -0,0 +1,536 @@ +package accept + +import ( + "context" + "database/sql" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/acceptrec" + "tidepool/internal/consume" + "tidepool/internal/errors" + "tidepool/internal/repo" + "tidepool/internal/store" +) + +// Round 2: the edit / author-delete lifecycle and the rest of admission policy. +// These build on the round-1 harness in outer_acceptance_test.go (acceptanceDB, +// newRepos, realEnqueuer, wireDispatcher, seedBridgedCommunity, the ac* world, +// and the query helpers). + +// A second bridged community + a repinned CID the edit lifecycle needs. +const ( + acCommunityB_DID = "did:plc:z72i7hdynmk6r22z27h6tvur" + acCommunityB_APID = "https://lemmy.world/c/science" + acCommunityB_Name = "science" + acCommunityB_Inbox = "https://lemmy.world/c/science/inbox" + + acPostCID2 = "bafyreib2rxk3rybk3aobmv5cjuql3bm2twh4jo5uxgf5kpqrsqxi3jgxte" + + acRevCreate = acPostRev + acRevUpdate = "3lzpostrev002" + acRevDelete = "3lzpostrev003" +) + +// --------------------------------------------------------------------------- +// Round-2 harness +// --------------------------------------------------------------------------- + +// engineWith builds an Engine with the round-1 seams plus the round-2 additions +// (APActors for the paused check, a mutable Options for the rate cap), letting +// each test override one field. +func engineWith(t *testing.T, conn *sql.DB, repos *repo.Manager, enqueuer consume.OutboundEnqueuer, mutate ...func(*Options)) *Engine { + t.Helper() + opts := Options{ + Repos: repos, + Enqueuer: enqueuer, + Actors: newMinter(t, conn), + Resolver: stubResolver{}, + Communities: store.NewCommunities(conn), + Objects: store.NewOutboundObjects(conn), + Prefs: store.NewFederationPrefs(conn), + Admissions: NewAdmissions(conn), + APActors: store.NewAPActors(conn), + UserOrigin: acUserOrigin, + } + for _, m := range mutate { + m(&opts) + } + engine, err := NewEngine(opts) + require.NoError(t, err) + return engine +} + +// pv2Record builds a valid postv2 record; the mutators shape the invalid / +// variant cases (titleless, over-cap, moved community, malformed createdAt). +func pv2Record(mutate ...func(map[string]any)) map[string]any { + r := map[string]any{ + "$type": "social.coves.community.postv2", + "community": acCommunityDID, + "title": "hello from atproto", + "content": "the body of the post", + "createdAt": "2026-08-12T10:00:00.000Z", + } + for _, m := range mutate { + m(r) + } + return r +} + +// postEvent assembles a postv2 commit event. A delete carries no record and no +// CID (that absence is the whole reason outbound_objects exists). +func postEvent(op, rkey, rev, cid string, timeUS int64, record map[string]any) *consume.JetstreamEvent { + return &consume.JetstreamEvent{ + DID: acAuthorDID, + TimeUS: timeUS, + Kind: "commit", + Commit: &consume.CommitEvent{ + Rev: rev, + Operation: op, + Collection: "social.coves.community.postv2", + RKey: rkey, + CID: cid, + Record: record, + }, + } +} + +func seedBridgedCommunityB(t *testing.T, conn *sql.DB) { + t.Helper() + _, err := store.NewCommunities(conn).UpsertCommunity(context.Background(), store.Community{ + APGroupID: acCommunityB_APID, + DID: acCommunityB_DID, + PreferredUsername: acCommunityB_Name, + Instance: acCommunityHost, + FollowState: store.FollowStateAccepted, + }) + require.NoError(t, err, "seed second bridged community") +} + +// seedPausedActor inserts an ap_actors row for the author with delivery_paused +// set: the paused admission check reads exactly this. +func seedPausedActor(t *testing.T, conn *sql.DB, paused bool) { + t.Helper() + _, err := conn.ExecContext(context.Background(), ` + INSERT INTO ap_actors (did, kind, actor_id, normalized_origin, local_part, + rsa_key_sealed, rsa_key_version, public_key_pem, delivery_paused) + VALUES ($1, 'person', $2, 'coves.social', 'author', '\x00'::bytea, 1, 'pem', $3)`, + acAuthorDID, acUserOrigin+"/ap/actor/"+acAuthorDID, paused) + require.NoError(t, err, "seed ap_actors row for the author") +} + +func activityKindCount(t *testing.T, conn *sql.DB, kind string) int { + t.Helper() + var n int + require.NoError(t, conn.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM outbound_activities WHERE kind = $1`, kind).Scan(&n)) + return n +} + +// acceptanceSubjectCID returns the CID the acceptance at postURI's rkey pins, or +// "" (with found=false) when no acceptance stands. +func acceptanceSubjectCID(t *testing.T, repos *repo.Manager, communityDID, postURI string) (string, bool) { + t.Helper() + rec, _, err := repos.GetRecord(context.Background(), communityDID, acceptrec.CollectionAcceptance, acceptrec.SubjectRKey(postURI)) + if errors.IsNotFound(err) { + return "", false + } + require.NoError(t, err) + subject, _ := rec["subject"].(map[string]any) + cid, _ := subject["cid"].(string) + return cid, true +} + +func removalStandsAt(t *testing.T, repos *repo.Manager, communityDID, postURI string) (map[string]any, bool) { + t.Helper() + rec, _, err := repos.GetRecord(context.Background(), communityDID, acceptrec.CollectionRemoval, acceptrec.SubjectRKey(postURI)) + if errors.IsNotFound(err) { + return nil, false + } + require.NoError(t, err) + return rec, true +} + +// admissionRowCount reports how many admissions rows exist for a post. +func admissionRowCount(t *testing.T, conn *sql.DB, communityDID, postURI string) int { + t.Helper() + var n int + require.NoError(t, conn.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM admissions WHERE community_did = $1 AND post_uri = $2`, + communityDID, postURI).Scan(&n)) + return n +} + +// admittedCreate runs a fresh accepted create and returns the ctx for reuse. +func admittedCreate(t *testing.T, dispatcher *consume.Dispatcher) { + t.Helper() + require.NoError(t, dispatcher.HandleEvent(context.Background(), + postEvent("create", acPostRKey, acRevCreate, acPostCID, acPostTimeUS, pv2Record()))) +} + +// --------------------------------------------------------------------------- +// E1 — edit repin: UPDATE with a new CID re-pins the acceptance + Update{Page} +// --------------------------------------------------------------------------- + +func TestEditRepinsAcceptanceAndEnqueuesUpdatePage(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + admittedCreate(t, dispatcher) + cid, ok := acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + require.True(t, ok) + require.Equal(t, acPostCID, cid, "precondition: the create pinned CID v1") + + // The edit: same post, NEW CID, same community, still a valid title. + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("update", acPostRKey, acRevUpdate, acPostCID2, acPostTimeUS+1, pv2Record()))) + + // The acceptance now pins the NEW CID at the SAME digest rkey. + cid, ok = acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + require.True(t, ok, "the acceptance must still stand after an edit") + assert.Equal(t, acPostCID2, cid, + "an edit re-pins the acceptance to the new CID at the same digest rkey (a repin, not a new record)") + + // Exactly one Update{Page}, under the update activity id with the bumped seq. + assert.Equal(t, 1, activityKindCount(t, conn, "Update"), + "an edit that still passes admission enqueues exactly one Update{Page}") + wantUpdateID := consume.ActivityID(acUserOrigin, acPostURI, "update", 1) + var n int + require.NoError(t, conn.QueryRowContext(ctx, + `SELECT COUNT(*) FROM outbound_activities WHERE activity_id = $1 AND kind = 'Update'`, wantUpdateID).Scan(&n)) + assert.Equal(t, 1, n, "the Update{Page} id is ActivityID(origin, postURI, update, seq=1)") + + // The outbound snapshot moved to the new version. + stored, err := store.NewOutboundObjects(conn).GetByATURI(ctx, acPostURI) + require.NoError(t, err) + assert.Equal(t, acPostCID2, stored.LastCID, "the outbound_objects snapshot follows the edit") + + // The ledger stays accepted, now against the new CID. + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusAccepted, status) + assert.Empty(t, code) + var evaluated, accepted string + require.NoError(t, conn.QueryRowContext(ctx, + `SELECT evaluated_cid, accepted_cid FROM admissions WHERE community_did = $1 AND post_uri = $2`, + acCommunityDID, acPostURI).Scan(&evaluated, &accepted)) + assert.Equal(t, acPostCID2, evaluated, "the ledger records the edited CID it evaluated") + assert.Equal(t, acPostCID2, accepted, "and the CID it re-accepted") + + // Replaying the edit enqueues nothing new (rev gate + idempotent id). + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("update", acPostRKey, acRevUpdate, acPostCID2, acPostTimeUS+1, pv2Record()))) + assert.Equal(t, 1, activityKindCount(t, conn, "Update"), "a replayed edit is a no-op") +} + +// --------------------------------------------------------------------------- +// E2 — edit fails admission: an accepted post edited titleless is REMOVED +// --------------------------------------------------------------------------- + +func TestEditThatFailsAdmissionRemovesAndEnqueuesDelete(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + admittedCreate(t, dispatcher) + _, ok := acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + require.True(t, ok, "precondition: the post was accepted") + + // The edit strips the title — Lemmy rejects a titleless post, so a post that + // WAS accepted and is edited to fail admission is REMOVED (not merely left + // alone): acceptance deleted + removal written + Delete{Page} enqueued. + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("update", acPostRKey, acRevUpdate, acPostCID2, acPostTimeUS+1, + pv2Record(func(r map[string]any) { delete(r, "title") }))), + "an admission-failing edit is a decided outcome (removal), not an event that retries forever") + + _, ok = acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + assert.False(t, ok, "the acceptance must be gone: the edited post no longer qualifies") + + removal, ok := removalStandsAt(t, repos, acCommunityDID, acPostURI) + require.True(t, ok, "a removal must stand at the digest rkey (atomic with the acceptance delete)") + assert.Equal(t, RemovalCodeAdmissionRevoked, removal["code"], + "the removal carries the admission-revoked code (PROPOSED — see the constant's ruling flag)") + subject, _ := removal["subject"].(map[string]any) + assert.Equal(t, acPostURI, subject["uri"], "the removal pins the post at-uri") + + assert.Equal(t, 1, activityKindCount(t, conn, "Delete"), + "a removed post's Delete{Page} withdraws it from Lemmy") + + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusRemoved, status, + "a post that WAS accepted then fails re-admission is REMOVED, not rejected (it federated once)") + assert.Equal(t, DecisionTitleRequired, code, + "the ledger records the SPECIFIC cause even though the firehose removal code is the open-set one") +} + +// --------------------------------------------------------------------------- +// E3 — author-delete: tombstone deletes the acceptance (NO removal) + Delete{Page} +// --------------------------------------------------------------------------- + +func TestAuthorDeleteRemovesAcceptanceWithoutRemovalRecord(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + admittedCreate(t, dispatcher) + _, ok := acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + require.True(t, ok, "precondition: the post was accepted") + + // The author deletes their postv2. The delete commit carries no record and no + // CID; everything the Delete{Page} needs is read from outbound_objects. + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("delete", acPostRKey, acRevDelete, "", acPostTimeUS+2, nil)), + "an author delete must reach the engine and take the acceptance down") + + _, ok = acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + assert.False(t, ok, "the acceptance must be deleted with the author's post") + + _, hasRemoval := removalStandsAt(t, repos, acCommunityDID, acPostURI) + assert.False(t, hasRemoval, + "author deletion is NOT moderation, so NO removal record is written — the acceptance "+ + "just goes away") + + assert.Equal(t, 1, activityKindCount(t, conn, "Delete"), + "the retraction is one Delete{Page}, built from the stored outbound state") + + stored, err := store.NewOutboundObjects(conn).GetByATURI(ctx, acPostURI) + require.NoError(t, err) + assert.True(t, stored.IsTombstoned(), "the post's outbound state is tombstoned") + + // RULING (flagged): the admissions ledger row is DELETED on author-delete — + // the post no longer exists to re-decide, and there is no removal record to + // explain, so a 'removed' row would falsely imply a moderation removal. See + // the report for the audit-history alternative. + assert.Zero(t, admissionRowCount(t, conn, acCommunityDID, acPostURI), + "the ledger row is removed: the decided post is gone and no removal record stands for it") + + // Replaying the delete is a no-op. + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("delete", acPostRKey, acRevDelete, "", acPostTimeUS+2, nil))) + assert.Equal(t, 1, activityKindCount(t, conn, "Delete"), "a replayed author delete is a no-op") +} + +// --------------------------------------------------------------------------- +// E4 — community immutability: an UPDATE that MOVES the post is discarded whole +// --------------------------------------------------------------------------- + +func TestUpdateThatMovesCommunityIsDiscarded(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + seedBridgedCommunityB(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + admittedCreate(t, dispatcher) // accepted into community A + + // The hijack: an edit whose `community` names community B. The lexicon makes + // `community` immutable; the whole event is discarded. + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("update", acPostRKey, acRevUpdate, acPostCID2, acPostTimeUS+1, + pv2Record(func(r map[string]any) { r["community"] = acCommunityB_DID }))), + "a community-moving edit is discarded quietly, not dead-lettered") + + // Community B must have NO acceptance for this post. + _, movedIn := acceptanceSubjectCID(t, repos, acCommunityB_DID, acPostURI) + assert.False(t, movedIn, + "the engine must refuse to move the post: no acceptance may appear in the target community") + + // Community A's acceptance is untouched (still the original version — the + // event was discarded WHOLE, so not even a repin lands). + cid, ok := acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + require.True(t, ok, "the original acceptance stands") + assert.Equal(t, acPostCID, cid, "the original community keeps the version it accepted; nothing moved") + + // No second activity was enqueued. + assert.Equal(t, 1, countRows(t, conn, "outbound_activities"), + "a discarded community-move enqueues nothing") + + // The ledger still binds the post to community A. + assert.Equal(t, 1, admissionRowCount(t, conn, acCommunityDID, acPostURI), + "the post's admission stays under the original community") + assert.Zero(t, admissionRowCount(t, conn, acCommunityB_DID, acPostURI), + "and no admission row is written for the target community") +} + +// --------------------------------------------------------------------------- +// E5 — the rest of admission policy (each records a distinct decision_code) +// --------------------------------------------------------------------------- + +func TestPausedAuthorPostIsRejected(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + seedPausedActor(t, conn, true) // the author already has an actor, now paused + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", acPostRKey, acRevCreate, acPostCID, acPostTimeUS, pv2Record()))) + + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusRejected, status) + assert.Equal(t, DecisionPaused, code, + "a paused (#account) author's post is rejected with a distinct paused code — delivery is halted") + _, ok := acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + assert.False(t, ok, "no acceptance for a paused author") + assert.Zero(t, countRows(t, conn, "outbound_activities"), "and nothing enqueued") +} + +func TestTitleRequiredAndTooLongAreRejectedOnCreate(t *testing.T) { + t.Run("titleless create", func(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", acPostRKey, acRevCreate, acPostCID, acPostTimeUS, + pv2Record(func(r map[string]any) { delete(r, "title") }))), + "a titleless post is a recorded rejection, not an error that retries forever") + + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusRejected, status) + assert.Equal(t, DecisionTitleRequired, code, + "a media-only postv2 rejects with title-required until a title-derivation decision exists") + assert.Zero(t, countRows(t, conn, "outbound_activities")) + }) + + t.Run("over-cap create", func(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + long := strings.Repeat("x", lemmyTitleCap+1) + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", acPostRKey, acRevCreate, acPostCID, acPostTimeUS, + pv2Record(func(r map[string]any) { r["title"] = long })))) + + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusRejected, status) + assert.Equal(t, DecisionTitleTooLong, code, + "a title over Lemmy's 200-char cap rejects with title-too-long") + }) +} + +func TestRateCapRejectsBeyondThePerAuthorPerCommunityLimit(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + // A deliberately tiny cap: two accepted posts, then the third is refused. + enq := realEnqueuer(t, conn) + dispatcher := wireDispatcher(t, conn, + engineWith(t, conn, repos, enq, func(o *Options) { o.MaxPerAuthorPerCommunity = 2 }), enq) + + rkeys := []string{"3lzpostrate01", "3lzpostrate02", "3lzpostrate03"} + revs := []string{"3lzraterev001", "3lzraterev002", "3lzraterev003"} + for i, rkey := range rkeys { + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", rkey, revs[i], acPostCID, acPostTimeUS+int64(i), pv2Record()))) + } + + third := "at://" + acAuthorDID + "/social.coves.community.postv2/" + rkeys[2] + status, code := admissionOf(t, conn, acCommunityDID, third) + assert.Equal(t, StatusRejected, status) + assert.Equal(t, DecisionRateLimit, code, + "the third accepted post in one community by one author exceeds the cap of 2 and is rate-limited") + + // Exactly two acceptances stand (the first two). + var accepted int + require.NoError(t, conn.QueryRowContext(ctx, + `SELECT COUNT(*) FROM admissions WHERE community_did = $1 AND status = 'accepted'`, + acCommunityDID).Scan(&accepted)) + assert.Equal(t, 2, accepted, "only the posts under the cap are accepted") +} + +func TestLexiconInvalidPostIsRejected(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + // createdAt is REQUIRED and must be an atproto datetime string; a number is a + // clear lexicon type violation, distinct from any admission-policy check. + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", acPostRKey, acRevCreate, acPostCID, acPostTimeUS, + pv2Record(func(r map[string]any) { r["createdAt"] = 12345 }))), + "a post we sign the acceptance for must fail closed on invalid input — recorded, not errored") + + status, code := admissionOf(t, conn, acCommunityDID, acPostURI) + assert.Equal(t, StatusRejected, status) + assert.Equal(t, DecisionLexiconInvalid, code, + "strict lexicon validation of the native input: invalid → rejection lexicon-invalid") + _, ok := acceptanceSubjectCID(t, repos, acCommunityDID, acPostURI) + assert.False(t, ok, "no acceptance is signed over input that does not validate") +} + +// Ban / community-tombstone / parent-lock are TASK 17 (moderation + community +// lifecycle). Pinned here as a skipped placeholder so the boundary is explicit +// and the engine does NOT silently implement them in task 16. +func TestBanAndCommunityLifecycleChecksAreTask17(t *testing.T) { + t.Skip("author-banned, community-unfollowed/tombstoned, and parent-locked admission " + + "checks belong to task 17 (echo-moderation + community lifecycle); task 16 does not implement them") +} + +// --------------------------------------------------------------------------- +// E6 — re-acceptance race / replay + no retroactive acceptance +// --------------------------------------------------------------------------- + +// A post rejected while its author was opted out stays rejected after the author +// re-enables and posts something NEW: content authored while opted out never +// federates (decision 11 — the author reposts). +func TestRejectedWhileOptedOutStaysRejectedAfterReEnable(t *testing.T) { + conn := acceptanceDB(t) + ctx := context.Background() + seedBridgedCommunity(t, conn) + repos := newRepos(t, conn) + dispatcher := wireDispatcher(t, conn, engineWith(t, conn, repos, realEnqueuer(t, conn)), realEnqueuer(t, conn)) + + // Opted out: the first post is rejected. + _, err := store.NewFederationPrefs(conn).Upsert(ctx, store.FederationPref{ + DID: acAuthorDID, Source: store.FederationPrefSourceRecord, + }) + require.NoError(t, err) + + firstRKey := "3lzpostopt0a" + firstURI := "at://" + acAuthorDID + "/social.coves.community.postv2/" + firstRKey + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", firstRKey, "3lzoptrev001", acPostCID, acPostTimeUS, pv2Record()))) + status, code := admissionOf(t, conn, acCommunityDID, firstURI) + require.Equal(t, StatusRejected, status) + require.Equal(t, DecisionOptedOut, code) + + // Re-enable: deleting the opt-out record restores default-on. + require.NoError(t, store.NewFederationPrefs(conn).Delete(ctx, acAuthorDID)) + + // A NEW post is accepted. + secondRKey := "3lzpostopt0b" + secondURI := "at://" + acAuthorDID + "/social.coves.community.postv2/" + secondRKey + require.NoError(t, dispatcher.HandleEvent(ctx, + postEvent("create", secondRKey, "3lzoptrev002", acPostCID, acPostTimeUS+1, pv2Record()))) + status2, _ := admissionOf(t, conn, acCommunityDID, secondURI) + assert.Equal(t, StatusAccepted, status2, "a new post after re-enabling federates") + + // The OLD post stays rejected — no retroactive acceptance. + status, code = admissionOf(t, conn, acCommunityDID, firstURI) + assert.Equal(t, StatusRejected, status, + "the post authored while opted out stays rejected: re-enabling does not federate it retroactively") + assert.Equal(t, DecisionOptedOut, code) + _, ok := acceptanceSubjectCID(t, repos, acCommunityDID, firstURI) + assert.False(t, ok, "the old post has no acceptance") +} diff --git a/internal/config/config.go b/internal/config/config.go --- a/internal/config/config.go +++ b/internal/config/config.go @@ -131,6 +131,12 @@ // default 0 = OFF). Delivery starts ONLY when this is >0 AND // ConsumerEnabled: until a deployment is wired end to end, the consumer // still records outbound state but the noop enqueuer federates nothing. OutboundWorkers int + // AdmissionMaxPerAuthorPerCommunity caps how many posts one native author may + // have accepted into one bridged community — the acceptance engine's flood + // guard (ADMISSION_MAX_PER_AUTHOR_PER_COMMUNITY, generous default 50). 0 means + // UNLIMITED. Tidepool signs the community's acceptance, so it must not let one + // account flood a Lemmy community it vouches for. + AdmissionMaxPerAuthorPerCommunity int // OutboundDryRun makes workers translate + log but POST nothing, leaving // deliveries pending (OUTBOUND_DRY_RUN, default false). OutboundDryRun bool @@ -471,6 +477,12 @@ // Outbound delivery (task 15), default OFF: workers start only when // OUTBOUND_WORKERS>0 AND the consumer is on, so a not-yet-wired deployment // keeps the noop enqueuer and federates nothing. cfg.OutboundWorkers, err = intVarNonNegative(logger, "OUTBOUND_WORKERS", 0) + if err != nil { + return nil, err + } + // The acceptance engine's per-author-per-community flood cap. Generous by + // default so a legitimate poster is never throttled; 0 disables it entirely. + cfg.AdmissionMaxPerAuthorPerCommunity, err = intVarNonNegative(logger, "ADMISSION_MAX_PER_AUTHOR_PER_COMMUNITY", 50) if err != nil { return nil, err } diff --git a/internal/db/migrations/021_admissions.sql b/internal/db/migrations/021_admissions.sql --- a/internal/db/migrations/021_admissions.sql +++ b/internal/db/migrations/021_admissions.sql @@ -16,6 +16,13 @@ -- log. CREATE TABLE admissions ( community_did TEXT NOT NULL, -- the bridged community the post was submitted to post_uri TEXT NOT NULL, -- the postv2 at-uri (author repo) + -- author_did is the postv2 author (the repo owner). It is what the + -- per-author-per-community flood cap counts on: Tidepool must not let one + -- native account flood a Lemmy community it vouches for, so the engine + -- counts a (author_did, community_did) author's ACCEPTED rows against + -- ADMISSION_MAX_PER_AUTHOR_PER_COMMUNITY. Denormalized from post_uri's + -- authority segment so the cap is one indexed COUNT rather than a LIKE scan. + author_did TEXT NOT NULL DEFAULT '', status TEXT NOT NULL CHECK (status IN ('pending', 'accepted', 'pending_reacceptance', 'rejected', 'removed')), -- decision_code is the machine-readable reason for the current status: '' @@ -49,6 +56,12 @@ -- ledger is small (one row per bridged post) but the index keeps that listing -- from scanning removed/accepted rows. CREATE INDEX idx_admissions_status ON admissions (status) WHERE status IN ('pending', 'rejected'); +-- The per-author-per-community flood cap counts accepted rows for one author in +-- one community. Leading (author_did, community_did) serves the WHERE; created_at +-- trails so a windowed variant of the cap stays index-only. +CREATE INDEX idx_admissions_author_community ON admissions (author_did, community_did, created_at); + -- +goose Down +DROP INDEX IF EXISTS idx_admissions_author_community; DROP INDEX IF EXISTS idx_admissions_status; DROP TABLE IF EXISTS admissions;