diff --git a/docs/PRD_AUTHOR_OWNED_POSTS.md b/docs/PRD_AUTHOR_OWNED_POSTS.md index df9c759..0a3dcbf 100644 --- a/docs/PRD_AUTHOR_OWNED_POSTS.md +++ b/docs/PRD_AUTHOR_OWNED_POSTS.md @@ -38,7 +38,18 @@ author-supplied created_at, delete-to-evade); per-origin-PDS quota explicitly deferred to Beta. Rev 2.6 (2026-08-08): task-3 second-opinion — fingerprint normalized to resolved-DID scope, release decoupled from request context, admission wiring -fail-loud, ActorClass fail-closed.** +fail-loud, ActorClass fail-closed. +Rev 2.7 (2026-08-08): task-5 plan review — post.getStatus pulled forward into +task 5 as the T2 observation surface (unauthenticated; mild disclosure of +rejected-post status accepted, owner-flagged); hosted-community detection = +community credential presence, NEVER hosted_by_did (attacker-controlled for +firehose-indexed communities); deleted_accounts marker table (migration 036 — +account deletion previously left no marker, so swept admissions could be +recreated by replayed events); author-delete resurrection loop closed (driver +excludes tombstoned posts; decider refuses them); §5.1 keeps the deprecated +community.post collection subscribed until task 8's drain; §9's T2 list +re-scoped — accepted-state arcs prove at T1 (no T2 community holds +credentials), T2 contracts assert consumer semantics via getStatus.** **Supersedes** the write-path architecture in `docs/federation-prd.md`: that document solves cross-instance posting by service-auth-forwarding the write to diff --git a/internal/api/handlers/post/getstatus.go b/internal/api/handlers/post/getstatus.go new file mode 100644 index 0000000..21996f6 --- /dev/null +++ b/internal/api/handlers/post/getstatus.go @@ -0,0 +1,36 @@ +package post + +import ( + "net/http" + + "Coves/internal/core/posts" +) + +// RED STUB (task 5, cycle 1). Signature only — HandleGetStatus writes nothing, +// so every assertion in getstatus_integration_test.go fails on the response +// rather than on a missing symbol. The implementation is GREEN's. + +// GetStatusHandler serves social.coves.community.post.getStatus: one +// community's decision about one post (docs/PRD_AUTHOR_OWNED_POSTS.md §3.4). +// +// UNAUTHENTICATED, deliberately, and the trade is recorded rather than hidden. +// The caller with the strongest need is an author on a DIFFERENT server whose +// post is pending on this one (§7): they have no account here, so there is no +// session to require, and service-auth is Beta scope. The cost is that a +// rejected post's status is mildly disclosed to anyone who can name its URI — +// accepted by the owner in PRD rev 2.7. That the route carries no auth +// middleware is declared in internal/api/routes/registration_test.go, which is +// the only place the whole HTTP surface is enumerated. +type GetStatusHandler struct { + service posts.StatusService +} + +// NewGetStatusHandler creates a new getStatus handler. +func NewGetStatusHandler(service posts.StatusService) *GetStatusHandler { + return &GetStatusHandler{service: service} +} + +// HandleGetStatus handles +// GET /xrpc/social.coves.community.post.getStatus?post=at://...&community=did:... +func (h *GetStatusHandler) HandleGetStatus(w http.ResponseWriter, r *http.Request) { +} diff --git a/internal/api/handlers/post/getstatus_integration_test.go b/internal/api/handlers/post/getstatus_integration_test.go new file mode 100644 index 0000000..bde4523 --- /dev/null +++ b/internal/api/handlers/post/getstatus_integration_test.go @@ -0,0 +1,433 @@ +//go:build integration + +package post_test + +import ( + "context" + "database/sql" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "Coves/internal/api/handlers/post" + "Coves/internal/core/posts" + "Coves/internal/db/postgres" + "Coves/tests/fixtures" + "Coves/tests/testkit" + + _ "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// What social.coves.community.post.getStatus answers, over the real admissions +// table (docs/PRD_AUTHOR_OWNED_POSTS.md §3.4, pulled into task 5 by rev 2.7). +// +// This is at T1 rather than in a handler unit test with a fake service, and the +// reason is the whole point of the endpoint. getStatus exists so that a client +// can learn a decision that lives NOWHERE ELSE: a rejection writes no community +// record (§3.3), so unlike acceptance and removal there is no firehose event, no +// repo record, and no other endpoint carrying it. The only source of truth is a +// row in community_post_admissions, and a test that faked the service would +// prove the JSON shape while leaving open the question that matters — whether +// the five statuses the table can actually hold each come out as something a +// client can act on. So every case below seeds its state through the real +// repository's own mutations, in the same way the consumer will. +// +// The pipeline tier then uses this endpoint as its observation surface for the +// consumer contracts (§9, rev 2.7): no T2 community holds credentials, so the +// accepted-state arcs prove here and at the repository, and T2 asserts consumer +// semantics by asking getStatus what the AppView concluded. + +// statusStack is the getStatus endpoint over the real admissions store, plus the +// repository the tests seed through. +type statusStack struct { + handler *post.GetStatusHandler + admissions posts.AdmissionRepository +} + +func newStatusStack(db *sql.DB) statusStack { + admissions := postgres.NewAdmissionRepository(db) + return statusStack{ + handler: post.NewGetStatusHandler(posts.NewStatusService(admissions)), + admissions: admissions, + } +} + +// statusSubject is one (community, post) pair with the post row seeded in the +// AUTHOR's repo shape — at:///social.coves.community.postv2/ — +// because that is where a post lives now (§3.1) and a URI in the old +// community-repo shape would exercise a normalization path this endpoint must +// not have. +type statusSubject struct { + CommunityDID string + AuthorDID string + PostURI string +} + +func newStatusSubject(t *testing.T, db *sql.DB) statusSubject { + t.Helper() + + ctx := context.Background() + name := testkit.UniqueIDWithPrefix(t, "stat") + + communityDID, err := fixtures.Community(ctx, db, name, "owner"+name) + require.NoErrorf(t, err, "seeding community %s", name) + + authorDID := fixtures.DID(testkit.UniqueID(t)) + rkey := testkit.TID() + postURI := "at://" + authorDID + "/social.coves.community.postv2/" + rkey + + _, err = db.ExecContext(ctx, ` + INSERT INTO posts (uri, cid, rkey, author_did, community_did, title, created_at) + VALUES ($1, $2, $3, $4, $5, $6, NOW()) + `, postURI, "bafyreistatusseed", rkey, authorDID, communityDID, "a post whose status someone is asking about") + require.NoError(t, err, "seeding the post row the admission is about") + + return statusSubject{CommunityDID: communityDID, AuthorDID: authorDID, PostURI: postURI} +} + +// getStatus drives the handler with NO Authorization header, which is half the +// contract: the caller this endpoint is built for is an author on another +// server with no account here (§7). That the ROUTE carries no auth middleware +// is declared in internal/api/routes/registration_test.go; what is proven here +// is that the handler itself serves a complete answer to an anonymous caller +// rather than degrading to an empty or partial one. +func getStatus(t *testing.T, h *post.GetStatusHandler, postURI, communityDID string) *httptest.ResponseRecorder { + t.Helper() + + target := "/xrpc/social.coves.community.post.getStatus?post=" + + url.QueryEscape(postURI) + "&community=" + url.QueryEscape(communityDID) + rec := httptest.NewRecorder() + h.HandleGetStatus(rec, httptest.NewRequest(http.MethodGet, target, nil)) + return rec +} + +// decodeStatus reads a 200 body as a generic map, so that a MISSING optional +// field and a field present-but-null are distinguishable. That distinction is +// load-bearing: the lexicon marks decisionCode, decisionAt and acceptanceUri +// optional, and a client that meets `"decisionCode": null` on a pending post has +// been handed a decision that does not exist. +func decodeStatus(t *testing.T, rec *httptest.ResponseRecorder) map[string]interface{} { + t.Helper() + + require.Equalf(t, http.StatusOK, rec.Code, "status = %d, want 200 (body: %s)", rec.Code, rec.Body.String()) + assert.Equal(t, "application/json", rec.Header().Get("Content-Type")) + + var body map[string]interface{} + require.NoErrorf(t, json.Unmarshal(rec.Body.Bytes(), &body), + "decoding the getStatus response (body: %q)", rec.Body.String()) + return body +} + +func TestGetStatus_Pending(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + _, err := stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: "bafyreipendingcontent", + }) + require.NoError(t, err) + + body := decodeStatus(t, getStatus(t, stack.handler, subject.PostURI, subject.CommunityDID)) + + assert.Equal(t, "pending", body["status"]) + + // A pending post has been decided about by nobody, and the response must + // say exactly that by carrying no decision fields at all. A client polling + // this endpoint for the accepted transition (§7's UX) reads their presence + // as "the wait is over". + assert.NotContains(t, body, "decisionCode", "a pending post carries no decision") + assert.NotContains(t, body, "decisionAt", "a pending post carries no decision time") + assert.NotContains(t, body, "acceptanceUri", "a pending post has no acceptance record to point at") +} + +func TestGetStatus_AcceptedNamesTheAcceptanceRecord(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + const acceptedCID = "bafyreiacceptedcontent" + rkey := testkit.TID() + acceptanceURI := "at://" + subject.CommunityDID + "/social.coves.community.acceptance/" + rkey + + _, err := stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: acceptedCID, + }) + require.NoError(t, err) + + result, err := stack.admissions.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, + AcceptanceRkey: rkey, + PinnedCID: acceptedCID, + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome, "fixture: the acceptance must have applied") + + body := decodeStatus(t, getStatus(t, stack.handler, subject.PostURI, subject.CommunityDID)) + + assert.Equal(t, "accepted", body["status"]) + + // The acceptance URI is what turns this answer from a claim into something + // verifiable: the caller can go read the community's signed attestation + // instead of trusting this AppView's summary of it. Omitting it would make + // getStatus the authority on a fact it is only reporting. + assert.Equal(t, acceptanceURI, body["acceptanceUri"], + "an accepted post must name its acceptance record so the caller can read the signed attestation") + assert.NotContains(t, body, "decisionCode", "acceptance is not a refusal and carries no code") +} + +func TestGetStatus_RejectedCarriesTheReasonAndWhen(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + const judgedCID = "bafyreirejectedcontent" + _, err := stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: judgedCID, + }) + require.NoError(t, err) + + before := time.Now().Add(-time.Minute) + result, err := stack.admissions.RecordRejection(ctx, posts.RecordRejectionCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: string(posts.DecisionRateLimitExceeded), + JudgedCID: judgedCID, + Redrivable: false, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome, "fixture: the rejection must have applied") + + body := decodeStatus(t, getStatus(t, stack.handler, subject.PostURI, subject.CommunityDID)) + + assert.Equal(t, "rejected", body["status"]) + + // THIS is the case the endpoint exists for. A rejection writes no community + // record (§3.3), so there is no acceptance to read, no removal to read, and + // nothing on the firehose — an author whose post vanished has no other way + // to learn it was refused, or why. A `rejected` with no code is a status + // that tells them only that asking was pointless. + assert.Equal(t, string(posts.DecisionRateLimitExceeded), body["decisionCode"], + "a rejection is invisible everywhere else, so the code is the only explanation the author will ever get") + + decisionAt, ok := body["decisionAt"].(string) + require.Truef(t, ok, "decisionAt must be present and a string on a rejected post; got %#v", body["decisionAt"]) + parsed, err := time.Parse(time.RFC3339, decisionAt) + require.NoErrorf(t, err, "decisionAt %q must be an RFC 3339 datetime, per the lexicon's datetime format", decisionAt) + assert.Truef(t, parsed.After(before), "decisionAt (%s) must record when the decision was made", parsed) + + assert.NotContains(t, body, "acceptanceUri", "a rejected post was never accepted, so there is no record to name") +} + +func TestGetStatus_RemovedCarriesTheModerationCode(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + _, err := stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: "bafyreiremovedcontent", + }) + require.NoError(t, err) + + result, err := stack.admissions.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + DecisionCode: string(posts.DecisionRuleViolation), + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, result.Outcome, "fixture: the removal must have applied") + + body := decodeStatus(t, getStatus(t, stack.handler, subject.PostURI, subject.CommunityDID)) + + assert.Equal(t, "removed", body["status"]) + assert.Equal(t, string(posts.DecisionRuleViolation), body["decisionCode"], + "a removal's code is what #removedPost renders to the author; a removal without one is an unexplained moderation act") + assert.NotContains(t, body, "acceptanceUri", + "a removal deletes the acceptance in the same commit (§3.3), so naming one would point at a record that no longer exists") +} + +func TestGetStatus_PendingReacceptance(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + const originalCID = "bafyreioriginalcontent" + rkey := testkit.TID() + acceptanceURI := "at://" + subject.CommunityDID + "/social.coves.community.acceptance/" + rkey + + _, err := stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: originalCID, + }) + require.NoError(t, err) + _, err = stack.admissions.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + AcceptanceURI: acceptanceURI, + AcceptanceRkey: rkey, + PinnedCID: originalCID, + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + + // The author edits. The standing acceptance now pins content that is no + // longer current, and §5.5 forbids rendering the new CID under it. + _, err = stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: subject.CommunityDID, + PostURI: subject.PostURI, + EvaluatedCID: "bafyreieditedcontent", + }) + require.NoError(t, err) + + body := decodeStatus(t, getStatus(t, stack.handler, subject.PostURI, subject.CommunityDID)) + + // The status is reported verbatim rather than collapsed into "pending". An + // author who edited an accepted post and is shown plain `pending` cannot + // tell that from a post that was never accepted at all, and the two have + // completely different next steps. + assert.Equal(t, "pending_reacceptance", body["status"], + "pending_reacceptance must not be flattened into pending: the author needs to know their edit un-published an accepted post") +} + +func TestGetStatus_UnknownSubjectIsNotFound(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + // The post and the community both exist; what does not exist is a decision. + // That is the ordinary state of a post the community has never been offered, + // and it must be answered as not-found rather than invented as `pending` — + // pending is a promise that someone is going to decide. + rec := getStatus(t, stack.handler, subject.PostURI, subject.CommunityDID) + + require.Equalf(t, http.StatusNotFound, rec.Code, + "a subject with no admission row must be 404, not a fabricated status (body: %s)", rec.Body.String()) + + var body map[string]interface{} + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &body)) + assert.Equal(t, "NotFound", body["error"], + "the XRPC error name is what a client switches on; the shared mapper spells post-not-found as NotFound (errors.go)") +} + +func TestGetStatus_RequiresBothHalvesOfTheSubject(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + stack := newStatusStack(db) + subject := newStatusSubject(t, db) + + // A post carries independent decisions from several communities (§2), so + // "the status of this post" is not a question with one answer. A request + // missing either half has to be refused rather than answered about whichever + // row happens to be found first. + cases := []struct { + name string + target string + }{ + {"no post", "/xrpc/social.coves.community.post.getStatus?community=" + url.QueryEscape(subject.CommunityDID)}, + {"no community", "/xrpc/social.coves.community.post.getStatus?post=" + url.QueryEscape(subject.PostURI)}, + {"neither", "/xrpc/social.coves.community.post.getStatus"}, + {"empty post", "/xrpc/social.coves.community.post.getStatus?post=&community=" + url.QueryEscape(subject.CommunityDID)}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rec := httptest.NewRecorder() + stack.handler.HandleGetStatus(rec, httptest.NewRequest(http.MethodGet, tc.target, nil)) + assert.Equalf(t, http.StatusBadRequest, rec.Code, + "an incomplete subject must be 400 (body: %s)", rec.Body.String()) + }) + } +} + +func TestGetStatus_ScopesTheAnswerToTheNamedCommunity(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + stack := newStatusStack(db) + + // One post, two communities, opposite decisions. This is the fork case the + // per-(community, post) key exists for (§2, §6.1), and it is the sharpest + // available proof that the community parameter is actually part of the + // lookup rather than decoration on a post-scoped query. + accepting := newStatusSubject(t, db) + + otherName := testkit.UniqueIDWithPrefix(t, "statfork") + forkDID, err := fixtures.Community(ctx, db, otherName, "owner"+otherName) + require.NoError(t, err) + + const cid = "bafyreitwocommunities" + rkey := testkit.TID() + acceptanceURI := "at://" + accepting.CommunityDID + "/social.coves.community.acceptance/" + rkey + + for _, communityDID := range []string{accepting.CommunityDID, forkDID} { + _, err = stack.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: communityDID, + PostURI: accepting.PostURI, + EvaluatedCID: cid, + }) + require.NoError(t, err) + } + + _, err = stack.admissions.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: accepting.CommunityDID, + PostURI: accepting.PostURI, + AcceptanceURI: acceptanceURI, + AcceptanceRkey: rkey, + PinnedCID: cid, + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + + _, err = stack.admissions.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: forkDID, + PostURI: accepting.PostURI, + DecisionCode: string(posts.DecisionOffTopic), + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + + accepted := decodeStatus(t, getStatus(t, stack.handler, accepting.PostURI, accepting.CommunityDID)) + assert.Equal(t, "accepted", accepted["status"]) + + removed := decodeStatus(t, getStatus(t, stack.handler, accepting.PostURI, forkDID)) + assert.Equal(t, "removed", removed["status"], + "the same post is accepted in one community and removed in another; an answer that ignored the community parameter would report one of them everywhere") + assert.Equal(t, string(posts.DecisionOffTopic), removed["decisionCode"]) +} diff --git a/internal/api/routes/registration_test.go b/internal/api/routes/registration_test.go index 5a4f2f5..5815685 100644 --- a/internal/api/routes/registration_test.go +++ b/internal/api/routes/registration_test.go @@ -182,6 +182,17 @@ var declaredRoutes = []declaredRoute{ {http.MethodPost, "/xrpc/social.coves.community.post.create", authRequired, 0, false}, {http.MethodPost, "/xrpc/social.coves.community.post.delete", authRequired, 0, false}, {http.MethodGet, "/xrpc/social.coves.community.post.get", authOptional, 0, false}, + // getStatus takes no auth at all, unlike post.get beside it, and the + // asymmetry is the decision this line exists to hold. The caller it is for + // is an author on ANOTHER server asking this host whether it accepted their + // post (PRD §7): they have no account here, so there is no session to + // require and no viewer state to personalise. A rejection is AppView-local + // and writes no community record (§3.3), so this endpoint is the only way + // that answer is reachable at all. The accepted cost, recorded in PRD rev + // 2.7: anyone who can name a post URI learns its status in a community. + // Adding OptionalAuth here would be harmless; adding RequireAuth would make + // the cross-server case unanswerable, which is why it is declared. + {http.MethodGet, "/xrpc/social.coves.community.post.getStatus", authNone, 0, false}, // RegisterVoteRoutes — social.coves.feed.vote.* {http.MethodPost, "/xrpc/social.coves.feed.vote.create", authRequired, 0, false}, diff --git a/internal/atproto/jetstream/acceptance_consumer_test.go b/internal/atproto/jetstream/acceptance_consumer_test.go new file mode 100644 index 0000000..534f7d5 --- /dev/null +++ b/internal/atproto/jetstream/acceptance_consumer_test.go @@ -0,0 +1,544 @@ +//go:build integration + +package jetstream + +import ( + "context" + "database/sql" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "Coves/internal/atproto/identity" + "Coves/internal/core/posts" + "Coves/internal/core/users" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + _ "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Ingesting the two records a COMMUNITY writes about a post: +// social.coves.community.acceptance and social.coves.community.removal +// (docs/PRD_AUTHOR_OWNED_POSTS.md §5.4, §5.2). +// +// These are the records that decide what a community shows. A post claiming a +// community and lacking an acceptance is never rendered in it (§2), so an +// acceptance event is the moment a post becomes visible and a removal event is +// the moment it stops — which makes two properties non-negotiable here: +// +// - THE REPO DID IS THE COMMUNITY, and it must be a community this AppView has +// indexed. Nothing else in the event says which community decided. An +// acceptance from an arbitrary repo, taken at face value, is a stranger +// publishing into someone else's feed. +// - THE PINNED CID IS PART OF THE DECISION. An acceptance is a strongRef, and +// agreeing to at://x/postv2/y is not agreeing to whatever that URI holds +// later. A consumer that dropped the CID comparison would let an author edit +// content past moderation after approval. +// +// AND CONVERGENCE IS NOT FREE. An acceptance can arrive for a post this AppView +// has never seen, and redrive alone cannot fix that: bounded retries cannot +// manufacture a post event that a relay-coverage gap will never deliver. §5.4's +// direct fetch is the mechanism, and the second half of this file is about the +// ways that fetch must refuse rather than the way it succeeds — it is an +// outbound request whose destination is chosen by a stranger's record. + +const ( + accPrefix = "did:plc:acc" + accCommunity = accPrefix + "community" + accAuthor = accPrefix + "author" + accOutsider = accPrefix + "outsider" +) + +// accFixture is a consumer wired for community-repo events, plus the stores the +// assertions read. +type accFixture struct { + consumer *PostEventConsumer + admissions posts.AdmissionRepository + db *sql.DB +} + +func newAccFixture(t *testing.T, db *sql.DB, opts ...PostEventConsumerOption) accFixture { + t.Helper() + + insertBridgedUser(t, db, accAuthor, "accauthor.test") + insertBridgedCommunity(t, db, accCommunity, "acccommunity.test", accAuthor) + + us := newMockUserService() + us.users[accAuthor] = &users.User{DID: accAuthor, Handle: "accauthor.test"} + + admissions := postgres.NewAdmissionRepository(db) + wired := append([]PostEventConsumerOption{ + WithAdmissions(admissions), + WithDeletedAccounts(postgres.NewDeletedAccountRepository(db)), + }, opts...) + + return accFixture{ + consumer: NewPostEventConsumer( + postgres.NewPostRepository(db), + postgres.NewCommunityRepository(db), + us, + db, + wired..., + ), + admissions: admissions, + db: db, + } +} + +// accPostURI is the author-repo URI an acceptance points at. +func accPostURI(rkey string) string { return "at://" + accAuthor + "/" + PostV2Collection + "/" + rkey } + +// indexPV2 puts a post in the index the ordinary way — through the consumer — +// so these tests start from a state the pipeline actually produces. +func (f accFixture) indexPV2(t *testing.T, rkey, cid string, timeUS int64) string { + t.Helper() + uri := accPostURI(rkey) + require.NoError(t, f.consumer.HandleEvent(context.Background(), pv2Event( + accAuthor, "create", rkey, testkit.TID(), cid, timeUS, + pv2Record(accCommunity, "a post awaiting a decision", "body"), + )), "fixture: indexing the subject post") + return uri +} + +// acceptanceEvent builds a community-repo acceptance commit. +// +// The rkey is derived from the subject rather than invented, because that is +// what the writers do (§3.2): one post has exactly one acceptance rkey per +// community, forever, which is what makes three independent writers converge on +// putRecord of the same record instead of allocating duplicate TIDs. +func acceptanceEvent(communityDID, postURI, pinnedCID, rev string, timeUS int64) *JetstreamEvent { + return revCommitEvent(communityDID, posts.AcceptanceCollection, "create", + posts.SubjectRkey(postURI), rev, "bafyreiacceptancerecord", timeUS, + map[string]interface{}{ + "$type": posts.AcceptanceCollection, + "subject": map[string]interface{}{"uri": postURI, "cid": pinnedCID}, + "createdAt": "2026-03-01T00:00:00Z", + }) +} + +// removalEvent builds a community-repo removal commit. +func removalEvent(communityDID, postURI, pinnedCID, code, rev string, timeUS int64) *JetstreamEvent { + return revCommitEvent(communityDID, posts.RemovalCollection, "create", + posts.SubjectRkey(postURI), rev, "bafyreiremovalrecord", timeUS, + map[string]interface{}{ + "$type": posts.RemovalCollection, + "subject": map[string]interface{}{"uri": postURI, "cid": pinnedCID}, + "code": code, + "createdAt": "2026-03-01T00:00:00Z", + }) +} + +func TestAcceptanceConsumer_MatchingCID_AcceptsAndStampsTheWatermark(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newAccFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccmatch" + uri := f.indexPV2(t, "accmatch", cid, base) + + rev := testkit.TID() + require.NoError(t, f.consumer.HandleEvent(ctx, acceptanceEvent(accCommunity, uri, cid, rev, base+1_000_000))) + + admission, err := f.admissions.Get(ctx, accCommunity, uri) + require.NoError(t, err) + require.NotNil(t, admission) + + assert.Equal(t, posts.AdmissionStatusAccepted, admission.Status, + "an acceptance pinning the CID the AppView has indexed is the community agreeing to exactly this content") + assertNullableStringPV2(t, cid, admission.AcceptedCID, "accepted_cid must be the CID the acceptance pinned") + assertNullableStringPV2(t, posts.SubjectRkey(uri), admission.AcceptanceRkey, + "the acceptance rkey is deterministic from the subject; storing a different one breaks the one-record-per-subject convergence the writers rely on") + + // The §5.2 tuple, and specifically its second half. The rank is derived from + // the OPERATION by the repository, never taken from the wire: a put ranks + // above a delete so that the removal commit {acceptance-delete, removal-put} + // converges on removed and the restore commit {removal-delete, + // acceptance-put} converges on accepted, whichever half the consumer sees + // first. A caller-supplied rank would let one mislabelled event reorder a + // commit permanently. + require.NotNil(t, admission.LastCommunityEvent, "a community event must stamp the subject-scoped watermark") + assert.Equal(t, rev, admission.LastCommunityEvent.Rev, + "the watermark rev must be the COMMIT's rev — it is the only clock that orders acceptance against removal, since they are different record URIs about the same subject") + assert.Equal(t, posts.CommunityOpPut, admission.LastCommunityEvent.OpRank, + "a record write ranks as a put; the rank is the operation's kind, derived repo-side") +} + +func TestAcceptanceConsumer_ReplayIsANoOpAndLeavesTheRowUntouched(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newAccFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccreplay" + uri := f.indexPV2(t, "accreplay", cid, base) + + // The redelivery is guaranteed, not hypothetical: the connector rewinds its + // cursor after every reconnect, the AppView consumes overlapping feeds, and + // the dead-letter redriver replays. This exact commit WILL arrive twice. + event := acceptanceEvent(accCommunity, uri, cid, testkit.TID(), base+1_000_000) + require.NoError(t, f.consumer.HandleEvent(ctx, event)) + + before := readAdmissionRow(t, db, accCommunity, uri) + + require.NoError(t, f.consumer.HandleEvent(ctx, event), + "an equal watermark is a replay, and a replay is a no-op — not an error the connector logs as a failure") + + after := readAdmissionRow(t, db, accCommunity, uri) + assert.Equal(t, before, after, + "a replayed acceptance must leave the row byte-identical. Re-stamping decision_at or updated_at would make the moderation audit trail a function of how many feeds happened to carry the event") +} + +func TestAcceptanceConsumer_FromANonCommunityRepo_IsRefusedAndRedrivable(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newAccFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccoutsider" + uri := f.indexPV2(t, "accoutsider", cid, base) + + // accOutsider is a DID with no communities row. Nothing in the record says + // which community decided — the repo IS the claim — so accepting this would + // let anyone with a PDS publish into any feed by writing a record that names + // someone else's post. + err := f.consumer.HandleEvent(ctx, acceptanceEvent(accOutsider, uri, cid, testkit.TID(), base+1_000_000)) + require.Error(t, err, "an acceptance from a repo that is not an indexed community must not be applied") + + // Transient, not permanent, and the reason is delivery order rather than + // leniency: BigSky preserves order within a repo, not across repos, so a + // community's first acceptance can genuinely outrun its own profile event. + // Marking this permanent would spend the redrive budget that resolves the + // race and discard a legitimate decision. + assert.NotErrorIs(t, err, ErrPermanentEvent, + "an unindexed community repo is an ordering failure; the redrive resolves it once the community profile arrives") + + assert.Zero(t, countRows(t, db, + `SELECT count(*) FROM community_post_admissions WHERE post_uri = $1 AND community_did = $2`, uri, accOutsider), + "the refused acceptance must not have opened an admission row for the outsider repo") +} + +func TestRemovalConsumer_RemovesWithTheCodeItCarries(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newAccFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccremove" + uri := f.indexPV2(t, "accremove", cid, base) + + revs := increasingTIDs(t, 2) + require.NoError(t, f.consumer.HandleEvent(ctx, acceptanceEvent(accCommunity, uri, cid, revs[0], base+1_000_000))) + require.NoError(t, f.consumer.HandleEvent(ctx, removalEvent( + accCommunity, uri, cid, string(posts.DecisionRuleViolation), revs[1], base+2_000_000))) + + admission, err := f.admissions.Get(ctx, accCommunity, uri) + require.NoError(t, err) + require.NotNil(t, admission) + + assert.Equal(t, posts.AdmissionStatusRemoved, admission.Status) + assertNullableStringPV2(t, string(posts.DecisionRuleViolation), admission.DecisionCode, + "the removal's code is what #removedPost renders to the author; dropping it turns a moderation act into an unexplained disappearance") + assert.NotNil(t, admission.DecisionAt, "a removal must record when it happened") +} + +func TestRemovalConsumer_PreemptiveRemovalCreatesTheRow(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newAccFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccpreempt" + uri := f.indexPV2(t, "accpreempt", cid, base) + + // A removal with no prior acceptance is VALID (§5.4). A community that has + // decided in advance about a post — an author it is about to ban, content it + // has already seen elsewhere — must be able to say so, and a consumer that + // required an acceptance first would drop exactly the decisions a community + // most wants to make early. + require.NoError(t, f.consumer.HandleEvent(ctx, removalEvent( + accCommunity, uri, cid, string(posts.DecisionSpam), testkit.TID(), base+1_000_000))) + + admission, err := f.admissions.Get(ctx, accCommunity, uri) + require.NoErrorf(t, err, "a pre-emptive removal must create the admission row it decides") + require.NotNil(t, admission) + assert.Equal(t, posts.AdmissionStatusRemoved, admission.Status) + assertNullableStringPV2(t, string(posts.DecisionSpam), admission.DecisionCode, "decision_code") +} + +// --------------------------------------------------------------------------- +// §5.4 direct fetch: acceptance before post +// --------------------------------------------------------------------------- + +// fakeAuthorPDS is an httptest server answering com.atproto.repo.getRecord for +// the author's postv2 record, standing in for the PDS a DID document points at. +// +// It asserts the request shape as it serves, because the fetch is the one place +// the AppView reads a record without the firehose: a getRecord aimed at the +// wrong repo or collection would return someone else's record, and the CID check +// downstream would happily verify it. +func fakeAuthorPDS(t *testing.T, expectRepo string, handler http.HandlerFunc) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "/xrpc/com.atproto.repo.getRecord", r.URL.Path) + assert.Equal(t, expectRepo, r.URL.Query().Get("repo")) + assert.Equal(t, PostV2Collection, r.URL.Query().Get("collection")) + handler(w, r) + })) +} + +// serveRecord writes a getRecord response body. +func serveRecord(t *testing.T, w http.ResponseWriter, uri, cid string, value map[string]interface{}) { + t.Helper() + w.Header().Set("Content-Type", "application/json") + require.NoError(t, json.NewEncoder(w).Encode(map[string]interface{}{ + "uri": uri, "cid": cid, "value": value, + })) +} + +// newFetcherAt builds a DirectPostFetcher pointed at srv, with the SSRF guard +// stood down. +// +// Standing it down is REQUIRED and is the whole reason the seam exists: +// httptest listens on loopback, which the guard blocks by design, so a fetcher +// that could not be relaxed could not be tested against a fake PDS at all. The +// guard's default is asserted separately, and behaviourally, in +// TestDirectPostFetcher_RefusesAPrivateHostByDefault. +func newFetcherAt(t *testing.T, authorDID, pdsURL string) *DirectPostFetcher { + t.Helper() + fetcher := NewDirectPostFetcher(&mockIdentityResolverForUser{ + identities: map[string]*identity.Identity{ + authorDID: {DID: authorDID, Handle: "accauthor.test", PDSURL: pdsURL}, + }, + }) + fetcher.allowPrivateHosts = true + return fetcher +} + +func TestAcceptanceConsumer_UnindexedPost_IsFetchedDirectlyAndAccepted(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccfetched" + rkey := "accfetch" + uri := accPostURI(rkey) + + srv := fakeAuthorPDS(t, accAuthor, func(w http.ResponseWriter, r *http.Request) { + serveRecord(t, w, uri, cid, pv2Record(accCommunity, "fetched straight from the PDS", "body")) + }) + defer srv.Close() + + f := newAccFixture(t, db, WithPostRecordFetcher(newFetcherAt(t, accAuthor, srv.URL))) + + // The post was NEVER indexed — no create event ever arrived, and none ever + // will if the relay does not crawl the author's PDS. This is the case §5.4 + // says redrive cannot solve: retries cannot manufacture an event nobody is + // going to send. Without the fetch, convergence requires full relay + // coverage, which is a bet rather than a guarantee. + require.NoError(t, f.consumer.HandleEvent(ctx, acceptanceEvent(accCommunity, uri, cid, testkit.TID(), base))) + + authorDID, communityDID, storedCID, _, _ := readPV2Post(t, db, uri) + assert.Equal(t, accAuthor, authorDID, "the fetched post is attributed to the repo it was read from") + assert.Equal(t, accCommunity, communityDID) + assert.Equal(t, cid, storedCID) + + admission, err := f.admissions.Get(ctx, accCommunity, uri) + require.NoError(t, err) + require.NotNil(t, admission) + assert.Equal(t, posts.AdmissionStatusAccepted, admission.Status, + "the fetch exists so the acceptance can be APPLIED; indexing the post and leaving the decision pending would solve half the problem and leave the post invisible") +} + +func TestAcceptanceConsumer_FetchedCIDMismatch_IsPermanentlyRefused(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + base := time.Now().UnixMicro() + + rkey := "accmismatch" + uri := accPostURI(rkey) + + srv := fakeAuthorPDS(t, accAuthor, func(w http.ResponseWriter, r *http.Request) { + // The PDS serves the CURRENT version. The acceptance pins an older one. + serveRecord(t, w, uri, "bafyreiaccnowcurrent", pv2Record(accCommunity, "the version now at that rkey", "body")) + }) + defer srv.Close() + + f := newAccFixture(t, db, WithPostRecordFetcher(newFetcherAt(t, accAuthor, srv.URL))) + + err := f.consumer.HandleEvent(ctx, acceptanceEvent(accCommunity, uri, "bafyreiaccpinnedold", testkit.TID(), base)) + + // The CID check is what makes the fetch trustworthy at all. Without it the + // AppView indexes whatever the author's PDS chooses to serve under that + // rkey — the author (or whoever holds their keys) picks the content, and the + // community's signed acceptance is made to cover it retroactively. + require.Error(t, err, "a fetched record whose CID is not the one the acceptance pinned must never be indexed under that acceptance") + assert.ErrorIs(t, err, ErrPermanentEvent, + "the pinned version is gone from the repo and no retry brings it back; the connector must dead-letter this with its redrive budget already spent rather than re-fetching the same mismatch ten times") + + assert.Zero(t, countRows(t, db, `SELECT count(*) FROM posts WHERE uri = $1`, uri), + "the unverified record must not be indexed") +} + +func TestAcceptanceConsumer_FetchedRecordNamingAnotherCommunity_IsPermanentlyRefused(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + base := time.Now().UnixMicro() + + const cid = "bafyreiaccwrongcommunity" + rkey := "accwrongcomm" + uri := accPostURI(rkey) + + srv := fakeAuthorPDS(t, accAuthor, func(w http.ResponseWriter, r *http.Request) { + // The record was submitted to a DIFFERENT community. + serveRecord(t, w, uri, cid, pv2Record("did:plc:accsomewhereelse", "submitted elsewhere", "body")) + }) + defer srv.Close() + + f := newAccFixture(t, db, WithPostRecordFetcher(newFetcherAt(t, accAuthor, srv.URL))) + + err := f.consumer.HandleEvent(ctx, acceptanceEvent(accCommunity, uri, cid, testkit.TID(), base)) + + // Cross-community acceptance is the privileged fork/import flow, and §10.2 + // is explicit that it is deliberately NOT built: the data model supports it, + // the flow that exercises it is future scope. Until it exists, a community + // accepting a post that names someone else is a community pulling another + // community's content into its feed on its own say-so. + require.Error(t, err, "a community may not accept a post whose record names a different community — the fork/import flow is deliberately not built (§10.2)") + assert.ErrorIs(t, err, ErrPermanentEvent, + "the record's community field is immutable across updates (§3.1), so this can never become valid; retrying it is pure noise") + + assert.Zero(t, countRows(t, db, `SELECT count(*) FROM posts WHERE uri = $1`, uri), + "the refused record must not be indexed") +} + +func TestDirectPostFetcher_RefusesAnOversizedBody(t *testing.T) { + t.Parallel() + + uri := accPostURI("accoversized") + + // A post record has a lexicon-bounded size; a PDS streaming megabytes is + // either broken or hostile. The cap has to be enforced by the FETCHER + // because the DID document that chose this host is attacker-controlled — any + // stranger who writes an acceptance record picks where this request goes, + // and an unbounded read there is a memory-exhaustion primitive handed to the + // public. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"uri":"` + uri + `","cid":"bafyreiaccbig","value":{"$type":"` + + PostV2Collection + `","community":"` + accCommunity + `","createdAt":"2026-03-01T00:00:00Z","content":"`)) + chunk := strings.Repeat("A", 64*1024) + for i := 0; i < 64; i++ { // 4 MiB of content + if _, err := w.Write([]byte(chunk)); err != nil { + return + } + } + _, _ = w.Write([]byte(`"}}`)) + })) + defer srv.Close() + + fetched, err := newFetcherAt(t, accAuthor, srv.URL).FetchPost(context.Background(), uri) + require.Error(t, err, "a response past the size cap must be an error, not a truncated record parsed as if it were whole") + assert.Nil(t, fetched) +} + +func TestDirectPostFetcher_RefusesAPrivateHostByDefault(t *testing.T) { + t.Parallel() + + uri := accPostURI("accssrf") + + var reached bool + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + serveRecord(t, w, uri, "bafyreiaccssrf", pv2Record(accCommunity, "should never be read", "body")) + })) + defer srv.Close() + + // The default constructor, with nothing stood down. httptest listens on + // loopback, which is precisely the class of address the guard blocks. + // + // This is asserted behaviourally rather than by reading the flag because the + // flag is not the protection — a fetcher that stored allowPrivateHosts and + // then built its client from a hardcoded false (or true) would pass a field + // check and fail this. + fetcher := NewDirectPostFetcher(&mockIdentityResolverForUser{ + identities: map[string]*identity.Identity{ + accAuthor: {DID: accAuthor, Handle: "accauthor.test", PDSURL: srv.URL}, + }, + }) + require.Falsef(t, fetcher.allowPrivateHosts, + "NewDirectPostFetcher must default to SSRF protection ON: the PDS this dials is named by a DID document anyone can publish") + + fetched, err := fetcher.FetchPost(context.Background(), uri) + require.Error(t, err, + "a PDS resolving to a private address must be refused: an acceptance record is public and unauthenticated input, so this fetch is a request forger pointed at whatever the AppView can reach on its own network") + assert.Nil(t, fetched) + assert.Falsef(t, reached, "the guard must refuse before the request is made, not after the server has already answered") +} + +// readAdmissionRow returns every mutable column of one admission row as a +// comparable value, so "the row did not change" can be asserted as a whole +// rather than field by field — a new column added later is covered without +// anyone remembering to extend an assertion. +func readAdmissionRow(t *testing.T, db *sql.DB, communityDID, postURI string) []interface{} { + t.Helper() + + var ( + status string + acceptanceURI, acceptanceRkey, accepted *string + decisionCode, evaluatedCID, rev *string + decisionAt *time.Time + opRank *int16 + redrivable bool + createdAt, updatedAt time.Time + ) + require.NoError(t, db.QueryRow(` + SELECT status, acceptance_uri, acceptance_rkey, accepted_cid, decision_code, decision_at, + evaluated_cid, redrivable, last_community_rev, last_community_op_rank, created_at, updated_at + FROM community_post_admissions WHERE community_did = $1 AND post_uri = $2 + `, communityDID, postURI).Scan( + &status, &acceptanceURI, &acceptanceRkey, &accepted, &decisionCode, &decisionAt, + &evaluatedCID, &redrivable, &rev, &opRank, &createdAt, &updatedAt, + )) + + deref := func(p *string) interface{} { + if p == nil { + return nil + } + return *p + } + var decidedAt interface{} + if decisionAt != nil { + decidedAt = decisionAt.UTC() + } + var rank interface{} + if opRank != nil { + rank = *opRank + } + return []interface{}{ + status, deref(acceptanceURI), deref(acceptanceRkey), deref(accepted), deref(decisionCode), + decidedAt, deref(evaluatedCID), redrivable, deref(rev), rank, createdAt.UTC(), updatedAt.UTC(), + } +} diff --git a/internal/atproto/jetstream/authorpost.go b/internal/atproto/jetstream/authorpost.go new file mode 100644 index 0000000..2c44009 --- /dev/null +++ b/internal/atproto/jetstream/authorpost.go @@ -0,0 +1,123 @@ +package jetstream + +import ( + "context" + "net/http" + + "Coves/internal/atproto/identity" + "Coves/internal/core/posts" +) + +// RED STUB (task 5, cycle 1). Declarations only — every function body here +// returns zero values, so the tests describing author-owned post ingestion +// compile and fail on their assertions rather than on missing symbols. The +// implementations, the HandleEvent dispatch for the three new collections, and +// the consumerWantedCollections entries are GREEN's. + +// PostV2Collection is the author-repo post record of +// docs/PRD_AUTHOR_OWNED_POSTS.md §3.1 — the §3.0 successor to the deprecated +// social.coves.community.post. +// +// The NSID is new rather than reused, and that is a safety property, not +// bookkeeping: a consumer built against the published community.post derives +// community = repo DID for that collection, so feeding it author-repo records +// under the same name would have it index authors as communities. A new NSID +// makes a stale consumer ignore the records entirely, which is the correct +// failure mode. +const PostV2Collection = "social.coves.community.postv2" + +// DeletedAccountLookup reports whether a DID names an account this AppView was +// asked to erase (migration 036, PRD rev 2.7). +// +// It exists because "no users row" stopped being an answer. Under author-owned +// posts an unknown author is a NORMAL state that must still index (§5.3), so +// the absence of a profile can no longer stand in for "this account is gone" — +// and without a marker, a redriven post event or a replayed acceptance quietly +// recreates the very rows the deletion swept. +// +// A lookup FAILURE must never be read as "not deleted". Failing open here means +// a database blip re-indexes an erased account's content, which is the one +// outcome a deletion is supposed to make impossible. +type DeletedAccountLookup interface { + IsAccountDeleted(ctx context.Context, did string) (bool, error) +} + +// WithAdmissions installs the per-(community, post) admission store. Without +// it, postv2 and acceptance/removal events have nowhere to record a decision. +func WithAdmissions(admissions posts.AdmissionRepository) PostEventConsumerOption { + return func(c *PostEventConsumer) { c.admissions = admissions } +} + +// WithDeletedAccounts installs the erased-account gate. Without it, no gate +// runs and every event indexes — the pre-036 behaviour. +func WithDeletedAccounts(lookup DeletedAccountLookup) PostEventConsumerOption { + return func(c *PostEventConsumer) { c.deletedAccounts = lookup } +} + +// WithPostRecordFetcher installs the §5.4 direct fetch used when an acceptance +// names a post this AppView has never indexed. +func WithPostRecordFetcher(fetcher PostRecordFetcher) PostEventConsumerOption { + return func(c *PostEventConsumer) { c.postFetcher = fetcher } +} + +// FetchedPost is one author-repo record read directly from its PDS. +// +// It carries the CID separately from the record because the CID is what the +// caller VERIFIES: an acceptance pins a strongRef, and a fetch that returned +// only the record body would leave the consumer indexing whatever the author's +// PDS felt like serving under that rkey. +type FetchedPost struct { + URI string + CID string + Record map[string]interface{} +} + +// PostRecordFetcher reads an author's post record straight from their PDS. +// +// This is what makes firehose-only ingestion actually CONVERGE (§5.4). +// Acceptance-before-post does not converge by dead-letter redrive alone: +// bounded retries cannot manufacture a post event that a relay-coverage gap +// will never deliver. Redrive stays the backstop for transient failures; this +// is the mechanism. +type PostRecordFetcher interface { + // FetchPost resolves the repo DID in postURI to a PDS and reads the record. + // The returned CID is the PDS's, unverified — checking it against the CID an + // acceptance pinned is the caller's job, because only the caller knows what + // was pinned. + FetchPost(ctx context.Context, postURI string) (*FetchedPost, error) +} + +// DirectPostFetcher is the production PostRecordFetcher: DID resolution, then +// com.atproto.repo.getRecord over an SSRF-guarded client with a size cap. +type DirectPostFetcher struct { + resolver identity.Resolver + + // allowPrivateHosts disables the SSRF protection that blocks private and + // loopback addresses. NEVER set outside tests. It exists for the same reason + // blueskypost's allowPrivateHost does: this package's own tests point the + // fetcher at an httptest server, which necessarily listens on loopback and + // would otherwise be refused by the guard that must stay on in production. + // + // The guard is not decorative here. The DID document that names the PDS is + // attacker-controlled — anyone can publish one — so an unguarded fetcher is + // a request forger pointed at whatever is reachable from the AppView's + // network, driven by any stranger who writes an acceptance record. + allowPrivateHosts bool +} + +// NewDirectPostFetcher wires the §5.4 fetch. SSRF protection is ON and there is +// no parameter to turn it off: a constructor that accepted a boolean is a +// constructor someone eventually passes true to from production wiring. +func NewDirectPostFetcher(resolver identity.Resolver) *DirectPostFetcher { + return &DirectPostFetcher{resolver: resolver} +} + +// httpClient builds the guarded client for one fetch. Declared here so the +// guard is derived from allowPrivateHosts at call time rather than baked into a +// client at construction, where a test seam could not reach it. +func (f *DirectPostFetcher) httpClient() *http.Client { return nil } + +// FetchPost implements PostRecordFetcher. +func (f *DirectPostFetcher) FetchPost(ctx context.Context, postURI string) (*FetchedPost, error) { + return nil, nil +} diff --git a/internal/atproto/jetstream/post_consumer.go b/internal/atproto/jetstream/post_consumer.go index 7ad9651..dc2b6b6 100644 --- a/internal/atproto/jetstream/post_consumer.go +++ b/internal/atproto/jetstream/post_consumer.go @@ -29,6 +29,19 @@ type PostEventConsumer struct { // before its author's profile. The identity is admitted only when its PDS // passes bridgeTrust. identityResolver identity.Resolver + + // RED STUB fields (task 5, cycle 1) — the collaborators author-owned post + // ingestion needs. Declared here so the options in authorpost.go compile; + // nothing reads them yet. See docs/PRD_AUTHOR_OWNED_POSTS.md §5.3-§5.6. + // + // admissions holds the per-(community, post) decision state. nil means the + // consumer is running in its pre-034 shape and records no admissions. + admissions posts.AdmissionRepository + // deletedAccounts gates events from erased accounts. nil means no gate. + deletedAccounts DeletedAccountLookup + // postFetcher resolves an acceptance whose subject was never indexed. nil + // means the dead-letter queue is the only convergence mechanism. + postFetcher PostRecordFetcher } // PostEventConsumerOption configures optional PostEventConsumer behaviour. diff --git a/internal/atproto/jetstream/postv2_consumer_test.go b/internal/atproto/jetstream/postv2_consumer_test.go new file mode 100644 index 0000000..254903f --- /dev/null +++ b/internal/atproto/jetstream/postv2_consumer_test.go @@ -0,0 +1,494 @@ +//go:build integration + +package jetstream + +import ( + "context" + "database/sql" + "testing" + "time" + + "Coves/internal/core/posts" + "Coves/internal/core/users" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + _ "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Ingesting social.coves.community.postv2 — the post record that now lives in +// the AUTHOR's repo (docs/PRD_AUTHOR_OWNED_POSTS.md §3.1, §5.3). +// +// The collection is new, and so is almost everything about how an event from it +// is read. Three inversions drive every case below: +// +// - AUTHORSHIP COMES FROM THE REPO, not from the record. There is no `author` +// field to read and none to trust; event.Did IS the author. The old +// consumer's central security check — repo DID must equal the record's +// community — inverts into its opposite: the repo DID must NOT be the +// community, and the community is a claim the record makes. +// - AN UNKNOWN AUTHOR IS NORMAL. Open federated posting means the author of a +// post may be someone this AppView has never indexed, so "no users row" can +// no longer refuse the event (§5.3, migration 034 dropped the FK). +// - THE POST ROW IS NO LONGER THE DECISION. Whether a community shows the post +// lives in community_post_admissions, and a postv2 event's job is to record +// content plus a pending admission — never to decide anything. +// +// A REFUSAL AND A SKIP ARE DIFFERENT ANSWERS, and several cases turn on which +// one is being asserted. The connector dead-letters exactly what a handler +// returns as an error: nil is "handled, nothing more to do" and never reaches +// the queue; an error wrapped in ErrPermanentEvent is dead-lettered with its +// redrive budget already spent; any other error is dead-lettered retryable. +// "No dead letter" below therefore means a nil return, and it is asserted +// wherever an event must be dropped without the queue growing. + +const ( + pv2Prefix = "did:plc:pv2" + pv2Community = pv2Prefix + "community" + pv2Author = pv2Prefix + "author" + pv2Other = pv2Prefix + "otherauthor" +) + +// pv2Fixture is a wired consumer plus the stores the assertions read. +type pv2Fixture struct { + consumer *PostEventConsumer + admissions posts.AdmissionRepository + db *sql.DB + users *mockUserService +} + +// newPV2Fixture indexes a community and returns a consumer wired with the three +// collaborators author-owned ingestion needs, all real: the admissions store and +// the deleted-account lookup both run against this test's Postgres clone, so a +// gate that reads the wrong table fails here rather than passing against a map. +func newPV2Fixture(t *testing.T, db *sql.DB) pv2Fixture { + t.Helper() + + insertBridgedUser(t, db, pv2Author, "pv2author.test") + insertBridgedCommunity(t, db, pv2Community, "pv2community.test", pv2Author) + + us := newMockUserService() + us.users[pv2Author] = &users.User{DID: pv2Author, Handle: "pv2author.test"} + + admissions := postgres.NewAdmissionRepository(db) + consumer := NewPostEventConsumer( + postgres.NewPostRepository(db), + postgres.NewCommunityRepository(db), + us, + db, + WithAdmissions(admissions), + WithDeletedAccounts(postgres.NewDeletedAccountRepository(db)), + ) + + return pv2Fixture{consumer: consumer, admissions: admissions, db: db, users: us} +} + +// pv2URI is the AT-URI of an author-repo post: the author's DID is the +// authority, which is the whole point of the flip. +func pv2URI(authorDID, rkey string) string { + return "at://" + authorDID + "/" + PostV2Collection + "/" + rkey +} + +// pv2Record builds a postv2 record body. It carries NO author field, by +// construction — the lexicon has none (§3.1), and a consumer that still read one +// would be reading a field only a forger would bother to send. +func pv2Record(communityDID, title, content string) map[string]interface{} { + return map[string]interface{}{ + "$type": PostV2Collection, + "community": communityDID, + "title": title, + "content": content, + "createdAt": "2026-03-01T00:00:00Z", + } +} + +// pv2Event builds a commit event in the AUTHOR's repo. +func pv2Event(authorDID, op, rkey, rev, cid string, timeUS int64, record map[string]interface{}) *JetstreamEvent { + return revCommitEvent(authorDID, PostV2Collection, op, rkey, rev, cid, timeUS, record) +} + +// readPV2Post returns the indexed row's identity columns, or fails naming the +// URI that is missing. +func readPV2Post(t *testing.T, db *sql.DB, uri string) (authorDID, communityDID, cid, title string, deletedAt *time.Time) { + t.Helper() + err := db.QueryRow( + `SELECT author_did, community_did, cid, title, deleted_at FROM posts WHERE uri = $1`, uri, + ).Scan(&authorDID, &communityDID, &cid, &title, &deletedAt) + require.NoErrorf(t, err, "no post row for %s", uri) + return authorDID, communityDID, cid, title, deletedAt +} + +func countRows(t *testing.T, db *sql.DB, query string, args ...interface{}) int { + t.Helper() + var n int + require.NoError(t, db.QueryRow(query, args...).Scan(&n)) + return n +} + +// markAccountDeleted writes the migration-036 erasure marker directly, which is +// what userRepo.Delete leaves behind. +func markAccountDeleted(t *testing.T, db *sql.DB, did string) { + t.Helper() + _, err := db.Exec( + `INSERT INTO deleted_accounts (did, deleted_at) VALUES ($1, NOW()) ON CONFLICT (did) DO NOTHING`, did) + require.NoErrorf(t, err, "marking %s deleted", did) +} + +func TestPostV2Consumer_Create_IndexesTheAuthorsPostAndOpensAPendingAdmission(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + const cid = "bafyreipv2create" + rkey := "pv2create" + uri := pv2URI(pv2Author, rkey) + + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "create", rkey, testkit.TID(), cid, time.Now().UnixMicro(), + pv2Record(pv2Community, "an author-signed post", "words the author is accountable for"), + ))) + + authorDID, communityDID, storedCID, title, deletedAt := readPV2Post(t, db, uri) + + // The author is the repo, not a field. If this ever reads from the record + // instead, any repo can claim any author — which is exactly the + // impersonation power the flip removed (§1). + assert.Equal(t, pv2Author, authorDID, + "author_did must come from event.Did: the record has no author field, and deriving one from anywhere else restores the forgery the flip removed") + assert.Equal(t, pv2Community, communityDID, + "community_did comes from the record — it is the author's submission target, a claim the community has not yet agreed to") + assert.Equal(t, cid, storedCID) + assert.Equal(t, "an author-signed post", title) + assert.Nil(t, deletedAt) + + admission, err := f.admissions.Get(ctx, pv2Community, uri) + require.NoErrorf(t, err, "indexing a postv2 must open the admission row the community will decide against") + require.NotNil(t, admission) + + // PENDING, not accepted. The post claims the community; the community has + // said nothing. §2 is explicit that a post lacking an acceptance is never + // shown in that community, and an indexer that opened the row as anything + // else would publish speech the community never agreed to carry. + assert.Equal(t, posts.AdmissionStatusPending, admission.Status) + assertNullableStringPV2(t, cid, admission.EvaluatedCID, + "evaluated_cid must be the CID this event carried: it is what the next decision judges and what an acceptance's pinned CID is compared against") + + // An author-repo event orders by the per-record rev gate, never by the + // community watermark. Stamping one here would let an author's edit outrank + // a moderator's removal — two repos, two unrelated revision clocks (§5.2). + assert.Nilf(t, admission.LastCommunityEvent, + "an author-repo event must not advance the community watermark; got %+v", admission.LastCommunityEvent) +} + +func TestPostV2Consumer_DeletedAuthor_IsDroppedWithoutADeadLetter(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + // The account was erased. Migration 036's marker is the only thing that says + // so: the users row is gone, and under §5.3 a missing users row means + // "federated author we have not indexed", which indexes normally. Without + // the marker this event silently re-creates the content the deletion swept — + // and it WILL arrive, because dead-letter redrives and overlapping feeds + // replay events long after the account is gone. + markAccountDeleted(t, db, pv2Other) + + rkey := "pv2deleted" + uri := pv2URI(pv2Other, rkey) + + err := f.consumer.HandleEvent(ctx, pv2Event( + pv2Other, "create", rkey, testkit.TID(), "bafyreipv2deleted", time.Now().UnixMicro(), + pv2Record(pv2Community, "a post from an erased account", "content the AppView was asked to forget"), + )) + + // Nil, not an error. The connector dead-letters whatever a handler returns, + // so refusing this event with an error would fill the queue with rows that + // redrive, fail identically, and retire — turning every erased account into + // a permanent stream of operational noise. The event is not a failure; it is + // an event with nothing to do. + require.NoError(t, err, + "an event from an erased account must be dropped as a no-op: returning an error dead-letters it, and the queue exists for failures, not for events the AppView correctly ignores") + + assert.Zero(t, countRows(t, db, `SELECT count(*) FROM posts WHERE uri = $1`, uri), + "the post of an erased account must not be indexed") + assert.Zero(t, countRows(t, db, `SELECT count(*) FROM community_post_admissions WHERE post_uri = $1`, uri), + "no admission row either: an admission for an erased account's post is the row migration 036 exists to stop being recreated") +} + +func TestPostV2Consumer_UnknownAuthorIndexesAnyway(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + // The §5.3 flip, and the case that separates "erased" from "never seen". + // pv2Other has no users row and no erasure marker — the ordinary state of an + // author on someone else's server. The old consumer refused this event, which + // the test architecture recorded as "federated authors cannot currently be + // indexed"; open federated posting makes that refusal a bug. + rkey := "pv2unknown" + uri := pv2URI(pv2Other, rkey) + + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Other, "create", rkey, testkit.TID(), "bafyreipv2unknown", time.Now().UnixMicro(), + pv2Record(pv2Community, "a post from a federated stranger", "posted from a PDS we have never met"), + )), "an author this AppView has never indexed must not block ingestion: that refusal is what made cross-server posting impossible") + + authorDID, _, _, _, _ := readPV2Post(t, db, uri) + assert.Equal(t, pv2Other, authorDID) + + admission, err := f.admissions.Get(ctx, pv2Community, uri) + require.NoError(t, err) + require.NotNil(t, admission) + assert.Equal(t, posts.AdmissionStatusPending, admission.Status) +} + +func TestPostV2Consumer_UnknownCommunity_IsRedrivable(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + const ghostCommunity = "did:plc:pv2ghostcommunity" + rkey := "pv2ghost" + + err := f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "create", rkey, testkit.TID(), "bafyreipv2ghost", time.Now().UnixMicro(), + pv2Record(ghostCommunity, "aimed at a community we have not indexed", "body"), + )) + + require.Error(t, err, "a post naming a community this AppView has never indexed cannot open an admission row against it") + + // The classification is the assertion. BigSky preserves order within a repo, + // not across repos, so a post can genuinely arrive before the community's own + // profile event — this is an ORDERING failure, and marking it permanent would + // discard every post that merely arrived early, with the redrive that would + // have fixed it already spent. + assert.NotErrorIs(t, err, ErrPermanentEvent, + "community-not-found is an ordering failure and must stay transient so the redrive succeeds once the community arrives") + + assert.Zero(t, countRows(t, db, `SELECT count(*) FROM posts WHERE uri = $1`, pv2URI(pv2Author, rkey)), + "the refused post must not have been indexed") +} + +func TestPostV2Consumer_UpdateChangingCommunity_IgnoresTheWholeEvent(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + otherName := "pv2secondcommunity" + const secondCommunity = pv2Prefix + "community2" + insertBridgedCommunity(t, db, secondCommunity, otherName+".test", pv2Author) + + rkey := "pv2retarget" + uri := pv2URI(pv2Author, rkey) + base := time.Now().UnixMicro() + revs := increasingTIDs(t, 2) + + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "create", rkey, revs[0], "bafyreipv2original", base, + pv2Record(pv2Community, "original title", "original body"), + ))) + + // The retarget attempt: same record, new community, and new content riding + // along. §3.1 says the ENTIRE event is invalid — discard it, do not merely + // keep the old community value. Applying the content while ignoring the + // community would leave the first community's admission holding a CID it + // never evaluated, silently publishing content nobody judged under a + // standing acceptance. + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "update", rkey, revs[1], "bafyreipv2retargeted", base+1_000_000, + pv2Record(secondCommunity, "retargeted title", "retargeted body"), + )), "an update that changes community is invalid, not an infrastructure failure: it must be skipped, not dead-lettered") + + _, communityDID, cid, title, _ := readPV2Post(t, db, uri) + assert.Equal(t, pv2Community, communityDID, "the community must not move; retargeting a post means writing a new record") + assert.Equalf(t, "bafyreipv2original", cid, + "the whole event is invalid, so the CID must not move either — a moved CID under an unmoved community is content the community never evaluated") + assert.Equal(t, "original title", title, "the content half of a rejected event must be rejected with it") + + assert.Zero(t, countRows(t, db, + `SELECT count(*) FROM community_post_admissions WHERE post_uri = $1 AND community_did = $2`, uri, secondCommunity), + "an ignored retarget must not open an admission in the community it named: that row would be a dangling decision about a post that never claimed this community") +} + +func TestPostV2Consumer_UpdateWithNewContent_ReopensAnAcceptedAdmission(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + rkey := "pv2edit" + uri := pv2URI(pv2Author, rkey) + base := time.Now().UnixMicro() + revs := increasingTIDs(t, 2) + + const originalCID = "bafyreipv2editv1" + const editedCID = "bafyreipv2editv2" + + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "create", rkey, revs[0], originalCID, base, + pv2Record(pv2Community, "before the edit", "the version the community judged"), + ))) + + // The community accepts the original, pinning that exact CID. + acceptanceRkey := testkit.TID() + accepted, err := f.admissions.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: pv2Community, + PostURI: uri, + AcceptanceURI: "at://" + pv2Community + "/social.coves.community.acceptance/" + acceptanceRkey, + AcceptanceRkey: acceptanceRkey, + PinnedCID: originalCID, + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, accepted.Outcome, "fixture: the acceptance must stand before the edit arrives") + + // The author edits. The standing acceptance now pins content that is no + // longer current, and §5.5 forbids rendering the new CID under it. + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "update", rkey, revs[1], editedCID, base+1_000_000, + pv2Record(pv2Community, "after the edit", "words the community has not seen"), + ))) + + _, _, storedCID, title, _ := readPV2Post(t, db, uri) + assert.Equal(t, editedCID, storedCID, "the edit's content must be indexed — it is what the community will re-judge") + assert.Equal(t, "after the edit", title) + + admission, err := f.admissions.Get(ctx, pv2Community, uri) + require.NoError(t, err) + require.NotNil(t, admission) + + assert.Equal(t, posts.AdmissionStatusPendingReacceptance, admission.Status, + "an edit under a standing acceptance must reopen the decision: auto-rendering the new CID under the old acceptance would let an author swap content past moderation after approval") + assertNullableStringPV2(t, editedCID, admission.EvaluatedCID, + "evaluated_cid must follow the content, or the re-decision judges the version the author replaced") + assertNullableStringPV2(t, originalCID, admission.AcceptedCID, + "the acceptance still pins the CID the community actually agreed to; moving it here would forge agreement to the edit") +} + +func TestPostV2Consumer_EditOfARemovedPost_IsSkippedNotDeadLettered(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + rkey := "pv2removededit" + uri := pv2URI(pv2Author, rkey) + base := time.Now().UnixMicro() + revs := increasingTIDs(t, 2) + + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "create", rkey, revs[0], "bafyreipv2removedv1", base, + pv2Record(pv2Community, "will be removed", "body"), + ))) + + // The precondition is asserted, not assumed. ApplyRemoval below creates the + // admission row when the subject is absent, so without this the whole test + // would pass against a consumer that ignores postv2 entirely — the edit + // would be a no-op for the wrong reason and the final assertion would read + // back a row nothing had ever contested. + _, _, _, _, deletedAt := readPV2Post(t, db, uri) + require.Nil(t, deletedAt, "fixture: the post must be indexed and live before the removal lands") + + removed, err := f.admissions.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: pv2Community, + PostURI: uri, + DecisionCode: string(posts.DecisionRuleViolation), + Watermark: posts.CommunityWatermark{Rev: testkit.TID()}, + }) + require.NoError(t, err) + require.Equal(t, posts.AdmissionApplied, removed.Outcome, "fixture: the removal must stand") + + // §5.5: removal is terminal against author-repo events. The repository + // answers this edit with skipped_terminal — a value, not an error — and the + // consumer must pass that through as success. Mapping a CAS skip onto an + // error return is the mistake migration 033's precedent exists to prevent: + // it routes the system WORKING into the dead-letter queue, where every + // redrive re-runs a decision that will refuse identically forever. + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "update", rkey, revs[1], "bafyreipv2removedv2", base+1_000_000, + pv2Record(pv2Community, "edited while removed", "laundering attempt"), + )), "a terminal admission skip is an outcome, not a failure: returning an error would dead-letter healthy skips") + + admission, err := f.admissions.Get(ctx, pv2Community, uri) + require.NoError(t, err) + require.NotNil(t, admission) + assert.Equal(t, posts.AdmissionStatusRemoved, admission.Status, + "editing a removed post must not reopen it; that is how a removed post gets laundered back through auto-acceptance") +} + +func TestPostV2Consumer_Delete_TombstonesTheRow(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + f := newPV2Fixture(t, db) + ctx := context.Background() + + rkey := "pv2delete" + uri := pv2URI(pv2Author, rkey) + base := time.Now().UnixMicro() + revs := increasingTIDs(t, 2) + + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "create", rkey, revs[0], "bafyreipv2delete", base, + pv2Record(pv2Community, "to be deleted by its author", "body that must survive the tombstone"), + ))) + + // A delete carries no record, exactly as Jetstream delivers it. + require.NoError(t, f.consumer.HandleEvent(ctx, pv2Event( + pv2Author, "delete", rkey, revs[1], "", base+1_000_000, nil, + ))) + + _, _, _, title, deletedAt := readPV2Post(t, db, uri) + require.NotNil(t, deletedAt, + "an author delete must SOFT-delete: the row is the rev gate's tombstone, the comment thread's parent, and what moderation still reads") + assert.Equal(t, "to be deleted by its author", title, + "a soft delete must not blank the content") + + // The host-side half — the community observing the tombstone and deleting + // its acceptance (§5.3) — is task 5b's scope and deliberately not asserted + // here. What matters at this point is that the tombstone exists for that + // sweep to find. +} + +// assertNullableStringPV2 asserts a nullable column holds exactly want. +// +// Named apart from the postgres package's helper of the same shape because this +// package has its own; the duplication is two lines against an import cycle. +func assertNullableStringPV2(t *testing.T, want string, got *string, what string) { + t.Helper() + if !assert.NotNilf(t, got, "%s: want %q, got NULL", what, want) { + return + } + assert.Equalf(t, want, *got, "%s", what) +} + +// increasingTIDs returns n real atProto TIDs whose lexicographic order is their +// generation order — what a repo's successive commits actually carry, and what +// the rev gate compares. Invented revs would prove the gate works on invented +// data. +func increasingTIDs(t *testing.T, n int) []string { + t.Helper() + revs := make([]string, n) + for i := range revs { + revs[i] = testkit.TID() + if i > 0 { + require.Greaterf(t, revs[i], revs[i-1], + "testkit.TID must emit lexicographically increasing revs; got %q after %q", revs[i], revs[i-1]) + } + } + return revs +} diff --git a/internal/core/posts/status.go b/internal/core/posts/status.go new file mode 100644 index 0000000..ca851cc --- /dev/null +++ b/internal/core/posts/status.go @@ -0,0 +1,83 @@ +package posts + +import ( + "context" + "time" +) + +// RED STUB (task 5, cycle 1). Signatures only — every method returns zero +// values so the tests that describe this surface compile and fail on their +// assertions rather than on a missing symbol. The implementation is GREEN's. + +// The read side of an admission decision: social.coves.community.post.getStatus +// (docs/PRD_AUTHOR_OWNED_POSTS.md §3.4). +// +// It exists because a rejection is AppView-LOCAL. §3.3 is explicit that a +// submission refused before it was ever accepted writes NO community record — +// spam must not be archived forever in the repo of the community that refused +// it — so there is nothing on the firehose for an author's client to read, and +// "did it get in, and if not, why" has no answer except to ask the host. +// +// It is deliberately its own service rather than a method on Service. Service +// is the write path plus post hydration and is implemented by test doubles all +// over the suite; a status query needs the admissions repository and nothing +// else, and widening the big interface to reach it would make every one of +// those doubles carry a method it has no opinion about. + +// PostStatus is one community's answer about one post. +// +// The optional fields are pointers rather than zero strings because their +// absence is meaningful and different from emptiness: a pending post has no +// decision to report, and rendering an empty code would tell an author their +// post was refused for a reason nobody can name. +type PostStatus struct { + // Status is the admission state (§6.1), verbatim: the same vocabulary the + // admissions table and the consumer speak, so a client that switches on it + // is switching on the real state machine and not a display translation. + Status AdmissionStatus + + // DecisionCode is set for rejected and removed, and only for those. It is + // the vocabulary of DecisionCode — both the codes a community publishes in + // a removal record and the admission-time codes that never reach a repo. + DecisionCode *string + + // DecisionAt is when the decision above was made. + DecisionAt *time.Time + + // AcceptanceURI names the live community acceptance record, so a client can + // go read the signed attestation rather than taking this AppView's word for + // it. Set only while an acceptance stands. + AcceptanceURI *string +} + +// GetStatusRequest names one subject: which community's answer, about which +// post. +// +// Both halves are required and neither has a default. A post can hold +// independent decisions from several communities (§2, forks), so "the status of +// this post" is not a question with one answer, and a request that omitted the +// community would have to invent one. +type GetStatusRequest struct { + PostURI string + CommunityDID string +} + +// StatusService answers getStatus. +type StatusService interface { + // GetStatus returns one community's decision about one post, or ErrNotFound + // when the community has never seen it. + GetStatus(ctx context.Context, req GetStatusRequest) (*PostStatus, error) +} + +type statusService struct { + admissions AdmissionRepository +} + +// NewStatusService wires the status query over the admissions store. +func NewStatusService(admissions AdmissionRepository) StatusService { + return &statusService{admissions: admissions} +} + +func (s *statusService) GetStatus(ctx context.Context, req GetStatusRequest) (*PostStatus, error) { + return nil, nil +} diff --git a/internal/db/postgres/admission_repo_schema_test.go b/internal/db/postgres/admission_repo_schema_test.go index 498d943..7f613fb 100644 --- a/internal/db/postgres/admission_repo_schema_test.go +++ b/internal/db/postgres/admission_repo_schema_test.go @@ -318,13 +318,15 @@ 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, 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. + // The expected-version parameter is the tripwire, and it has now fired + // twice: migration 035 (post_submissions) and 036 (deleted_accounts) both + // sit on top of 034, so both have 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, 36, testkit.MigrateDownOne(t, db, 36), + "036 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") 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") + "035 sits on top of 034 and must be rolled back next; 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") diff --git a/internal/db/postgres/deleted_account_repo.go b/internal/db/postgres/deleted_account_repo.go new file mode 100644 index 0000000..a9d3808 --- /dev/null +++ b/internal/db/postgres/deleted_account_repo.go @@ -0,0 +1,35 @@ +package postgres + +import ( + "context" + "database/sql" +) + +// RED STUB (task 5, cycle 1). Signatures only; the query is GREEN's. + +// DeletedAccountRepository reads the migration-036 erasure markers. +// +// It satisfies jetstream.DeletedAccountLookup structurally rather than by +// importing it: the interface is declared where it is CONSUMED (the ingestion +// consumer), which is what keeps the storage layer from depending on the +// firehose layer for a single method. +type DeletedAccountRepository struct { + db *sql.DB +} + +// NewDeletedAccountRepository wires the lookup over the AppView database. +func NewDeletedAccountRepository(db *sql.DB) *DeletedAccountRepository { + return &DeletedAccountRepository{db: db} +} + +// IsAccountDeleted reports whether this DID names an account the AppView was +// asked to erase. +// +// A query failure must come back as an error, never as false. Under +// author-owned posts an unknown author indexes normally (§5.3), so a false here +// is indistinguishable from a healthy answer — a database blip would silently +// re-index the content a deletion erased, which is the exact outcome the marker +// table exists to prevent. +func (r *DeletedAccountRepository) IsAccountDeleted(ctx context.Context, did string) (bool, error) { + return false, nil +} diff --git a/internal/db/postgres/deleted_accounts_schema_test.go b/internal/db/postgres/deleted_accounts_schema_test.go new file mode 100644 index 0000000..7875f99 --- /dev/null +++ b/internal/db/postgres/deleted_accounts_schema_test.go @@ -0,0 +1,212 @@ +//go:build integration + +package postgres + +import ( + "context" + "testing" + "time" + + "Coves/internal/core/users" + "Coves/tests/fixtures" + "Coves/tests/testkit" + + _ "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Migration 036's marker table, and the thing it exists to make impossible. +// +// Account deletion used to leave NO trace. userRepo.Delete removes the users +// row, the posts, and (since 034) the admission rows — and then the firehose +// redelivers a post event for that same author, or a dead letter for it is +// redriven, and every one of those swept rows comes straight back. The AppView +// re-indexes the content of an account it was asked to erase, and nothing in +// the schema can tell it not to: an absent users row is indistinguishable from +// an author who simply has not been indexed yet, which under author-owned posts +// (§5.3) is a state the consumer is REQUIRED to accept. +// +// deleted_accounts is what makes those two cases distinguishable. A row here +// means "this DID was erased on purpose"; no row means "never seen". The +// ingestion gate that reads it is tested in internal/atproto/jetstream; what is +// tested here is the half that has to be true for the gate to mean anything: +// the marker is written, it is written ATOMICALLY WITH the deletion, and +// re-registration clears it. +// +// See docs/PRD_AUTHOR_OWNED_POSTS.md rev 2.7 (§5 status header). + +const deletedAccountsTable = "deleted_accounts" + +func TestDeletedAccountsTable_Columns(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + + requireTableExists(t, db, deletedAccountsTable) + + type columnShape struct { + dataType string + nullable bool + } + want := map[string]columnShape{ + "did": {"text", false}, + "deleted_at": {"timestamp with time zone", false}, + "deleted_rev": {"text", true}, + } + + rows, err := db.QueryContext(ctx, ` + SELECT column_name, data_type, is_nullable + FROM information_schema.columns + WHERE table_schema = current_schema() AND table_name = $1 + `, deletedAccountsTable) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + got := map[string]columnShape{} + for rows.Next() { + var name, dataType, isNullable string + require.NoError(t, rows.Scan(&name, &dataType, &isNullable)) + got[name] = columnShape{dataType: dataType, nullable: isNullable == "YES"} + } + require.NoError(t, rows.Err()) + + for name, wantShape := range want { + gotShape, ok := got[name] + if !assert.Truef(t, ok, "%s.%s is missing", deletedAccountsTable, name) { + continue + } + assert.Equalf(t, wantShape.dataType, gotShape.dataType, "%s.%s type", deletedAccountsTable, name) + assert.Equalf(t, wantShape.nullable, gotShape.nullable, "%s.%s nullability", deletedAccountsTable, name) + } + + assert.Equal(t, []string{"did"}, primaryKeyColumns(t, db, deletedAccountsTable), + "the DID is the whole key: one marker per account, so a re-delete updates rather than accumulating rows the gate would have to deduplicate") + + // deleted_at is NOT NULL because the marker's only job is to be READ by a + // consumer deciding whether to index an event, and a marker with no time is + // a marker that cannot participate in any retention or audit answer later. + // deleted_rev is nullable because nothing knows the account's repo revision + // at AppView-deletion time — the deletion is a local administrative act, not + // a commit — and a column that had to be filled would be filled with a lie. + assert.Falsef(t, got["deleted_at"].nullable, "deleted_at must be NOT NULL") + assert.Truef(t, got["deleted_rev"].nullable, "deleted_rev must be nullable") +} + +func TestUserRepo_Delete_LeavesADeletionMarker(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + requireTableExists(t, db, deletedAccountsTable) + + handle := testkit.UniqueIDWithPrefix(t, "delmark") + did := fixtures.DID(handle) + fixtures.User(t, db, handle+".test", did) + + before := time.Now().Add(-time.Second) + require.NoError(t, NewUserRepository(db).Delete(ctx, did)) + + var deletedAt time.Time + var deletedRev *string + err := db.QueryRowContext(ctx, + `SELECT deleted_at, deleted_rev FROM deleted_accounts WHERE did = $1`, did, + ).Scan(&deletedAt, &deletedRev) + require.NoErrorf(t, err, + "deleting %s left no marker row. Without one, a redriven post event for this author re-indexes the content the deletion erased, "+ + "and the consumer cannot tell an erased account from one that has simply not been indexed yet", did) + + assert.Truef(t, deletedAt.After(before), "deleted_at (%s) must record when the deletion happened", deletedAt) + assert.Nil(t, deletedRev, + "deleted_rev must be left NULL: the AppView does not know the account's repo revision at deletion time, and inventing one would put a fabricated watermark where a real comparison happens") +} + +func TestUserRepo_Delete_MarkerIsWrittenInTheSameTransaction(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + requireTableExists(t, db, deletedAccountsTable) + + // A DID with content but NO users row. Delete sweeps the content and then + // finds nothing to delete from users, which is the one failure the method + // already reports (users.ErrUserNotFound) — and it makes this the cheapest + // honest probe of atomicity there is. + // + // The claim under test is narrow and load-bearing: the marker INSERT must be + // a statement of the deletion transaction, not a separate write afterwards. + // A marker written outside it survives a rollback, and then a DID that was + // never actually erased is permanently refused by the ingestion gate — the + // account's own future posts stop indexing, silently, with no row anywhere + // explaining why. + name := testkit.UniqueIDWithPrefix(t, "delatomic") + communityDID, err := fixtures.Community(ctx, db, name, "owner"+name) + require.NoError(t, err) + + ghostAuthor := fixtures.DID(testkit.UniqueID(t)) + postURI := "at://" + ghostAuthor + "/social.coves.community.postv2/" + testkit.TID() + _, err = db.ExecContext(ctx, ` + INSERT INTO community_post_admissions (community_did, post_uri, status, created_at, updated_at) + VALUES ($1, $2, 'pending', NOW(), NOW()) + `, communityDID, postURI) + require.NoError(t, err) + + deleteErr := NewUserRepository(db).Delete(ctx, ghostAuthor) + require.ErrorIsf(t, deleteErr, users.ErrUserNotFound, + "fixture: deleting a DID with no users row must fail, which is what gives this test a rolled-back transaction to inspect") + + var markers int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM deleted_accounts WHERE did = $1`, ghostAuthor).Scan(&markers)) + assert.Zerof(t, markers, + "the deletion FAILED and rolled back, but a marker for %s survived: the marker insert is running outside the deletion transaction. "+ + "A marker for an account that still exists is worse than no marker at all — the ingestion gate refuses every future event from a live account", ghostAuthor) + + // And the rollback really did roll back, so the surviving marker above could + // only have come from a write outside the transaction. + var admissions int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM community_post_admissions WHERE post_uri = $1`, postURI).Scan(&admissions)) + require.Equal(t, 1, admissions, + "fixture: the failed deletion must have rolled its content sweep back, or this test proves nothing about where the marker was written") +} + +func TestUserRepo_Create_ClearsTheDeletionMarker(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + requireTableExists(t, db, deletedAccountsTable) + + // Re-registration is the marker's exit. A DID that comes back — the same + // person signing up again on the same PDS, or an account restored after a + // mistaken deletion — must index normally, and a marker left behind would + // make the AppView refuse their new posts forever with nothing to show for + // it. Create is the assertion point because it is the one statement both + // service paths funnel through: IndexUser calls CreateUser (service.go:457) + // and RegisterAccount ends in the same repository insert. + repo := NewUserRepository(db) + + handle := testkit.UniqueIDWithPrefix(t, "rereg") + did := fixtures.DID(handle) + fixtures.User(t, db, handle+".test", did) + require.NoError(t, repo.Delete(ctx, did)) + + var markers int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM deleted_accounts WHERE did = $1`, did).Scan(&markers)) + require.Equal(t, 1, markers, "fixture: the deletion must have left a marker for the re-registration to clear") + + _, err := repo.Create(ctx, &users.User{ + DID: did, + Handle: handle + ".test", + PDSURL: testkit.Endpoints().PDS.BaseURL, + }) + require.NoError(t, err, "a deleted DID must be able to register again") + + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM deleted_accounts WHERE did = $1`, did).Scan(&markers)) + assert.Zerof(t, markers, + "re-registering %s left the deletion marker standing. The ingestion gate reads this table, so the account would index its profile and then have every post it writes silently dropped", did) +}