diff --git a/docs/PRD_AUTHOR_OWNED_POSTS.md b/docs/PRD_AUTHOR_OWNED_POSTS.md index c4cdf80..0a349a9 100644 --- a/docs/PRD_AUTHOR_OWNED_POSTS.md +++ b/docs/PRD_AUTHOR_OWNED_POSTS.md @@ -28,7 +28,14 @@ tuple CAS over a removal. Stale/terminal skips are outcome values, not errors (033 precedent — sentinels would dead-letter healthy skips). Rev 2.4 (2026-08-08): rejection narrowed to pending-only CAS with judged CID; op-rank derived repo-side; NULL-evaluated acceptance treated as pin-trusting -(task-2 second-opinion catches).** +(task-2 second-opinion catches). +Rev 2.5 (2026-08-08): §4.1 corrected by task-3 plan review — ban source is +community_memberships.is_banned behind a BanLookup interface (no +moderation.ban ingestion exists; no production ban writer yet); rate +limits/dedupe get a synchronous post_submissions ledger (migration 035) — +the posts table is unusable as a limiter substrate (ingestion lag, +author-supplied created_at, delete-to-evade); per-origin-PDS quota +explicitly deferred to Beta.** **Supersedes** the write-path architecture in `docs/federation-prd.md`: that document solves cross-instance posting by service-auth-forwarding the write to @@ -271,9 +278,34 @@ authorization + rate limits — nothing else**. The docstring's Therefore `admitPost` (§5.6) is **extraction plus new policy**, not a behavior-preserving refactor. New checks arriving with it, each with an -explicit error code and tests: ban lookup against indexed -`social.coves.moderation.ban` state, and per-author/per-community submission -rate limits (§8). The spec stops claiming otherwise. +explicit error code and tests: ban enforcement, per-author/per-community +submission rate limits, and duplicate-submission dedupe (§8). + +**Ban source, honestly (task-3 plan-review correction):** the only ban state +in the system is `community_memberships.is_banned` — and no production code +path writes it today (the memberships repo has no non-test callers; no +`social.coves.moderation.ban` consumer exists and the collection is not in +`consumerWantedCollections`). `admitPost` therefore enforces bans through a +`BanLookup` interface backed by that column, making enforcement live the +moment a ban writer ships (moderation write path and/or ban-record +ingestion — future scope, not this loop). Non-membership reads as +not-banned; any lookup FAILURE fails the request closed — failing open on a +ban would turn a database blip into a global unban. + +**Rate-limit substrate:** limits and dedupe are backed by a synchronous +`post_submissions` ledger (migration 035, mirroring `aggregator_posts`) with +a canonical-record fingerprint and a UNIQUE-insert dedupe gate, +reserve-then-confirm around the PDS write. The `posts` table cannot back +them: it is firehose-fed (ingestion lag hides the very burst being limited), +its `created_at` is author-supplied (attacker-controlled windows once writes +flip to author repos), and its indexes exclude soft-deleted rows +(delete-to-evade). Refused submissions consume no quota. Dedupe precedes the +rate limit (a client retry storm must not burn quota) and applies to every +actor class; trusted aggregators keep their historical no-limit status and +registered aggregators are governed by their existing limiter only. +§8's per-origin-PDS quota is **deferred** to the Beta remote path (it +requires PDS resolution, §7) — recorded here so §8 does not silently become +fiction. ### 4.2 Flow diff --git a/internal/api/handlers/post/errors_test.go b/internal/api/handlers/post/errors_test.go index 2209ed3..9807f83 100644 --- a/internal/api/handlers/post/errors_test.go +++ b/internal/api/handlers/post/errors_test.go @@ -157,6 +157,26 @@ func TestAggregatorErrorCodes(t *testing.T) { } } +// A submission refused as a repeat is a 409, and it must not be confused with +// anything else. +// +// Two client behaviours depend on the distinction. A 409 says "your post +// already exists, stop retrying and go look for it", which is exactly what a +// client whose response was lost needs to hear; a 429 says "wait", and a client +// told to wait would resend the same content on a timer forever. And the code +// must be its own — folding it into the generic AlreadyExists that +// coreerrors.ConflictError produces would leave a client unable to tell a +// refused submission from a record the indexer already holds. +func TestDuplicateSubmissionIsItsOwnConflict(t *testing.T) { + rec := httptest.NewRecorder() + handleServiceError(rec, fmt.Errorf("createPost: %w", posts.ErrDuplicateSubmission)) + + body := assertXRPCError(t, rec, http.StatusConflict, "DuplicateSubmission") + if strings.Contains(body.Message, "createPost") { + t.Errorf("wrapper context leaked into the client message: %q", body.Message) + } +} + // posts.ErrCommunityNotFound must keep beating the generic not-found rule that // also matches it. func TestCommunityNotFoundBeatsGenericNotFound(t *testing.T) { diff --git a/internal/config/config.go b/internal/config/config.go index 4ea98cc..a816ee1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -106,6 +106,9 @@ type Config struct { Signup SignupConfig Media MediaConfig + // Submissions bounds what one author may post into one community. + Submissions SubmissionsConfig + // CursorSecret is the HMAC key that signs pagination cursors, preventing // clients from forging or tampering with them. CursorSecret string @@ -326,6 +329,34 @@ type MediaConfig struct { AllowUnproxiedMedia bool } +// SubmissionsConfig bounds what one author may submit to one community +// (docs/PRD_AUTHOR_OWNED_POSTS.md §8). +// +// It mirrors posts.SubmissionLimits field for field rather than embedding it. +// The duplication is deliberate: this package is imported by everything that +// starts a process, and giving it a dependency on a core domain package would +// make the domain's import graph the startup path's problem. The mapping is one +// struct literal at wiring time. +// +// EVERY FIELD IS REQUIRED. There is no "unset means unlimited" reading, which +// is the whole reason these are validated at startup: a quota that evaporates +// when someone forgets an environment variable is indistinguishable, in +// production, from having no quota at all — and it fails open, silently, on the +// one path that exists to bound abuse. +type SubmissionsConfig struct { + // MaxPerAuthorPerCommunity is how many posts one author may have admitted + // to one community inside Window. + MaxPerAuthorPerCommunity int + + // Window is the rolling window the quota is counted over. + Window time.Duration + + // DedupeWindow scopes how long an identical resubmission is refused as a + // repeat. It is separate from Window because the two answer different + // questions: one bounds volume, the other catches retries. + DedupeWindow time.Duration +} + // TokenEndpointEnabled reports whether the signup-token endpoint can operate. // It needs both the captcha secret and (from PDSConfig) an admin password to // mint invite codes, so the caller passes the latter in. diff --git a/internal/config/submissions_test.go b/internal/config/submissions_test.go new file mode 100644 index 0000000..39b62db --- /dev/null +++ b/internal/config/submissions_test.go @@ -0,0 +1,115 @@ +package config + +import ( + "strings" + "testing" + "time" +) + +// The per-author submission quota of docs/PRD_AUTHOR_OWNED_POSTS.md §8 is +// configuration, and configuration that goes missing must stop the process. +// +// The failure this guards against is specific: an operator who never sets +// POST_SUBMISSIONS_MAX_PER_COMMUNITY gets a zero, a limit check written as +// `count >= limit` then refuses everything (or, written the other way, admits +// everything), and either way the behaviour is decided by an omission rather +// than by a decision. §8's quotas exist to absorb the fact that anyone can +// write unlimited records naming any community — so "unset" cannot be allowed +// to mean "unlimited", and validating at startup is the only place the answer +// is cheap. + +func TestLoad_SubmissionQuotaHasWorkingDefaults(t *testing.T) { + clearEnv(t) + t.Setenv("IS_DEV_ENV", "true") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load() returned error: %v", err) + } + + if cfg.Submissions.MaxPerAuthorPerCommunity <= 0 { + t.Errorf("Submissions.MaxPerAuthorPerCommunity = %d, want a positive default; "+ + "a zero here is a quota decided by omission", + cfg.Submissions.MaxPerAuthorPerCommunity) + } + if cfg.Submissions.Window <= 0 { + t.Errorf("Submissions.Window = %s, want a positive rolling window", cfg.Submissions.Window) + } + if cfg.Submissions.DedupeWindow <= 0 { + t.Errorf("Submissions.DedupeWindow = %s, want a positive dedupe window", cfg.Submissions.DedupeWindow) + } +} + +func TestLoad_SubmissionQuotaIsReadFromTheEnvironment(t *testing.T) { + clearEnv(t) + t.Setenv("IS_DEV_ENV", "true") + t.Setenv("POST_SUBMISSIONS_MAX_PER_COMMUNITY", "7") + t.Setenv("POST_SUBMISSIONS_WINDOW", "30m") + t.Setenv("POST_SUBMISSIONS_DEDUPE_WINDOW", "10m") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load() returned error: %v", err) + } + + if cfg.Submissions.MaxPerAuthorPerCommunity != 7 { + t.Errorf("MaxPerAuthorPerCommunity = %d, want 7", cfg.Submissions.MaxPerAuthorPerCommunity) + } + if cfg.Submissions.Window != 30*time.Minute { + t.Errorf("Window = %s, want 30m", cfg.Submissions.Window) + } + if cfg.Submissions.DedupeWindow != 10*time.Minute { + t.Errorf("DedupeWindow = %s, want 10m", cfg.Submissions.DedupeWindow) + } +} + +// A config assembled with the quota left at its zero value must not validate. +// This is the assertion that makes "unset means unlimited" unrepresentable +// rather than merely discouraged. +func TestValidate_RejectsAnUnsetSubmissionQuota(t *testing.T) { + base := func() *Config { + return &Config{ + IsDevEnv: true, + Database: DatabaseConfig{URL: "postgres://u:p@db/coves", MaxOpenConns: 25, MaxIdleConns: 25}, + Server: ServerConfig{Port: "8080", ReadHeaderTimeout: time.Second, ReadTimeout: 30 * time.Second, WriteTimeout: 30 * time.Second, IdleTimeout: 60 * time.Second, ShutdownTimeout: 15 * time.Second}, + Instance: InstanceConfig{DID: "did:web:coves.social", Domain: "coves.social"}, + CursorSecret: devCursorSecret, + Jetstream: JetstreamConfig{FeedsSpec: "self=ws://localhost:6008"}, // coves:allow-host-literal: a non-empty spec so Validate's unrelated JETSTREAM_FEEDS rule is satisfied; parsing lives elsewhere and nothing here dials it + Submissions: SubmissionsConfig{ + MaxPerAuthorPerCommunity: 10, + Window: time.Hour, + DedupeWindow: time.Hour, + }, + } + } + + // The control: the fully-specified config validates, so a failure below is + // about the field that was cleared and not about the fixture. + if err := base().Validate(); err != nil { + t.Fatalf("the fully-specified config must validate; got: %v", err) + } + + for _, tc := range []struct { + name string + clear func(*Config) + want string + }{ + {"no per-community limit", func(c *Config) { c.Submissions.MaxPerAuthorPerCommunity = 0 }, "POST_SUBMISSIONS_MAX_PER_COMMUNITY"}, + {"no window", func(c *Config) { c.Submissions.Window = 0 }, "POST_SUBMISSIONS_WINDOW"}, + {"no dedupe window", func(c *Config) { c.Submissions.DedupeWindow = 0 }, "POST_SUBMISSIONS_DEDUPE_WINDOW"}, + {"a negative limit", func(c *Config) { c.Submissions.MaxPerAuthorPerCommunity = -1 }, "POST_SUBMISSIONS_MAX_PER_COMMUNITY"}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := base() + tc.clear(cfg) + + err := cfg.Validate() + if err == nil { + t.Fatal("Validate() accepted a submission quota that is not a quota; the process would start with abuse limits silently disabled") + } + if !strings.Contains(err.Error(), tc.want) { + t.Errorf("error should name %s so an operator can fix it; got:\n%s", tc.want, err.Error()) + } + }) + } +} diff --git a/internal/config/testing.go b/internal/config/testing.go index fe91146..c74f268 100644 --- a/internal/config/testing.go +++ b/internal/config/testing.go @@ -33,6 +33,8 @@ var loadedEnvVars = []string{ "IMAGE_PROXY_CLEANUP_INTERVAL_MINUTES", "IMAGE_PROXY_FETCH_TIMEOUT_SECONDS", "IMAGE_PROXY_MAX_SOURCE_SIZE_MB", "ALLOW_UNPROXIED_MEDIA", + "POST_SUBMISSIONS_MAX_PER_COMMUNITY", "POST_SUBMISSIONS_WINDOW", + "POST_SUBMISSIONS_DEDUPE_WINDOW", } // ClearEnvForTest blanks every environment variable Load reads, restoring them diff --git a/internal/core/posts/admissions.go b/internal/core/posts/admissions.go index 93c84da..660563f 100644 --- a/internal/core/posts/admissions.go +++ b/internal/core/posts/admissions.go @@ -341,3 +341,58 @@ type AdmissionRepository interface { // make a moderator re-review what they had already cleared. ListByStatusForCommunity(ctx context.Context, communityDID string, status AdmissionStatus, limit int, cursor *string) ([]*Admission, *string, error) } + +// DecisionCode is the reason a post was refused or removed — the value stored +// in community_post_admissions.decision_code and, for the subset that a +// community publishes, in a social.coves.community.removal record's `code`. +// +// It is a plain named string with NO validation function, deliberately. The +// removal lexicon spells `code` as `knownValues`, which is an OPEN set by +// definition — "values are not limited to this set" (§3.3) — so that new codes +// ship without a lexicon break. A Go-side IsValid would re-close what the +// lexicon deliberately left open, and would start rejecting perfectly legal +// codes minted by a remote community running a newer build than ours. +type DecisionCode string + +// The six codes the removal lexicon names (§3.3). These are the vocabulary a +// COMMUNITY publishes: they appear in removal records, so a client reading the +// firehose meets them. Kebab-case is the lexicon style guide's convention for +// fixed strings. +const ( + DecisionRuleViolation DecisionCode = "rule-violation" + DecisionSpam DecisionCode = "spam" + DecisionOffTopic DecisionCode = "off-topic" + DecisionIllegalContent DecisionCode = "illegal-content" + DecisionAuthorBanned DecisionCode = "author-banned" + DecisionModeratorDiscretion DecisionCode = "moderator-discretion" +) + +// The admission-time codes. These never reach a community repo: §3.3 is +// explicit that a submission refused before it was ever accepted writes NO +// record, because spam must not bloat the community's repository. They live in +// the same vocabulary anyway — the admissions table's decision_code column +// stores both kinds, and getStatus serves both to the author who asked why. +const ( + // DecisionRateLimitExceeded: the author is over their per-community + // submission quota (§8). + DecisionRateLimitExceeded DecisionCode = "rate-limit-exceeded" + + // DecisionDuplicateSubmission: an identical submission from this author to + // this community is already on the ledger for the current window. + DecisionDuplicateSubmission DecisionCode = "duplicate-submission" + + // DecisionCommunityNotFound: the at-identifier names no community this + // AppView has indexed. + DecisionCommunityNotFound DecisionCode = "community-not-found" + + // DecisionCommunityPrivate: a private community refusing a regular user. + // + // It is deliberately the answer for a BANNED user of a private community + // too — see admitPost's check order, which explains why a ban must not be + // disclosed through a privacy wall. + DecisionCommunityPrivate DecisionCode = "community-private" + + // DecisionAggregatorNotAuthorized: a registered aggregator the community + // has not authorized (or whose authorization it has revoked). + DecisionAggregatorNotAuthorized DecisionCode = "aggregator-not-authorized" +) diff --git a/internal/core/posts/admit.go b/internal/core/posts/admit.go new file mode 100644 index 0000000..23a0b41 --- /dev/null +++ b/internal/core/posts/admit.go @@ -0,0 +1,329 @@ +package posts + +import ( + "context" + "time" + + "Coves/internal/core/communities" +) + +// admitPost: the single decision point for "may this submission become a post +// in this community?" (PRD_AUTHOR_OWNED_POSTS.md §4.1, §5.6, §8). +// +// It is extraction PLUS new policy, and §4.1 is blunt about which is which. +// What CreatePost enforces today is community existence, a private-visibility +// block for regular users, and aggregator authorization + the aggregator's own +// hourly quota — nothing else. The service docstring's "membership/ban +// validation" was aspirational: there is no ban lookup anywhere on the write +// path, and no per-author rate limiting at all. Both arrive here. +// +// WHY A SEPARATE FUNCTION RATHER THAN MORE STEPS IN CreatePost. The same +// decision has to be made from three places: the synchronous local-community +// fast path (§4.2 step 4), the firehose consumer when a post for a community we +// host arrives from someone else's PDS (§5.6), and the notify endpoint (§7). +// A decision that lived inside CreatePost would be reachable only from the +// first, so the other two would silently admit what the first refuses. + +// ActorClass is what the CALLER has already established the submitter to be. +// +// It is an INPUT rather than something this decision derives, and that is the +// whole point. Today's classification reads TRUSTED_AGGREGATOR_DIDS (falling +// back to KAGI_AGGREGATOR_DID) out of the process environment inside CreatePost +// (service.go step 3). A decision function that reached for os.Getenv itself +// could not have its trusted-actor branch tested alongside t.Parallel — Go's +// own testing package refuses t.Setenv there — and, worse, would hide "who is +// trusted" from the call site of the security decision it governs. +type ActorClass string + +const ( + // ActorUser is a person posting on their own behalf. Every check applies. + ActorUser ActorClass = "user" + + // ActorRegisteredAggregator is a service the AppView has indexed a + // social.coves.aggregator.service declaration for. It is held to the + // community's authorization record and to its OWN hourly quota + // (aggregators.ValidateAggregatorPost), not to membership or visibility. + ActorRegisteredAggregator ActorClass = "registered_aggregator" + + // ActorTrustedAggregator is a service named in TRUSTED_AGGREGATOR_DIDS — + // the temporary env-var mechanism that predates a real authorization + // endpoint. It skips visibility, ban and authorization checks entirely, + // which is the existing behaviour and is preserved deliberately. + ActorTrustedAggregator ActorClass = "trusted_aggregator" +) + +// AdmissionRequest is one submission, described in the terms the decision needs +// and no others. There is no record and no blob here: admitPost runs BEFORE any +// of that work, so that a refusal costs a lookup rather than an upload. +type AdmissionRequest struct { + // Actor is the class the caller resolved. See ActorClass. + Actor ActorClass + + // AuthorDID is the authenticated author. CreatePost has already proven it + // matches the DID on the request; this decision trusts that. + AuthorDID string + + // Community is the at-identifier as the client sent it — a handle + // (!gardening.communities.coves.social) or a DID. Resolving it is the + // decision's first step, because a community that does not resolve is the + // first thing that can refuse a submission. + Community string + + // Fingerprint identifies WHAT is being submitted: the hash of the canonical + // record with createdAt removed (see submissionFingerprint). It is the + // dedupe key, and it must exclude the timestamp or every resubmission of + // identical content would look new. + Fingerprint string +} + +// AdmissionDecision is the answer: admitted, or refused with a code. +// +// A refusal is a VALUE rather than an error, matching AdmissionOutcome above +// and the project's standing preference for error codes over booleans. The +// caller has to translate the code into whatever its transport speaks — a +// sentinel error for CreatePost, an admissions row for the firehose engine — +// and a refusal returned as an error would push the second of those into the +// dead-letter queue, which is meant to hold genuine failures. +type AdmissionDecision struct { + // Code is the reason for a refusal, and empty for an admission. + Code DecisionCode + + // Community is the resolved community, populated on admission so that + // CreatePost does not fetch it a second time. It is the one piece of state + // the decision has already paid for that its caller would otherwise re-buy. + Community *communities.Community + + // Reservation is the ledger row that was inserted for this submission. It + // is present on admission and must be released if the PDS write that + // follows fails — see SubmissionLedger. + Reservation *SubmissionReservation + + // Cause carries the underlying error behind a refusal, when there is one, + // so the caller can wrap it and keep it matchable. + // + // It exists for exactly one case today: aggregator authorization. The API + // boundary maps that refusal through aggregators.IsUnauthorized and + // aggregators.IsRateLimited (internal/api/handlers/post/errors.go), which + // are predicates over the AGGREGATORS package's sentinels. Collapsing that + // error into a bare DecisionCode would turn a 403 "stop asking" and a 429 + // "ask later" into the same answer, and a well-behaved aggregator would + // retry a permanent refusal forever. + Cause error +} + +// Admitted reports whether the submission may proceed. There is no separate +// bool field: two representations of one fact drift, and the code is the one +// that has to be right. +func (d AdmissionDecision) Admitted() bool { return d.Code == "" } + +// SubmissionReservation identifies the ledger row admitPost inserted for a +// submission, so a caller whose subsequent PDS write failed can release it. +type SubmissionReservation struct { + ID int64 +} + +// SubmissionLimits bounds what one author may submit to one community (§8). +// +// Every field is required. There is deliberately no "zero means unlimited" +// reading: a quota that silently disappears when an environment variable is +// missing is not a quota, so config.Validate refuses to start the process with +// any of these unset. +type SubmissionLimits struct { + // MaxPerAuthorPerCommunity is how many submissions one author may have + // admitted to one community inside Window. + MaxPerAuthorPerCommunity int + + // Window is the rolling window the quota is counted over, matching the + // aggregator limiter's semantics (aggregators.RateLimitWindow): a COUNT of + // ledger rows newer than now-Window, not a fixed bucket that empties on the + // hour and lets an author spend twice across the boundary. + Window time.Duration + + // DedupeWindow is the width of the bucket that scopes dedupe uniqueness. + // Without it the ledger's unique constraint would forbid an author from + // ever reposting identical content again, which is a different and much + // stronger policy than "do not accept the same thing twice right now". + DedupeWindow time.Duration +} + +// Clock is the decision's only source of time. +// +// Injected rather than called directly so that window expiry is testable +// without waiting for one: docs/TEST_ARCHITECTURE.md §3.3 records that +// time.Sleep in a test fails the audit, and that a rate limiter's window is +// crossed through an injected clock. +type Clock func() time.Time + +// CommunityLookup resolves an at-identifier and fetches the community behind +// it. Satisfied by communities.Service. +type CommunityLookup interface { + // ResolveCommunityIdentifier turns a handle or a DID into a DID. + ResolveCommunityIdentifier(ctx context.Context, identifier string) (string, error) + + // GetByDID returns the indexed community. + GetByDID(ctx context.Context, did string) (*communities.Community, error) +} + +// BanLookup answers whether an author is banned from a community, by returning +// the membership row that carries the answer. +// +// It returns the whole membership rather than a bool so that the translation of +// "no membership row" into "not banned" happens in ONE place — inside +// admitPost, next to the comment that explains why an error is not the same +// thing. Satisfied by communities.Service. +type BanLookup interface { + // GetMembership returns the author's membership of the community, or an + // error wrapping communities.ErrMembershipNotFound when there is none. + GetMembership(ctx context.Context, userDID, communityIdentifier string) (*communities.Membership, error) +} + +// AggregatorAuthorizer checks a registered aggregator's authorization and its +// own quota. Satisfied by aggregators.Service. +type AggregatorAuthorizer interface { + ValidateAggregatorPost(ctx context.Context, aggregatorDID, communityDID string) error +} + +// ReserveSubmissionCommand is one row of the post_submissions ledger. +type ReserveSubmissionCommand struct { + AuthorDID string + CommunityDID string + + // Fingerprint is the content hash — see AdmissionRequest.Fingerprint. + Fingerprint string + + // DedupeBucket is the index of the DedupeWindow this submission falls in, + // derived from the injected clock. It is part of the unique key, which is + // what makes dedupe expire. + DedupeBucket int64 +} + +// SubmissionLedger records admitted submissions, and IS the dedupe gate. +// +// RESERVE-THEN-CONFIRM. The row goes in BEFORE the PDS write and is released if +// that write fails. The alternative — record after a successful write — leaves +// a window in which two concurrent identical submissions both pass the check +// and both get written, which is precisely the double-tap this exists to stop. +// A leaked reservation (process died between the insert and the release) costs +// the author one quota slot until the window rolls; a missed one costs the +// community a duplicate post. The asymmetry decides the direction. +// +// THE INSERT IS THE CHECK. Dedupe is not a SELECT followed by an INSERT: it is +// the INSERT, with the unique constraint as the arbiter. A read-then-write +// would reopen the same race under concurrency, and the database is the only +// participant that can serialize it. +type SubmissionLedger interface { + // Reserve inserts the ledger row for a submission. A unique-constraint + // violation means an identical submission is already recorded for this + // window, and is reported as ErrDuplicateSubmission rather than as a driver + // error — the caller has to tell "someone already posted this" apart from + // "the database is unwell". + Reserve(ctx context.Context, cmd ReserveSubmissionCommand) (SubmissionReservation, error) + + // Release removes a reservation whose submission never became a post. It is + // idempotent: releasing a row that is already gone is not an error, because + // the caller reaches this path while already handling a failure and must + // not be handed a second one. + Release(ctx context.Context, reservation SubmissionReservation) error + + // CountSince counts one author's submissions to one community at or after + // `since` — the rolling-window quota query. + CountSince(ctx context.Context, authorDID, communityDID string, since time.Time) (int, error) +} + +// AdmissionPolicy is the collaborator set the new §8 policy needs, over and +// above what postService already holds. +type AdmissionPolicy struct { + Ledger SubmissionLedger + Bans BanLookup + Limits SubmissionLimits + Now Clock +} + +// WithAdmissionPolicy enables the ban check, dedupe and per-author rate limit +// on CreatePost. +func WithAdmissionPolicy(policy AdmissionPolicy) PostServiceOption { + return func(s *postService) { s.admission = &policy } +} + +// admissionDeps is everything admitPost reads, gathered so the decision is a +// function of its arguments rather than of a service's field set. +type admissionDeps struct { + communities CommunityLookup + bans BanLookup + aggregators AggregatorAuthorizer + ledger SubmissionLedger + limits SubmissionLimits + now Clock +} + +// admitPost decides whether one submission may become a post. +// +// CHECK ORDER — each step its own refusal, and the order is load-bearing: +// +// 1. Community resolution. Nothing else can be evaluated against a community +// that does not exist. +// +// 2. Private visibility, for regular users. A banned member of a PRIVATE +// community is refused with DecisionCommunityPrivate, NOT with +// DecisionAuthorBanned — the ban lookup is not even consulted. A private +// community's moderation state is behind the same wall as its content, and +// answering "you are banned" would confirm to an outsider both that the +// community exists and that a moderator has acted on them. +// +// 3. Ban, for regular users in public communities. A membership row with no +// ban, or NO membership row at all, is not a ban — that is the ordinary +// case, since posting to a public community does not require joining it. +// Any OTHER lookup error FAILS the request. Failing open here would turn a +// Postgres blip into a global unban for its duration, which is the one +// failure mode a ban check exists to prevent. +// +// 4. Aggregator authorization, for registered aggregators. Existing +// semantics, existing sentinels, carried on the decision's Cause. +// +// 5. Dedupe, for EVERY actor class. An aggregator re-polling an RSS feed and +// resubmitting an identical item is the canonical case, so exempting +// trusted actors here would exempt the exact traffic the check is for. +// +// 6. Rate limit, for regular users only. +// +// DEDUPE BEFORE RATE LIMIT. A client whose response was lost retries; if the +// retry burned quota, a flaky connection would rate-limit a user who posted +// once. Dedupe recognises the retry for what it is and refuses it without +// charging for it. +// +// TRUSTED AGGREGATORS skip 2, 3, 4 (existing behaviour) and also skip 6: they +// have no submission limit today, and inventing one here would be a silent +// production behaviour change smuggled in under a refactor. REGISTERED +// aggregators skip 6 for a different reason — they are already governed by +// their own hourly quota inside ValidateAggregatorPost (step 4), and applying +// the new per-author limit as well would silently halve an authorized +// aggregator's throughput. +// +// A REFUSAL CONSUMES NO QUOTA: no refusal leaves a ledger row behind, including +// the rate-limit refusal itself, whose reservation is released before it +// returns. Otherwise an author who kept retrying past their limit would extend +// their own lockout indefinitely. +// +// A non-nil error means the decision could NOT be made — a lookup failed — and +// is distinct from a refusal, which is a decision. +func admitPost(ctx context.Context, deps admissionDeps, req AdmissionRequest) (AdmissionDecision, error) { + return AdmissionDecision{}, nil +} + +// dedupeBucket is the index of the window `now` falls in, so that two +// submissions in the same window collide on the ledger's unique key and two +// submissions a window apart do not. +func dedupeBucket(now time.Time, window time.Duration) int64 { + return 0 +} + +// submissionFingerprint hashes what a moderator would judge about a record: +// everything except createdAt. +// +// The timestamp has to go. It is stamped by the server at submission time +// (service.go step 9), so it differs on every attempt — including the retry +// after a lost response, which is the case dedupe exists to catch. A +// fingerprint that included it would never match anything. +func submissionFingerprint(record PostRecord) string { + return "" +} diff --git a/internal/core/posts/admit_matrix_test.go b/internal/core/posts/admit_matrix_test.go new file mode 100644 index 0000000..0f77e17 --- /dev/null +++ b/internal/core/posts/admit_matrix_test.go @@ -0,0 +1,983 @@ +package posts + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "Coves/internal/core/aggregators" + "Coves/internal/core/communities" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The admitPost decision matrix (docs/PRD_AUTHOR_OWNED_POSTS.md §4.1, §5.6, §8). +// +// admitPost is a pure function over injected lookups and an injected clock, so +// every branch is reachable here without Postgres, without a PDS, and without +// waiting for a rate-limit window to roll. That is the point of the extraction: +// the same checks used to be interleaved with blob uploads and PDS writes +// inside CreatePost, where the only way to reach the private-community branch +// was to provision a private community on a real PDS. +// +// The outer contract — that CreatePost actually consults this, against real +// rows, in the right place in its flow — is service_admission_test.go. What is +// proven HERE is the policy itself, at the width the policy has. +// +// THE LEDGER FAKE IS A MODEL, NOT A RECORDER. stubLedger enforces the same +// unique key the real table does and answers CountSince from the same rows it +// accepted, so "ten admitted then the eleventh refused" is a genuine boundary +// crossing rather than a canned answer. A recorder-shaped fake would let an +// implementation that never consulted the count pass every case below. + +const ( + admitAuthorDID = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" + admitAggregatorDID = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb" + admitCommunityDID = "did:plc:cccccccccccccccccccccccc" + admitCommunityHandle = "!gardening.communities.coves.social" +) + +// --------------------------------------------------------------------------- +// Fakes +// --------------------------------------------------------------------------- + +// stubCommunities answers with one community, or with whichever failure it was +// handed. It counts its calls so a test can prove a later check never ran. +type stubCommunities struct { + community *communities.Community + resolveErr error + getErr error + + resolveCalls int + getCalls int +} + +func (s *stubCommunities) ResolveCommunityIdentifier(_ context.Context, _ string) (string, error) { + s.resolveCalls++ + if s.resolveErr != nil { + return "", s.resolveErr + } + return s.community.DID, nil +} + +func (s *stubCommunities) GetByDID(_ context.Context, _ string) (*communities.Community, error) { + s.getCalls++ + if s.getErr != nil { + return nil, s.getErr + } + return s.community, nil +} + +// stubBans is the community_memberships lookup. Its default — the zero value — +// is the ordinary case: no membership row, because posting in a public +// community has never required joining it. +type stubBans struct { + membership *communities.Membership + err error + + calls int + lastIdentifier string + lastAuthorDID string +} + +func (s *stubBans) GetMembership(_ context.Context, userDID, communityIdentifier string) (*communities.Membership, error) { + s.calls++ + s.lastAuthorDID = userDID + s.lastIdentifier = communityIdentifier + if s.err != nil { + return nil, s.err + } + if s.membership == nil { + return nil, communities.ErrMembershipNotFound + } + return s.membership, nil +} + +// stubAggregatorAuthorizer stands in for aggregators.Service, whose own +// authorization and hourly-quota rules have their own tests. +type stubAggregatorAuthorizer struct { + err error + calls int +} + +func (s *stubAggregatorAuthorizer) ValidateAggregatorPost(_ context.Context, _, _ string) error { + s.calls++ + return s.err +} + +// ledgerRow is one live reservation. +type ledgerRow struct { + id int64 + cmd ReserveSubmissionCommand + at time.Time +} + +// stubLedger models post_submissions in memory: the same unique key, the same +// rolling-window count, and Release genuinely removing the row. +type stubLedger struct { + now Clock + + rows []ledgerRow + nextID int64 + + // reserveErr and countErr force the infrastructure-failure paths. A + // duplicate is NOT set this way — it emerges from the unique key, like it + // does in Postgres. + reserveErr error + countErr error + + reserveCalls []ReserveSubmissionCommand + releaseCalls []SubmissionReservation +} + +func (l *stubLedger) Reserve(_ context.Context, cmd ReserveSubmissionCommand) (SubmissionReservation, error) { + l.reserveCalls = append(l.reserveCalls, cmd) + if l.reserveErr != nil { + return SubmissionReservation{}, l.reserveErr + } + for _, row := range l.rows { + if row.cmd == cmd { + return SubmissionReservation{}, ErrDuplicateSubmission + } + } + l.nextID++ + l.rows = append(l.rows, ledgerRow{id: l.nextID, cmd: cmd, at: l.now()}) + return SubmissionReservation{ID: l.nextID}, nil +} + +func (l *stubLedger) Release(_ context.Context, reservation SubmissionReservation) error { + l.releaseCalls = append(l.releaseCalls, reservation) + kept := l.rows[:0] + for _, row := range l.rows { + if row.id != reservation.ID { + kept = append(kept, row) + } + } + l.rows = kept + return nil +} + +func (l *stubLedger) CountSince(_ context.Context, authorDID, communityDID string, since time.Time) (int, error) { + if l.countErr != nil { + return 0, l.countErr + } + count := 0 + for _, row := range l.rows { + if row.cmd.AuthorDID == authorDID && row.cmd.CommunityDID == communityDID && !row.at.Before(since) { + count++ + } + } + return count, nil +} + +// liveRows is what the ledger holds after the decision — the assertion behind +// "a refusal consumes no quota". +func (l *stubLedger) liveRows() int { return len(l.rows) } + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +// admitHarness is the default world: a public community, an author with no +// membership row, an authorized aggregator, an empty ledger, and a clock that +// only moves when a test moves it. +type admitHarness struct { + communities *stubCommunities + bans *stubBans + aggregators *stubAggregatorAuthorizer + ledger *stubLedger + limits SubmissionLimits + now time.Time +} + +func newAdmitHarness() *admitHarness { + h := &admitHarness{ + communities: &stubCommunities{community: &communities.Community{ + DID: admitCommunityDID, + Handle: admitCommunityHandle, + Visibility: "public", + }}, + bans: &stubBans{}, + aggregators: &stubAggregatorAuthorizer{}, + limits: SubmissionLimits{ + MaxPerAuthorPerCommunity: 3, + Window: time.Hour, + DedupeWindow: time.Hour, + }, + now: time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC), + } + h.ledger = &stubLedger{now: h.clock()} + return h +} + +// clock hands out a Clock that reads the harness's mutable instant, so +// advancing time after the ledger was built still moves the ledger's clock. +func (h *admitHarness) clock() Clock { + return func() time.Time { return h.now } +} + +func (h *admitHarness) advance(d time.Duration) { h.now = h.now.Add(d) } + +func (h *admitHarness) deps() admissionDeps { + return admissionDeps{ + communities: h.communities, + bans: h.bans, + aggregators: h.aggregators, + ledger: h.ledger, + limits: h.limits, + now: h.clock(), + } +} + +// admit runs the decision for a user submitting `fingerprint`. +func (h *admitHarness) admit(t *testing.T, actor ActorClass, fingerprint string) (AdmissionDecision, error) { + t.Helper() + authorDID := admitAuthorDID + if actor != ActorUser { + authorDID = admitAggregatorDID + } + return admitPost(context.Background(), h.deps(), AdmissionRequest{ + Actor: actor, + AuthorDID: authorDID, + Community: admitCommunityHandle, + Fingerprint: fingerprint, + }) +} + +// banned is a membership row with the ban flag set. +func banned() *communities.Membership { + return &communities.Membership{ + UserDID: admitAuthorDID, + CommunityDID: admitCommunityDID, + IsBanned: true, + } +} + +// --------------------------------------------------------------------------- +// The matrix +// --------------------------------------------------------------------------- + +func TestAdmitPost_DecisionMatrix(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + actor ActorClass + setup func(*admitHarness) + wantCode DecisionCode + why string + }{ + { + name: "a community nobody has indexed", + actor: ActorUser, + setup: func(h *admitHarness) { + h.communities.resolveErr = communities.ErrCommunityNotFound + }, + wantCode: DecisionCommunityNotFound, + why: "nothing else can be evaluated against a community that does not exist", + }, + { + name: "an identifier that resolves to a community the index has since lost", + actor: ActorUser, + setup: func(h *admitHarness) { + h.communities.getErr = communities.ErrCommunityNotFound + }, + wantCode: DecisionCommunityNotFound, + why: "resolution and fetch are two lookups, and either failing to find it is the same answer to the client", + }, + { + name: "a regular user submitting to a private community", + actor: ActorUser, + setup: func(h *admitHarness) { + h.communities.community.Visibility = "private" + }, + wantCode: DecisionCommunityPrivate, + why: "Alpha admits public and unlisted only; membership for private communities is Beta", + }, + { + name: "an unlisted community is not a private one", + actor: ActorUser, + setup: func(h *admitHarness) { + h.communities.community.Visibility = "unlisted" + }, + wantCode: "", + why: "unlisted means undiscoverable, not closed — only 'private' blocks a submission", + }, + { + name: "a BANNED user of a PRIVATE community", + actor: ActorUser, + setup: func(h *admitHarness) { + h.communities.community.Visibility = "private" + h.bans.membership = banned() + }, + wantCode: DecisionCommunityPrivate, + why: "a ban must not be disclosed through a privacy wall: answering author-banned would " + + "confirm to an outsider both that the community exists and that a moderator has acted on them", + }, + { + name: "a banned member of a public community", + actor: ActorUser, + setup: func(h *admitHarness) { + h.bans.membership = banned() + }, + wantCode: DecisionAuthorBanned, + why: "the check §4.1 admits does not exist yet, and this is it", + }, + { + name: "a member in good standing", + actor: ActorUser, + setup: func(h *admitHarness) { + h.bans.membership = &communities.Membership{ + UserDID: admitAuthorDID, CommunityDID: admitCommunityDID, IsBanned: false, + } + }, + wantCode: "", + why: "a membership row is not itself a refusal", + }, + { + name: "a non-member of a public community", + actor: ActorUser, + setup: func(h *admitHarness) { h.bans.membership = nil }, + wantCode: "", + why: "ErrMembershipNotFound is a VALUE meaning 'not banned' — posting in a public " + + "community has never required joining it, so the absent row is the common case", + }, + { + name: "a registered aggregator the community never authorized", + actor: ActorRegisteredAggregator, + setup: func(h *admitHarness) { + h.aggregators.err = aggregators.ErrNotAuthorized + }, + wantCode: DecisionAggregatorNotAuthorized, + why: "existing semantics: being registered with the instance is not permission to write anywhere", + }, + { + name: "a registered aggregator over its OWN hourly quota", + actor: ActorRegisteredAggregator, + setup: func(h *admitHarness) { + h.aggregators.err = aggregators.ErrRateLimitExceeded + }, + wantCode: DecisionAggregatorNotAuthorized, + why: "the aggregator limiter answers through the same call; the sentinel it carries is what tells 403 from 429", + }, + { + name: "a registered aggregator submitting into a PRIVATE community", + actor: ActorRegisteredAggregator, + setup: func(h *admitHarness) { + h.communities.community.Visibility = "private" + }, + wantCode: "", + why: "aggregators are authorized services rather than members, so visibility says nothing " + + "about them — this is today's behaviour and the extraction must not change it", + }, + { + name: "a trusted aggregator submitting into a PRIVATE community it is banned from", + actor: ActorTrustedAggregator, + setup: func(h *admitHarness) { + h.communities.community.Visibility = "private" + h.bans.membership = banned() + h.aggregators.err = aggregators.ErrNotAuthorized + }, + wantCode: "", + why: "a trusted aggregator skips visibility, ban and authorization — all three, deliberately", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + tc.setup(h) + + decision, err := h.admit(t, tc.actor, "fingerprint-1") + require.NoErrorf(t, err, "a policy refusal is a decision, not an error: %s", tc.why) + + assert.Equalf(t, tc.wantCode, decision.Code, "%s", tc.why) + assert.Equalf(t, tc.wantCode == "", decision.Admitted(), + "Admitted() must agree with the code it is derived from") + + if decision.Admitted() { + require.NotNil(t, decision.Community, + "an admission carries the resolved community so CreatePost does not fetch it a second time") + assert.Equal(t, admitCommunityDID, decision.Community.DID) + require.NotNil(t, decision.Reservation, + "an admission carries the ledger row it reserved, so a failed PDS write can release it") + assert.Equal(t, 1, h.ledger.liveRows()) + return + } + + // The other half of every refusal: it cost nothing. + assert.Zerof(t, h.ledger.liveRows(), + "a refused submission left a ledger row behind, so it burned quota it was never granted") + }) + } +} + +// --------------------------------------------------------------------------- +// Check order +// --------------------------------------------------------------------------- + +// The nondisclosure rule is not just about the CODE returned — the ban lookup +// must not run at all. A private community that queried moderation state before +// refusing would still leak through timing, and would make a banned outsider's +// probe indistinguishable from a member's in the logs. +func TestAdmitPost_PrivateCommunityNeverConsultsModerationState(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.communities.community.Visibility = "private" + h.bans.membership = banned() + + decision, err := h.admit(t, ActorUser, "probe") + require.NoError(t, err) + + require.Equal(t, DecisionCommunityPrivate, decision.Code) + assert.Zero(t, h.bans.calls, + "the privacy wall must refuse before moderation state is read, not after") +} + +// A community that does not resolve short-circuits everything downstream. An +// implementation that gathered every input before deciding would issue a ban +// lookup, an authorization check and a ledger insert against a community DID it +// had just failed to find. +func TestAdmitPost_AnAbsentCommunityStopsEveryLaterCheck(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.communities.resolveErr = communities.ErrCommunityNotFound + + decision, err := h.admit(t, ActorUser, "probe") + require.NoError(t, err) + require.Equal(t, DecisionCommunityNotFound, decision.Code) + + assert.Zero(t, h.bans.calls, "a ban lookup ran against a community that does not exist") + assert.Zero(t, h.aggregators.calls, "an authorization check ran against a community that does not exist") + assert.Empty(t, h.ledger.reserveCalls, "a ledger row was reserved against a community that does not exist") +} + +// The ban lookup is scoped to the RESOLVED community, not to whatever +// at-identifier the client happened to send. Handles are mutable; a ban keyed +// by handle stops applying the moment a community renames itself. +func TestAdmitPost_BanIsLookedUpByResolvedDID(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + _, err := h.admit(t, ActorUser, "probe") + require.NoError(t, err) + + require.Equal(t, 1, h.bans.calls) + assert.Equal(t, admitCommunityDID, h.bans.lastIdentifier, + "the ban must be looked up against the resolved DID: a handle is mutable, and a ban keyed to one stops applying at rename") + assert.Equal(t, admitAuthorDID, h.bans.lastAuthorDID) +} + +// Neither aggregator class is subject to the ban lookup. Registered and trusted +// aggregators are services rather than members; there is no membership row to +// find, and asking for one on every syndicated item is a query per post for an +// answer that is structurally always the same. +func TestAdmitPost_AggregatorsAreNotBanChecked(t *testing.T) { + t.Parallel() + + for _, actor := range []ActorClass{ActorRegisteredAggregator, ActorTrustedAggregator} { + t.Run(string(actor), func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.bans.membership = banned() + + decision, err := h.admit(t, actor, "syndicated") + require.NoError(t, err) + + assert.True(t, decision.Admitted(), "an aggregator was refused by a membership rule that does not apply to it") + assert.Zero(t, h.bans.calls, "the ban lookup ran for an actor class that has no membership") + }) + } +} + +// Only registered aggregators meet the authorization check. A trusted one is +// authorized by configuration, and a regular user has no aggregator identity to +// check at all. +func TestAdmitPost_OnlyRegisteredAggregatorsAreAuthorizationChecked(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + actor ActorClass + wantCalls int + }{ + {ActorUser, 0}, + {ActorRegisteredAggregator, 1}, + {ActorTrustedAggregator, 0}, + } { + t.Run(string(tc.actor), func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + _, err := h.admit(t, tc.actor, "item") + require.NoError(t, err) + + assert.Equal(t, tc.wantCalls, h.aggregators.calls) + }) + } +} + +// An aggregator refusal must carry the aggregators-package sentinel that +// caused it. +// +// The two refusals share one DecisionCode but mean opposite things to a machine +// client: 403 says stop asking, 429 says ask later. The API boundary tells them +// apart with aggregators.IsUnauthorized and aggregators.IsRateLimited over the +// error CreatePost returns (internal/api/handlers/post/errors.go), so the +// sentinel has to survive the decision. Collapsed into a bare code, a +// well-behaved aggregator would retry a permanent refusal forever. +func TestAdmitPost_AnAggregatorRefusalKeepsItsSentinel(t *testing.T) { + t.Parallel() + + for _, sentinel := range []error{aggregators.ErrNotAuthorized, aggregators.ErrRateLimitExceeded} { + t.Run(sentinel.Error(), func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.aggregators.err = fmt.Errorf("aggregators: %w", sentinel) + + decision, err := h.admit(t, ActorRegisteredAggregator, "item") + require.NoError(t, err) + require.Equal(t, DecisionAggregatorNotAuthorized, decision.Code) + + require.NotNil(t, decision.Cause, + "the refusal dropped the aggregator sentinel, so the boundary cannot tell 403 from 429") + assert.ErrorIs(t, decision.Cause, sentinel) + }) + } +} + +// --------------------------------------------------------------------------- +// Failing closed +// --------------------------------------------------------------------------- + +// A ban lookup that fails for any reason OTHER than "no such membership" must +// fail the request. +// +// This is the single most consequential line in the whole decision. Treating an +// unreachable database as "not banned" would turn a Postgres blip into a global +// unban for its duration — every banned author in every community able to post +// again, with nothing in the logs but a warning. The safe direction is to +// refuse a submission we cannot evaluate. +func TestAdmitPost_ABanLookupFailureFailsTheRequest(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.bans.err = errors.New("dial tcp 10.0.0.4:5432: connect: connection refused") + + decision, err := h.admit(t, ActorUser, "probe") + + require.Error(t, err, "an unevaluable ban check must fail the request, never fall through to admitted") + assert.False(t, decision.Admitted()) + assert.Empty(t, h.ledger.reserveCalls, "a submission we could not evaluate must not reserve quota") +} + +// The infrastructure failures that are NOT policy answers. Each is a decision +// that could not be made, which is a different thing from a refusal — and a +// caller that mapped them to a 4xx would tell a client its perfectly good +// request was its own fault. +func TestAdmitPost_InfrastructureFailuresAreErrorsNotRefusals(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + setup func(*admitHarness) + }{ + { + name: "the community index is unreachable", + setup: func(h *admitHarness) { h.communities.resolveErr = errors.New("connection reset by peer") }, + }, + { + name: "the community fetch fails after resolution succeeded", + setup: func(h *admitHarness) { h.communities.getErr = errors.New("connection reset by peer") }, + }, + { + name: "the ledger insert fails for a reason that is not a duplicate", + setup: func(h *admitHarness) { h.ledger.reserveErr = errors.New("deadlock detected") }, + }, + { + name: "the quota count fails", + setup: func(h *admitHarness) { h.ledger.countErr = errors.New("statement timeout") }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + tc.setup(h) + + decision, err := h.admit(t, ActorUser, "probe") + require.Error(t, err) + assert.False(t, decision.Admitted(), + "a decision that could not be made must not read as an admission") + assert.Emptyf(t, decision.Code, + "an infrastructure failure must not be dressed up as a policy code (%q); the client would be told to stop retrying something that will work in a second", decision.Code) + }) + } +} + +// A malformed community identifier is a client error, and it has to stay one: +// the API boundary turns a validation error into a 400 naming the bad field, +// while an unclassified error becomes an opaque 500. +func TestAdmitPost_AMalformedIdentifierStaysAValidationError(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.communities.resolveErr = communities.NewValidationError("community", "handle must start with !") + + _, err := h.admit(t, ActorUser, "probe") + require.Error(t, err) + assert.True(t, IsValidationError(err), + "a malformed identifier must reach the boundary as a validation error, not as an unclassified failure: %v", err) +} + +// A quota refusal must leave the ledger exactly as it found it. The reservation +// is inserted before the count is taken — that is what closes the concurrent +// double-tap — so the refusal path is responsible for taking it back out. +// Without that, an author who kept retrying past their limit would extend their +// own lockout with every attempt. +func TestAdmitPost_ARateLimitRefusalReleasesItsOwnReservation(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + for i := 0; i < h.limits.MaxPerAuthorPerCommunity; i++ { + decision, err := h.admit(t, ActorUser, fmt.Sprintf("inside-quota-%d", i)) + require.NoError(t, err) + require.Truef(t, decision.Admitted(), "submission %d of %d was refused inside the quota with %q", + i+1, h.limits.MaxPerAuthorPerCommunity, decision.Code) + } + + decision, err := h.admit(t, ActorUser, "one-too-many") + require.NoError(t, err) + require.Equal(t, DecisionRateLimitExceeded, decision.Code) + + assert.Equal(t, h.limits.MaxPerAuthorPerCommunity, h.ledger.liveRows(), + "the refused submission left its reservation on the ledger, so retrying extends the author's own lockout") + assert.NotEmpty(t, h.ledger.releaseCalls, + "the rate-limit path must release the reservation it took before counting") +} + +// --------------------------------------------------------------------------- +// Quota +// --------------------------------------------------------------------------- + +// The boundary itself: N admitted, N+1 refused. Asserting only "the fourth +// fails" would pass against an implementation that refused the third too. +func TestAdmitPost_ExactlyTheLimitIsAdmitted(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.MaxPerAuthorPerCommunity = 5 + + for i := 0; i < 5; i++ { + decision, err := h.admit(t, ActorUser, fmt.Sprintf("item-%d", i)) + require.NoError(t, err) + assert.Truef(t, decision.Admitted(), + "submission %d of 5 was refused with %q, so the limit is being applied one short", i+1, decision.Code) + } + + decision, err := h.admit(t, ActorUser, "item-5") + require.NoError(t, err) + assert.Equal(t, DecisionRateLimitExceeded, decision.Code, + "the sixth submission must be refused, or the limit is being applied one too generously") +} + +// The window is ROLLING, not a bucket that empties on the hour: rows age out +// individually, so an author who filled their quota gets one slot back exactly +// one window after the submission that took it — not all of them at a +// boundary, which would let them spend a double quota either side of it. +func TestAdmitPost_QuotaRecoversAsTheWindowRolls(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.MaxPerAuthorPerCommunity = 2 + h.limits.Window = time.Hour + + first, err := h.admit(t, ActorUser, "first") + require.NoError(t, err) + require.True(t, first.Admitted()) + + h.advance(30 * time.Minute) + second, err := h.admit(t, ActorUser, "second") + require.NoError(t, err) + require.True(t, second.Admitted()) + + third, err := h.admit(t, ActorUser, "third") + require.NoError(t, err) + require.Equal(t, DecisionRateLimitExceeded, third.Code, "the quota is two and both are inside the window") + + // Cross the window relative to the FIRST submission only. One slot frees; + // the second submission is still 30 minutes inside the window. + h.advance(31 * time.Minute) + fourth, err := h.admit(t, ActorUser, "fourth") + require.NoError(t, err) + assert.True(t, fourth.Admitted(), + "the first submission has aged out of the rolling window, so its slot must be available again") + + fifth, err := h.admit(t, ActorUser, "fifth") + require.NoError(t, err) + assert.Equal(t, DecisionRateLimitExceeded, fifth.Code, + "only ONE slot aged out; a window that emptied wholesale would admit this too") +} + +// Neither aggregator class is metered by the new per-author limit. +// +// A trusted aggregator has no submission limit today and inventing one here +// would be a production behaviour change smuggled in under a refactor. A +// registered one is already metered by its own hourly quota inside +// ValidateAggregatorPost, and counting it twice would silently halve the +// throughput every authorized aggregator was granted. +func TestAdmitPost_AggregatorsAreNotMeteredByTheNewLimit(t *testing.T) { + t.Parallel() + + for _, actor := range []ActorClass{ActorRegisteredAggregator, ActorTrustedAggregator} { + t.Run(string(actor), func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.MaxPerAuthorPerCommunity = 2 + + for i := 0; i < 5; i++ { + decision, err := h.admit(t, actor, fmt.Sprintf("syndicated-%d", i)) + require.NoError(t, err) + assert.Truef(t, decision.Admitted(), + "item %d was refused with %q; %s is not subject to the per-author limit", i+1, decision.Code, actor) + } + }) + } +} + +// --------------------------------------------------------------------------- +// Dedupe +// --------------------------------------------------------------------------- + +// Every actor class deduplicates. An RSS aggregator re-polling a feed and +// resubmitting an identical item is the canonical case, so exempting trusted +// actors would exempt precisely the traffic this check exists for. +func TestAdmitPost_IdenticalResubmissionIsRefusedForEveryActorClass(t *testing.T) { + t.Parallel() + + for _, actor := range []ActorClass{ActorUser, ActorRegisteredAggregator, ActorTrustedAggregator} { + t.Run(string(actor), func(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + + first, err := h.admit(t, actor, "the same item") + require.NoError(t, err) + require.True(t, first.Admitted()) + + second, err := h.admit(t, actor, "the same item") + require.NoError(t, err) + assert.Equal(t, DecisionDuplicateSubmission, second.Code, + "an identical resubmission must be refused as a repeat") + assert.Equal(t, 1, h.ledger.liveRows(), + "the refused duplicate must not add a second row") + }) + } +} + +// Dedupe runs BEFORE the rate limit, and this is the case that tells them +// apart: an author already at their quota who retries something they have +// already sent. If the order were reversed, a client whose response was lost +// would be told to slow down when what actually happened is that its post +// already exists — and its retry would have burned a quota slot for a post it +// did not make. +func TestAdmitPost_DedupeAnswersBeforeTheQuotaDoes(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.MaxPerAuthorPerCommunity = 2 + + for i := 0; i < 2; i++ { + decision, err := h.admit(t, ActorUser, fmt.Sprintf("item-%d", i)) + require.NoError(t, err) + require.True(t, decision.Admitted()) + } + + // At quota AND a repeat. Both refusals apply; dedupe is the honest one. + decision, err := h.admit(t, ActorUser, "item-0") + require.NoError(t, err) + assert.Equal(t, DecisionDuplicateSubmission, decision.Code, + "a retry of an already-accepted submission must be reported as a repeat, not as a quota breach") +} + +// Dedupe expires. The ledger's unique key is scoped to a window bucket +// precisely so that "do not accept the same thing twice right now" does not +// silently become "this author may never post this content again". +func TestAdmitPost_DedupeExpiresWithItsWindow(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.DedupeWindow = time.Hour + h.limits.MaxPerAuthorPerCommunity = 10 + + first, err := h.admit(t, ActorUser, "a link worth reposting") + require.NoError(t, err) + require.True(t, first.Admitted()) + + h.advance(2 * time.Hour) + + second, err := h.admit(t, ActorUser, "a link worth reposting") + require.NoError(t, err) + assert.True(t, second.Admitted(), + "identical content a dedupe window later is a repost, not a duplicate submission") + + require.Len(t, h.ledger.reserveCalls, 2) + assert.NotEqual(t, h.ledger.reserveCalls[0].DedupeBucket, h.ledger.reserveCalls[1].DedupeBucket, + "the two submissions must fall in different dedupe buckets, or the unique key would still collide") +} + +// Two submissions inside one window share a bucket — the other half of the +// property above, and the one that actually makes the unique key bite. +func TestAdmitPost_SubmissionsInsideOneWindowShareABucket(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.DedupeWindow = time.Hour + + _, err := h.admit(t, ActorUser, "first") + require.NoError(t, err) + + h.advance(5 * time.Minute) + _, err = h.admit(t, ActorUser, "second") + require.NoError(t, err) + + require.Len(t, h.ledger.reserveCalls, 2) + assert.Equal(t, h.ledger.reserveCalls[0].DedupeBucket, h.ledger.reserveCalls[1].DedupeBucket, + "submissions five minutes apart in an hourly window must share a bucket") +} + +// What the ledger is actually asked to store. The fingerprint is the client's +// content hash and must arrive unmodified, and the community must be the +// resolved DID for the same reason the ban lookup is. +func TestAdmitPost_TheReservationDescribesTheSubmission(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + _, err := h.admit(t, ActorUser, "sha256-of-the-canonical-record") + require.NoError(t, err) + + require.Len(t, h.ledger.reserveCalls, 1) + reserved := h.ledger.reserveCalls[0] + assert.Equal(t, admitAuthorDID, reserved.AuthorDID) + assert.Equal(t, admitCommunityDID, reserved.CommunityDID, + "the ledger is keyed by resolved DID; a handle would let a rename reset both the quota and the dedupe key") + assert.Equal(t, "sha256-of-the-canonical-record", reserved.Fingerprint) +} + +// Quota and dedupe are per (author, community): a user at their limit in one +// community must still be able to post in another. A limit that leaked across +// communities would make one busy community silence its author everywhere. +func TestAdmitPost_QuotaIsScopedToOneCommunity(t *testing.T) { + t.Parallel() + + h := newAdmitHarness() + h.limits.MaxPerAuthorPerCommunity = 1 + + first, err := h.admit(t, ActorUser, "item") + require.NoError(t, err) + require.True(t, first.Admitted()) + + refused, err := h.admit(t, ActorUser, "another item") + require.NoError(t, err) + require.Equal(t, DecisionRateLimitExceeded, refused.Code) + + // The same author, the same content, a different community. + h.communities.community = &communities.Community{ + DID: "did:plc:dddddddddddddddddddddddd", + Handle: "!woodworking.communities.coves.social", + Visibility: "public", + } + + elsewhere, err := h.admit(t, ActorUser, "item") + require.NoError(t, err) + assert.True(t, elsewhere.Admitted(), + "the quota is per community; being at the limit in one must not silence the author in another") +} + +// --------------------------------------------------------------------------- +// The sentinel's wording +// --------------------------------------------------------------------------- + +// IsConflict classifies by substring, so a duplicate SUBMISSION worded like a +// duplicate KEY would be misread as an indexing conflict — a post refused at +// the admission gate reported to its author as one that already exists. +func TestErrDuplicateSubmissionIsNotAStorageConflict(t *testing.T) { + t.Parallel() + + assert.False(t, IsConflict(ErrDuplicateSubmission), + "ErrDuplicateSubmission's message collides with IsConflict's substring match (%q); reword the sentinel", + ErrDuplicateSubmission.Error()) + assert.False(t, IsConflict(fmt.Errorf("createPost: %w", ErrDuplicateSubmission)), + "the wrapped form must not be misclassified either — that is the shape the boundary actually sees") +} + +// --------------------------------------------------------------------------- +// Fingerprint +// --------------------------------------------------------------------------- + +// The fingerprint is what makes two submissions "identical". createdAt is +// stamped per attempt, so including it would make every retry look new and +// dedupe would never fire; everything a moderator would judge must be included, +// or two genuinely different posts would collide and the second would be +// refused as a repeat of the first. +func TestSubmissionFingerprint(t *testing.T) { + t.Parallel() + + base := func() PostRecord { + title, content := "A title", "Some body text" + return PostRecord{ + Type: "social.coves.community.postv2", + Community: admitCommunityDID, + Author: admitAuthorDID, + Title: &title, + Content: &content, + CreatedAt: "2026-08-01T12:00:00Z", + } + } + + t.Run("createdAt is excluded", func(t *testing.T) { + t.Parallel() + + later := base() + later.CreatedAt = "2026-08-01T12:00:09Z" + + assert.Equal(t, submissionFingerprint(base()), submissionFingerprint(later), + "the server stamps createdAt per attempt, so a fingerprint that included it would never match a retry") + }) + + t.Run("a non-empty fingerprint", func(t *testing.T) { + t.Parallel() + + assert.NotEmpty(t, submissionFingerprint(base()), + "an empty fingerprint would make every submission collide with every other") + }) + + for _, tc := range []struct { + field string + mutate func(*PostRecord) + }{ + {"title", func(r *PostRecord) { title := "A different title"; r.Title = &title }}, + {"content", func(r *PostRecord) { content := "Different body text"; r.Content = &content }}, + {"community", func(r *PostRecord) { r.Community = "did:plc:dddddddddddddddddddddddd" }}, + {"author", func(r *PostRecord) { r.Author = "did:plc:eeeeeeeeeeeeeeeeeeeeeeee" }}, + {"embed", func(r *PostRecord) { + r.Embed = map[string]interface{}{"$type": "social.coves.embed.external"} + }}, + } { + t.Run("a different "+tc.field+" is a different submission", func(t *testing.T) { + t.Parallel() + + changed := base() + tc.mutate(&changed) + assert.NotEqual(t, submissionFingerprint(base()), submissionFingerprint(changed), + "two posts differing in %s would collide, and the second would be refused as a repeat of the first", tc.field) + }) + } +} diff --git a/internal/core/posts/errors.go b/internal/core/posts/errors.go index 8d2dffd..713d9de 100644 --- a/internal/core/posts/errors.go +++ b/internal/core/posts/errors.go @@ -34,6 +34,20 @@ var ( // ErrActorNotFound is returned when the requested actor does not exist ErrActorNotFound = errors.New("actor not found") + + // ErrDuplicateSubmission is returned when an author resubmits content + // identical to something already on the submission ledger for the current + // dedupe window (PRD_AUTHOR_OWNED_POSTS.md §8). + // + // THE WORDING IS LOAD-BEARING. IsConflict below classifies an error by + // looking for "duplicate key", "already exists" or "already indexed" in its + // text, because a genuine index conflict arrives from the driver as a + // string rather than as a typed error. This sentinel must therefore avoid + // all three phrasings: a duplicate SUBMISSION is a client being refused at + // the admission gate, while a conflict is the indexer meeting a record it + // already has, and collapsing them would let a refused post be reported as + // successfully indexed. + ErrDuplicateSubmission = errors.New("an identical submission from this author to this community was refused as a repeat") ) // ValidationError is the shared validation error type. It is aliased rather diff --git a/internal/core/posts/service.go b/internal/core/posts/service.go index 877a0c9..99d0870 100644 --- a/internal/core/posts/service.go +++ b/internal/core/posts/service.go @@ -34,6 +34,7 @@ type postService struct { unfurlService unfurl.Service blueskyService blueskypost.Service blockChecker BlockChecker + admission *AdmissionPolicy pdsURL string } diff --git a/internal/core/posts/service_admission_test.go b/internal/core/posts/service_admission_test.go new file mode 100644 index 0000000..6b43841 --- /dev/null +++ b/internal/core/posts/service_admission_test.go @@ -0,0 +1,394 @@ +//go:build integration + +package posts_test + +import ( + "context" + "database/sql" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "Coves/internal/api/middleware" + "Coves/internal/core/communities" + "Coves/internal/core/posts" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The outer contract for admission: what CreatePost does when the admission +// policy of docs/PRD_AUTHOR_OWNED_POSTS.md §4.1/§8 refuses a submission. +// +// The decision itself is covered at width in admit_matrix_test.go against +// fakes. What is unproven without these is the WIRING, and the wiring is where +// this kind of check historically goes wrong: +// +// - that CreatePost consults the policy AT ALL. §4.1 corrects rev 1 of the +// spec on exactly this point — the service docstring has claimed +// "membership/ban validation" since the beginning and no ban lookup has +// ever existed on the write path. +// - that it consults it BEFORE the PDS write, so a refusal leaves no record +// in a community that refused it. +// - that the ledger rows the quota is counted against are the ones CreatePost +// itself writes. Like the aggregator quota (service_aggregator_test.go), +// the producer and the consumer of that counter are the same code path, and +// a service that never wrote them would pass every unit test and never rate +// limit anything. +// - that a failed PDS write RELEASES the row it reserved. This is the one +// behaviour no fake can prove, because it is about what survives in the +// database after a write that did not happen. +// +// ON SEEDING is_banned DIRECTLY. Nothing in production writes +// community_memberships at all today — not CreateMembership, not +// UpdateMembership; both are reachable only from tests. There is no ban +// endpoint, no moderation consumer, and no firehose path that sets the column. +// So these tests seed it through the repository, and that is an honest +// admission of an incomplete feature rather than a shortcut: §4.1 specifies the +// ban LOOKUP, and the record type that will eventually write it +// (social.coves.moderation.ban) is not this task's work. When it lands, this +// seeding becomes the moderation path and these assertions do not change. + +// admissionFixture is the post service wired with the §8 admission policy over +// real Postgres, a real PDS, and a clock the test controls. +type admissionFixture struct { + base *postFixture + service posts.Service + repo communities.Repository + clock *testClock + limits posts.SubmissionLimits +} + +// testClock is the injected Clock. Time moves only when a test moves it, which +// is what lets a rolling window be crossed without a sleep — docs/ +// TEST_ARCHITECTURE.md §3.3 bans the alternative outright. +// +// It is mutex-guarded because CreatePost may read it from more than one +// goroutine, and the race detector is on in CI. +type testClock struct { + mu sync.Mutex + now time.Time +} + +func (c *testClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *testClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(d) +} + +// newAdmissionFixture provisions a community on the test PDS and points a +// policy-wired post service at it. +// +// The limit is deliberately small. Each admitted submission costs a real +// createRecord against a real PDS, and the assertion that matters is the +// BOUNDARY — N admitted, N+1 refused — which three proves exactly as well as +// three hundred would. +func newAdmissionFixture(t *testing.T) *admissionFixture { + t.Helper() + + base := newPostFixture(t) + limits := posts.SubmissionLimits{ + MaxPerAuthorPerCommunity: 3, + Window: time.Hour, + DedupeWindow: time.Hour, + } + + // Anchored at the real present rather than at a fixed date: post_submissions + // stamps created_at server-side (NOW()), while the rolling-window query is + // computed from THIS clock, so the two have to agree about roughly when + // "now" is. Advancing forwards is always safe — it ages real rows out of the + // window, which is the direction every test here moves. + clock := &testClock{now: time.Now().UTC()} + + return &admissionFixture{ + base: base, + service: posts.NewPostService( + postgres.NewPostRepository(base.db), base.communityService, + nil, nil, nil, nil, base.pds.URL(), + posts.WithAdmissionPolicy(posts.AdmissionPolicy{ + Ledger: postgres.NewSubmissionLedger(base.db), + Bans: base.communityService, + Limits: limits, + Now: clock.Now, + })), + repo: postgres.NewCommunityRepository(base.db), + clock: clock, + limits: limits, + } +} + +// submit posts as the fixture's author. Unlike postFixture.createPost it +// returns the error, because every test here is about a refusal. +func (f *admissionFixture) submit(t *testing.T, communityDID, title string) (*posts.CreatePostResponse, error) { + t.Helper() + + content := "a body that makes this a complete post" + return f.service.CreatePost( + middleware.SetTestUserDID(context.Background(), f.base.author.DID), + posts.CreatePostRequest{ + Community: communityDID, + Title: &title, + Content: &content, + AuthorDID: f.base.author.DID, + }) +} + +// setBanned seeds (or updates) the author's membership of the fixture's +// community with the given ban state. +func (f *admissionFixture) setBanned(t *testing.T, banned bool) { + t.Helper() + + ctx := context.Background() + membership := &communities.Membership{ + UserDID: f.base.author.DID, + CommunityDID: f.base.community.DID, + JoinedAt: time.Now().UTC(), + LastActiveAt: time.Now().UTC(), + IsBanned: banned, + } + + if _, err := f.repo.GetMembership(ctx, f.base.author.DID, f.base.community.DID); err != nil { + require.ErrorIs(t, err, communities.ErrMembershipNotFound) + _, createErr := f.repo.CreateMembership(ctx, membership) + require.NoError(t, createErr) + return + } + + _, err := f.repo.UpdateMembership(ctx, membership) + require.NoError(t, err) +} + +// ledgerRows counts what the submission ledger holds for the author in one +// community — the number the quota is enforced against. +// +// Read with raw SQL rather than through the repository on purpose: the +// repository is the thing under test here, and a test that asked it to report +// its own state would pass against an implementation that recorded nothing. +func (f *admissionFixture) ledgerRows(t *testing.T, communityDID string) int { + t.Helper() + return countSubmissions(t, f.base.db, f.base.author.DID, communityDID) +} + +func countSubmissions(t *testing.T, db *sql.DB, authorDID, communityDID string) int { + t.Helper() + + var count int + require.NoError(t, db.QueryRowContext(context.Background(), ` + SELECT count(*) FROM post_submissions WHERE author_did = $1 AND community_did = $2 + `, authorDID, communityDID).Scan(&count), + "the post_submissions ledger (migration 035) must exist for the quota to be countable") + return count +} + +// anotherCommunity provisions a second community on the same PDS, so that a +// per-community rule can be shown to be per-community. +func (f *admissionFixture) anotherCommunity(t *testing.T) *communities.Community { + t.Helper() + + name := testkit.UniqueIDWithPrefix(t, "ad") + require.LessOrEqualf(t, len("c-"+name), testkit.MaxIDLength, + "the generated community name %q makes a handle label the PDS will refuse", name) + + community, err := f.base.communityService.CreateCommunity(context.Background(), communities.CreateCommunityRequest{ + Name: name, + DisplayName: "Somewhere else", + Description: "a second community, to prove the quota is scoped to one", + Visibility: "public", + CreatedByDID: f.base.author.DID, + }) + require.NoError(t, err) + return community +} + +// --------------------------------------------------------------------------- + +// A ban stops the next post, and lifting it lets the author back in. +// +// Both halves matter. A ban that could not be lifted would be a data-loss bug +// dressed as a moderation feature, and it is the kind that only shows up when a +// moderator tries to undo a mistake. +func TestService_ABannedMemberIsRefusedAndAnUnbanRestoresThem(t *testing.T) { + t.Parallel() + + f := newAdmissionFixture(t) + + // Before the ban, the author is an ordinary poster in a public community. + _, err := f.submit(t, f.base.community.DID, "posted while in good standing") + require.NoError(t, err) + + f.setBanned(t, true) + + _, err = f.submit(t, f.base.community.DID, "posted after the ban") + require.Error(t, err) + assert.ErrorIsf(t, err, posts.ErrBanned, + "the handler maps this sentinel to a 403 Banned, so the sentinel identity is the contract; got: %v", err) + + // Refused means nothing was written and nothing was billed. The check runs + // ahead of the PDS write; an implementation that reordered them would leave + // a banned author's post in the community's repository, from which the + // firehose would index it before anything noticed. + assert.Equal(t, 1, f.ledgerRows(t, f.base.community.DID), + "the refused submission consumed quota it was never granted") + + f.setBanned(t, false) + + _, err = f.submit(t, f.base.community.DID, "posted after the unban") + assert.NoError(t, err, "lifting a ban must take effect on the next post, not on the next restart") +} + +// The per-author quota: N admitted, N+1 refused, and the limit is per +// community. +// +// Asserting only that "the fourth fails" would pass against an implementation +// that refused the third too, which is why every submission inside the quota is +// individually required to succeed. +func TestService_TheAuthorQuotaStopsTheNextSubmission(t *testing.T) { + t.Parallel() + + f := newAdmissionFixture(t) + + for i := 0; i < f.limits.MaxPerAuthorPerCommunity; i++ { + _, err := f.submit(t, f.base.community.DID, fmt.Sprintf("submission %d", i)) + require.NoErrorf(t, err, "submission %d of %d was refused inside the quota", + i+1, f.limits.MaxPerAuthorPerCommunity) + } + require.Equal(t, f.limits.MaxPerAuthorPerCommunity, f.ledgerRows(t, f.base.community.DID)) + + _, err := f.submit(t, f.base.community.DID, "one submission too many") + require.Error(t, err) + assert.ErrorIsf(t, err, posts.ErrRateLimitExceeded, + "the handler maps this to a 429; got: %v", err) + + // A refused submission is not billed, so an author cannot spend past their + // quota by ignoring the error — and, more to the point, cannot extend their + // own lockout by retrying. + assert.Equal(t, f.limits.MaxPerAuthorPerCommunity, f.ledgerRows(t, f.base.community.DID)) + + // The same author, at their limit here, is unaffected there. A quota that + // leaked across communities would let one busy community silence its author + // everywhere on the instance. + elsewhere := f.anotherCommunity(t) + _, err = f.submit(t, elsewhere.DID, "a submission somewhere else") + assert.NoError(t, err, "the quota is per (author, community); being at the limit in one must not close the others") +} + +// An identical resubmission is a repeat, not a new post. +// +// The canonical case is a client that retried after a lost response, and the +// answer has to be distinguishable from a quota breach: 409 tells the client its +// post already exists, 429 tells it to wait. A submission refused as a duplicate +// must also not be billed, or a flaky connection would rate-limit a user who +// posted once. +func TestService_AnIdenticalResubmissionIsRefusedAsADuplicate(t *testing.T) { + t.Parallel() + + f := newAdmissionFixture(t) + + _, err := f.submit(t, f.base.community.DID, "the very same post") + require.NoError(t, err) + + _, err = f.submit(t, f.base.community.DID, "the very same post") + require.Error(t, err) + assert.ErrorIsf(t, err, posts.ErrDuplicateSubmission, + "the handler maps this to a 409 DuplicateSubmission; got: %v", err) + + assert.Equal(t, 1, f.ledgerRows(t, f.base.community.DID), + "the duplicate must be refused without adding a second ledger row") + + // And a genuinely different post from the same author still goes through: + // the refusal is about identical content, not about having posted recently. + _, err = f.submit(t, f.base.community.DID, "a different post entirely") + assert.NoError(t, err) +} + +// A PDS write that fails must give the reservation back. +// +// The ledger row goes in BEFORE the record is written — that ordering is what +// closes the concurrent double-tap, since the unique constraint is the only +// arbiter that both racing requests can agree on. The cost of that choice is +// that the failure path owes the author their slot back. If it does not pay, +// every PDS hiccup permanently consumes one submission from the author's quota +// AND blocks them from retrying the same content at all, which turns a +// transient outage into a per-user lockout that outlives it. +func TestService_AFailedPDSWriteReleasesTheReservation(t *testing.T) { + t.Parallel() + + f := newAdmissionFixture(t) + + // A PDS that refuses every write. Pointing the community's stored pds_url at + // it is how the failure is injected: createPostOnPDS reads the URL off the + // community row it just fetched (service.go, "each community can be hosted + // on a different PDS instance"), so this is the real write path failing for + // a real reason rather than a stubbed-out client. + broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, `{"error":"InternalServerError"}`, http.StatusInternalServerError) + })) + t.Cleanup(broken.Close) + + healthyURL := communityPDSURL(t, f.base.db, f.base.community.DID) + setCommunityPDSURL(t, f.base.db, f.base.community.DID, broken.URL) + + const repeatable = "a post whose write will fail the first time" + _, err := f.submit(t, f.base.community.DID, repeatable) + require.Error(t, err, "the PDS refused the write, so CreatePost must report a failure") + + assert.Zerof(t, f.ledgerRows(t, f.base.community.DID), + "the reservation for a post that was never written is still on the ledger: it has burned a quota slot and will refuse the retry as a duplicate") + + setCommunityPDSURL(t, f.base.db, f.base.community.DID, healthyURL) + + // The retry a client would actually send: byte-identical content. It must be + // admitted, which is only possible if the reservation was released. + resp, err := f.submit(t, f.base.community.DID, repeatable) + require.NoErrorf(t, err, "the identical retry after a failed write was refused, so the failure path leaked its reservation") + require.NotEmpty(t, resp.URI) + + assert.Equal(t, 1, f.ledgerRows(t, f.base.community.DID), + "exactly one submission survived: the failed attempt released its row and the retry took a fresh one") + + // The record really is in the community's repo, so "admitted" here means a + // post exists rather than merely that no error came back. + record := f.base.communityAccount(t).GetRecord(t, postCollection, rkeyOf(t, resp.URI)) + assert.Equal(t, f.base.author.DID, record.Value["author"]) +} + +// communityPDSURL reads a community's stored PDS, so a test that repoints it +// can put back what was actually there rather than what it assumed. +func communityPDSURL(t *testing.T, db *sql.DB, communityDID string) string { + t.Helper() + + var pdsURL string + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT pds_url FROM communities WHERE did = $1`, communityDID).Scan(&pdsURL)) + require.NotEmpty(t, pdsURL) + return pdsURL +} + +// setCommunityPDSURL repoints a community at a different PDS. +// +// Written with SQL because there is no service method for it: a community's PDS +// is chosen when it is provisioned and never moves. That is exactly why it is a +// usable seam for an unreachable-PDS test — the value is read fresh on every +// write (EnsureFreshToken re-fetches the row), and nothing caches it. +func setCommunityPDSURL(t *testing.T, db *sql.DB, communityDID, pdsURL string) { + t.Helper() + + result, err := db.ExecContext(context.Background(), + `UPDATE communities SET pds_url = $1 WHERE did = $2`, pdsURL, communityDID) + require.NoError(t, err) + + affected, err := result.RowsAffected() + require.NoError(t, err) + require.EqualValues(t, 1, affected, "no community row was repointed, so the test would prove nothing") +} diff --git a/internal/db/postgres/admission_repo_schema_test.go b/internal/db/postgres/admission_repo_schema_test.go index 2eb7499..498d943 100644 --- a/internal/db/postgres/admission_repo_schema_test.go +++ b/internal/db/postgres/admission_repo_schema_test.go @@ -318,9 +318,13 @@ func TestMigration034_DownRestoresTheAuthorForeignKeyUnvalidated(t *testing.T) { require.NoError(t, err, "with fk_author dropped, a federated author's post must index even though no users row exists for them") - // The expected-version parameter is the tripwire: when migration 035 lands, - // this call fails with the remedy in its message instead of silently - // rolling back 035's Down and leaving 034's untested. + // The expected-version parameter is the tripwire, and it has fired once + // already: migration 035 (post_submissions) now sits on top of 034, so it + // has to come off first. Rolling back explicitly, one asserted step at a + // time, is what keeps the assertions below pointed at 034's Down rather than + // at whatever happens to be newest. + require.EqualValues(t, 35, testkit.MigrateDownOne(t, db, 35), + "035 sits on top of 034 and must be rolled back first; asserting which migration came off is what stops this test drifting onto a newer one") assert.EqualValues(t, 34, testkit.MigrateDownOne(t, db, 34), "this test asserts on 034's Down section; rolling back a different migration would prove nothing about it") @@ -361,7 +365,7 @@ func requireTableExists(t *testing.T, db *sql.DB, table string) { SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = $1 `, table).Scan(&count)) - require.Equalf(t, 1, count, "table %s does not exist; migration 034 has not been written", table) + require.Equalf(t, 1, count, "table %s does not exist; the migration that creates it has not been written", table) } // primaryKeyColumns returns the table's primary key columns in key order. diff --git a/internal/db/postgres/submission_ledger_repo.go b/internal/db/postgres/submission_ledger_repo.go new file mode 100644 index 0000000..648d5dd --- /dev/null +++ b/internal/db/postgres/submission_ledger_repo.go @@ -0,0 +1,39 @@ +package postgres + +import ( + "context" + "database/sql" + "time" + + "Coves/internal/core/posts" +) + +// PostgreSQL storage for the post_submissions ledger (migration 035) — the +// rows that both deduplicate submissions and meter the per-author quota of +// docs/PRD_AUTHOR_OWNED_POSTS.md §8. +// +// It mirrors the aggregator limiter of migration 012: a row per accepted +// submission, a rolling-window COUNT over an index that leads with the pair +// being metered. The difference is that Reserve is written BEFORE the PDS +// write rather than after it, because here the insert is also the dedupe gate +// — see posts.SubmissionLedger for why that ordering is the safe one. +type submissionLedger struct { + db *sql.DB +} + +// NewSubmissionLedger creates the post_submissions repository. +func NewSubmissionLedger(db *sql.DB) posts.SubmissionLedger { + return &submissionLedger{db: db} +} + +func (l *submissionLedger) Reserve(ctx context.Context, cmd posts.ReserveSubmissionCommand) (posts.SubmissionReservation, error) { + return posts.SubmissionReservation{}, nil +} + +func (l *submissionLedger) Release(ctx context.Context, reservation posts.SubmissionReservation) error { + return nil +} + +func (l *submissionLedger) CountSince(ctx context.Context, authorDID, communityDID string, since time.Time) (int, error) { + return 0, nil +} diff --git a/internal/db/postgres/submission_ledger_schema_test.go b/internal/db/postgres/submission_ledger_schema_test.go new file mode 100644 index 0000000..1f1038d --- /dev/null +++ b/internal/db/postgres/submission_ledger_schema_test.go @@ -0,0 +1,270 @@ +//go:build integration + +package postgres + +import ( + "context" + "database/sql" + "fmt" + "strings" + "testing" + + "Coves/tests/fixtures" + "Coves/tests/testkit" + + "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Migration 035's shape, read out of the catalog rather than out of the .sql +// file — the same distinction admission_repo_schema_test.go draws for 034: a +// grep proves the text was written, these prove the database ended up in the +// state the text was supposed to produce. +// +// Three things here are load-bearing and silent when wrong: +// +// - THE UNIQUE CONSTRAINT IS THE DEDUPE GATE. admitPost does not SELECT and +// then INSERT; it INSERTs and reads the unique violation as the answer, +// because the database is the only participant two racing double-taps both +// talk to. Without the constraint, dedupe still "works" under every +// sequential test and silently stops working under concurrency. +// - NO FOREIGN KEYS. Migration 034 dropped posts.fk_author precisely because +// a federated author has no users row (§5.3); a ledger that referenced one +// would make the very submissions 034 exists to admit unrecordable, and the +// refusal would arrive as a dead letter rather than as a decision. +// - THE RATE-LIMIT INDEX. The quota is a COUNT over (author, community) in a +// rolling window, run on the write path of every post. Migration 012 built +// idx_aggregator_posts_rate_limit for the identical query shape; without +// the equivalent here the count degrades to a scan of every submission the +// instance has ever accepted. + +const submissionsTable = "post_submissions" + +func TestSubmissionsTable_Columns(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + + requireTableExists(t, db, submissionsTable) + + type columnShape struct { + dataType string + nullable bool + } + want := map[string]columnShape{ + // The surrogate key exists so a reservation can be RELEASED by identity. + // Releasing by the natural key would work too, right up until the row + // being released is not the one this request inserted. + "id": {"bigint", false}, + "author_did": {"text", false}, + "community_did": {"text", false}, + + // The hash of the canonical record minus createdAt. Text rather than + // bytea so it is greppable in an incident and comparable in psql; the + // column is never interpreted, only equated. + "fingerprint": {"text", false}, + + // The window index the dedupe key is scoped to, derived from the + // application's injected clock. It is an integer rather than a + // timestamp deliberately: a timestamp here invites comparison against + // created_at, and the two come from different clocks — one the app's, + // one the database's. + "dedupe_bucket": {"bigint", false}, + + // Server-stamped, because it is what the rolling window is measured + // against and a client-supplied time would be a client-supplied quota. + "created_at": {"timestamp with time zone", false}, + } + + rows, err := db.QueryContext(ctx, ` + SELECT column_name, data_type, is_nullable, coalesce(column_default, '') + FROM information_schema.columns + WHERE table_schema = current_schema() AND table_name = $1 + `, submissionsTable) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + got := map[string]columnShape{} + defaults := map[string]string{} + for rows.Next() { + var name, dataType, isNullable, columnDefault string + require.NoError(t, rows.Scan(&name, &dataType, &isNullable, &columnDefault)) + got[name] = columnShape{dataType: dataType, nullable: isNullable == "YES"} + defaults[name] = columnDefault + } + require.NoError(t, rows.Err()) + + for name, wantShape := range want { + gotShape, ok := got[name] + if !assert.Truef(t, ok, "%s.%s is missing", submissionsTable, name) { + continue + } + assert.Equalf(t, wantShape.dataType, gotShape.dataType, "%s.%s type", submissionsTable, name) + assert.Equalf(t, wantShape.nullable, gotShape.nullable, "%s.%s nullability", submissionsTable, name) + } + + assert.Containsf(t, strings.ToLower(defaults["created_at"]), "now()", + "created_at must be stamped by the server: the rolling window is measured against it, so a caller that could set it could set its own quota") + + assert.Equal(t, []string{"id"}, primaryKeyColumns(t, db, submissionsTable), + "the primary key is the surrogate id, so a reservation can be released by identity") +} + +// The dedupe gate itself. Both halves are asserted — that the constraint exists +// over exactly the right columns, and that Postgres actually refuses the second +// insert — because a constraint over the wrong column set is present in the +// catalog and useless in practice. +func TestSubmissionsTable_DedupeKeyIsUniqueAndEnforced(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + requireTableExists(t, db, submissionsTable) + + wantKey := []string{"author_did", "community_did", "fingerprint", "dedupe_bucket"} + + t.Run("the constraint exists over the dedupe key", func(t *testing.T) { + found := uniqueKeyColumnSets(t, db, submissionsTable) + var matched bool + for _, columns := range found { + if assert.ObjectsAreEqual(wantKey, columns) { + matched = true + } + } + assert.Truef(t, matched, + "no UNIQUE key over %v; the INSERT is the dedupe gate, so without it two concurrent identical submissions both succeed. Unique keys found: %v", + wantKey, found) + }) + + t.Run("a repeat inside the same bucket is refused", func(t *testing.T) { + author, community := newSubmissionSubject(t) + + require.NoError(t, insertSubmission(t, db, author, community, "fp-repeat", 100)) + assert.Error(t, insertSubmission(t, db, author, community, "fp-repeat", 100), + "an identical submission in the same dedupe window must be refused by the database, not merely by a prior SELECT") + }) + + t.Run("the same content in a later bucket is a repost, not a duplicate", func(t *testing.T) { + author, community := newSubmissionSubject(t) + + require.NoError(t, insertSubmission(t, db, author, community, "fp-later", 100)) + assert.NoError(t, insertSubmission(t, db, author, community, "fp-later", 101), + "the bucket is what makes dedupe expire; without it an author could never repost the same content again") + }) + + t.Run("the key is scoped per author and per community", func(t *testing.T) { + author, community := newSubmissionSubject(t) + otherAuthor, otherCommunity := newSubmissionSubject(t) + + require.NoError(t, insertSubmission(t, db, author, community, "fp-scope", 100)) + assert.NoError(t, insertSubmission(t, db, otherAuthor, community, "fp-scope", 100), + "two authors posting the same link are not duplicates of each other") + assert.NoError(t, insertSubmission(t, db, author, otherCommunity, "fp-scope", 100), + "cross-posting the same content to a second community is not a duplicate") + }) +} + +func TestSubmissionsTable_HasNoForeignKeys(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + requireTableExists(t, db, submissionsTable) + + rows, err := db.QueryContext(ctx, ` + SELECT conname, pg_get_constraintdef(oid) FROM pg_constraint + WHERE conrelid = $1::regclass AND contype = 'f' + `, submissionsTable) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + for rows.Next() { + var name, definition string + require.NoError(t, rows.Scan(&name, &definition)) + assert.Failf(t, "the submission ledger has a foreign key", + "constraint %s: %s — migration 034 dropped posts.fk_author because a federated author has no users row, "+ + "and a community may be one this AppView has not indexed; an FK here turns an ordinary submission into an insert failure", + name, definition) + } + require.NoError(t, rows.Err()) + + // And the behaviour that follows from it: a DID this instance has never + // heard of can still be metered. + unknownAuthor := fixtures.DID(testkit.UniqueID(t)) + unknownCommunity := fixtures.DID(testkit.UniqueID(t)) + assert.NoError(t, insertSubmission(t, db, unknownAuthor, unknownCommunity, "fp-federated", 100), + "a submission from an author with no users row must record; that author is exactly who §5.3 exists for") +} + +func TestSubmissionsTable_RateLimitIndex(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + requireTableExists(t, db, submissionsTable) + + definitions := indexDefinitions(t, db, submissionsTable) + + var matched string + for name, definition := range definitions { + if indexColumns(definition) == "author_did, community_did, created_at" { + matched = name + } + } + assert.NotEmptyf(t, matched, + "no index on (author_did, community_did, created_at); that is the rolling-window quota query, run on the write path of every post — migration 012 built idx_aggregator_posts_rate_limit for the identical shape. Indexes found: %v", + definitions) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +// newSubmissionSubject returns an author and a community that no other subtest +// shares, so subtests of one parallel test cannot collide on the dedupe key. +// Neither needs to exist anywhere else — that is the point of the no-FK rule. +func newSubmissionSubject(t *testing.T) (authorDID, communityDID string) { + t.Helper() + return fixtures.DID(testkit.UniqueID(t)), fixtures.DID(testkit.UniqueID(t)) +} + +func insertSubmission(t *testing.T, db *sql.DB, authorDID, communityDID, fingerprint string, bucket int64) error { + t.Helper() + + _, err := db.ExecContext(context.Background(), fmt.Sprintf(` + INSERT INTO %s (author_did, community_did, fingerprint, dedupe_bucket) + VALUES ($1, $2, $3, $4) + `, submissionsTable), authorDID, communityDID, fingerprint, bucket) + return err +} + +// uniqueKeyColumnSets returns the column sets covered by a UNIQUE constraint or +// a unique index, in key order. +// +// Both spellings are accepted because both enforce the same thing and Postgres +// reports them differently: a table constraint appears in pg_constraint, while +// a bare CREATE UNIQUE INDEX appears only in pg_index. Insisting on one would +// fail a migration that closed the race perfectly well the other way. +func uniqueKeyColumnSets(t *testing.T, db *sql.DB, table string) [][]string { + t.Helper() + + rows, err := db.QueryContext(context.Background(), ` + SELECT array_agg(a.attname ORDER BY k.ord) + FROM pg_index i + JOIN unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord) ON true + JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum + WHERE i.indrelid = $1::regclass AND i.indisunique + GROUP BY i.indexrelid + `, table) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + var sets [][]string + for rows.Next() { + var columns []string + require.NoError(t, rows.Scan(pq.Array(&columns))) + sets = append(sets, columns) + } + require.NoError(t, rows.Err()) + return sets +}