From 9978064af343b1adbda8731303a714ea82933e82 Mon Sep 17 00:00:00 2001 From: Bretton Date: Wed, 29 Jul 2026 11:27:55 -0700 Subject: [PATCH] =?UTF-8?q?test:=20post=20strangler=20=E2=80=94=20intra-re?= =?UTF-8?q?po=20spoof=20proof,=20consumer=20gates=20pinned,=202=20defects?= =?UTF-8?q?=20filed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 4 task 12. post_e2e_test.go and post_delete_test.go die (net −1,402 LOC, coverage up): consumer duplicate-create/duplicate-delete tombstone dating and author-not-found-is-transient-with-replay pinned at T1; service write-forward + full delete-authorization matrix incl. PDS-record-survives; T2 ingestion contract for social.coves.community.post with an INTRA-repo repo-ownership spoof (review traced Jetstream's parallel scheduler: cross- repo ordering does not exist, so the forged post and its bounding create share one repo — mutation-tested to fail loudly when the check is removed) plus batch-read semantics, DualAuth 401s, and author-feed reads. Two more production defects found and filed: deleted posts remain fully served via getComments (missing deleted_at filter), and DeletePost's idempotency branch is unreachable (PDS answers 400, not the expected not-found) — pinned to fail loudly when fixed. The user-journey starvation flake is mitigated (ordered 60s/75s waits + a starvation-vs-dead-firehose tally) pending its task-16 rebuild. make ci green: 3496 tests, 0 skips. Co-Authored-By: Claude Fable 5 --- .../error_taxonomy_transient_test.go | 66 ++ .../atproto/jetstream/post_delete_test.go | 172 ++++ .../communities/service_provisioning_test.go | 9 + internal/core/posts/harness_test.go | 30 + .../core/posts/service_writeforward_test.go | 346 +++++++ loop_state.md | 23 +- tests/ci/pending_contracts.txt | 1 - tests/e2e/post_contract_test.go | 592 ++++++++++++ tests/integration/helpers.go | 43 - tests/integration/post_delete_test.go | 841 ------------------ tests/integration/post_e2e_test.go | 662 -------------- tests/integration/user_journey_e2e_test.go | 151 +++- 12 files changed, 1382 insertions(+), 1554 deletions(-) create mode 100644 internal/atproto/jetstream/post_delete_test.go create mode 100644 internal/core/posts/harness_test.go create mode 100644 internal/core/posts/service_writeforward_test.go create mode 100644 tests/e2e/post_contract_test.go delete mode 100644 tests/integration/post_delete_test.go delete mode 100644 tests/integration/post_e2e_test.go diff --git a/internal/atproto/jetstream/error_taxonomy_transient_test.go b/internal/atproto/jetstream/error_taxonomy_transient_test.go index 311e602..ca7c5a1 100644 --- a/internal/atproto/jetstream/error_taxonomy_transient_test.go +++ b/internal/atproto/jetstream/error_taxonomy_transient_test.go @@ -6,6 +6,7 @@ import ( "context" "testing" + "Coves/internal/core/users" "Coves/internal/db/postgres" "Coves/tests/testkit" @@ -43,6 +44,71 @@ func TestPostConsumer_CommunityNotFound_IsTransient(t *testing.T) { assert.Contains(t, err.Error(), "community not found") } +// The post consumer's OTHER ordering gate, and the one with a genuine chance of +// firing in production: BigSky preserves commit order within a repo, not across +// repos, so a post in a community's repo can reach the AppView before the +// author's own social.coves.actor.signup does. +// +// Its classification is the whole point. An unknown author is refused — the +// consumer will not index a post for a DID it has never seen — but refused +// TRANSIENTLY, so the event dead-letters with redrive attempts remaining and +// succeeds on replay once the author arrives. Wrapping it as ErrPermanentEvent +// would look like a tightening of the same security check and would instead +// discard every post that merely arrived early. +// +// The second half is what makes that claim more than a spelling assertion: the +// identical event, replayed after the author is indexed, is accepted. That is +// the redrive the transient classification promises. +func TestPostConsumer_AuthorNotFound_IsTransientAndSucceedsOnReplay(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + const ( + community = "did:plc:jstaxauthorgatecomm" + lateAuthor = "did:plc:jstaxlateauthor" + ) + insertBridgedUser(t, db, "did:plc:jstaxgateowner", "gateowner.test") + insertBridgedCommunity(t, db, community, "gatecommunity.test", "did:plc:jstaxgateowner") + + us := newMockUserService() + c := NewPostEventConsumer(postgres.NewPostRepository(db), postgres.NewCommunityRepository(db), us, db) + + event := taxonomyEvent( + community, "social.coves.community.post", "create", "authorgate", + map[string]interface{}{ + "$type": "social.coves.community.post", + "community": community, + "author": lateAuthor, + "title": "arrived before its author", + "createdAt": "2026-01-01T00:00:00Z", + }, + ) + + err := c.HandleEvent(context.Background(), event) + require.Error(t, err, "a post whose author has never been seen must not be indexed") + assert.NotErrorIs(t, err, ErrPermanentEvent, + "author-not-found is an ORDERING failure and must stay transient so the redrive can succeed") + assert.Contains(t, err.Error(), "author not found") + + uri := "at://" + community + "/social.coves.community.post/authorgate" + var rows int + require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM posts WHERE uri = $1`, uri).Scan(&rows)) + require.Equal(t, 0, rows, "the rejected post must not have been indexed") + + // The author signs up; the dead-lettered event is redriven. Both halves of + // "indexed" are needed and they are not the same thing: the consumer asks + // the user service, and the posts table's fk_author constraint asks the + // users table — an author the service knows but the database does not still + // fails, one layer further down. + insertBridgedUser(t, db, lateAuthor, "lateauthor.test") + us.users[lateAuthor] = &users.User{DID: lateAuthor, Handle: "lateauthor.test"} + require.NoError(t, c.HandleEvent(context.Background(), event), + "the same event must be accepted once the author is indexed — that is what 'transient' buys") + + require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM posts WHERE uri = $1`, uri).Scan(&rows)) + assert.Equal(t, 1, rows, "the redriven post must be indexed exactly once") +} + func TestCommunityConsumer_SubscriptionCommunityNotFound_IsTransient(t *testing.T) { t.Parallel() db := testkit.DB(t) diff --git a/internal/atproto/jetstream/post_delete_test.go b/internal/atproto/jetstream/post_delete_test.go new file mode 100644 index 0000000..9b196d5 --- /dev/null +++ b/internal/atproto/jetstream/post_delete_test.go @@ -0,0 +1,172 @@ +//go:build integration + +package jetstream + +import ( + "context" + "database/sql" + "testing" + "time" + + "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" +) + +// What a post DELETE event does to the row, which is the half of deletion that +// no serving endpoint can show. +// +// social.coves.community.post's ingestion contract (tests/e2e/post_contract_test.go) +// proves the observable half: after a delete the post is gone from +// social.coves.community.post.get and stays gone. It cannot distinguish the two +// ways of being gone, because both look identical through the endpoint — +// GetViewsByURIs filters `deleted_at IS NULL`, so a hard-deleted row and a +// soft-deleted one both come back as notFoundPost. +// +// The difference matters. deletePost sets deleted_at rather than removing the +// row (post_consumer.go), and that choice is load-bearing in three places: +// comment threads keep their parent, the rev gate has a row to hang a tombstone +// on so a replayed create cannot resurrect the post, and moderation can still +// see what was published. A refactor to a hard DELETE would keep every T2 +// assertion green and quietly break all three. +// +// These are the assertions that were inside tests/integration/post_delete_test.go's +// TestPostDeletion_JetstreamConsumer, rewritten against the real database at the +// tier where the row is visible. + +const ( + delTestPrefix = "did:plc:jsdel" + delTestCommunity = delTestPrefix + "community" + delTestAuthor = delTestPrefix + "author" + + // Successive commits of one repo. TIDs are lexicographically ordered, which + // is what the rev gate compares. + delRevCreate = "3ldeltestaa2a" + delRevDelete = "3ldeltestaa2b" + delRevLater = "3ldeltestaa2c" +) + +// newDeleteFixture indexes the user and community a post needs and returns a +// consumer wired to them. +func newDeleteFixture(t *testing.T, db *sql.DB) *PostEventConsumer { + t.Helper() + insertBridgedUser(t, db, delTestAuthor, "delauthor.test") + insertBridgedCommunity(t, db, delTestCommunity, "delcommunity.test", delTestAuthor) + + us := newMockUserService() + us.users[delTestAuthor] = &users.User{DID: delTestAuthor, Handle: "delauthor.test"} + return NewPostEventConsumer(postgres.NewPostRepository(db), postgres.NewCommunityRepository(db), us, db) +} + +// delPostEvent builds a post commit for rkey with the given operation and rev. +// A delete carries no record, exactly as Jetstream delivers it. +func delPostEvent(op, rkey, rev, cid string, timeUS int64) *JetstreamEvent { + var record map[string]interface{} + if op != "delete" { + record = map[string]interface{}{ + "$type": "social.coves.community.post", + "community": delTestCommunity, + "author": delTestAuthor, + "title": "delete target", + "content": "body that must survive the tombstone", + "createdAt": "2026-03-01T00:00:00Z", + } + } + return revCommitEvent(delTestCommunity, "social.coves.community.post", op, rkey, rev, cid, timeUS, record) +} + +// readDeletedPost returns the row's tombstone and content. A missing row is a +// fatal error rather than a nil result: every caller here has just asserted the +// post exists, so its absence is the hard-delete regression these tests exist +// to catch, and it should say so at the point it happens. +func readDeletedPost(t *testing.T, db *sql.DB, uri string) (deletedAt *time.Time, title, content string) { + t.Helper() + err := db.QueryRow( + `SELECT deleted_at, title, content FROM posts WHERE uri = $1`, uri, + ).Scan(&deletedAt, &title, &content) + require.NoErrorf(t, err, "the post row for %s is gone: a delete must SOFT-delete (set deleted_at), "+ + "not remove the row — comment threads, the rev-gate tombstone and moderation all read it", uri) + return deletedAt, title, content +} + +func TestPostConsumer_Delete_IsSoftAndKeepsTheRow(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + c := newDeleteFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + uri := "at://" + delTestCommunity + "/social.coves.community.post/delsoft" + + require.NoError(t, c.HandleEvent(ctx, delPostEvent("create", "delsoft", delRevCreate, "bafdelsoft", base))) + + deletedAt, _, _ := readDeletedPost(t, db, uri) + require.Nil(t, deletedAt, "fixture: a freshly indexed post is not deleted") + + require.NoError(t, c.HandleEvent(ctx, delPostEvent("delete", "delsoft", delRevDelete, "", base+1_000_000))) + + deletedAt, title, content := readDeletedPost(t, db, uri) + require.NotNil(t, deletedAt, "the delete event must set deleted_at") + assert.Equal(t, "delete target", title, + "a soft delete must not blank the content: the row is what moderation and the thread view read") + assert.Equal(t, "body that must survive the tombstone", content) +} + +func TestPostConsumer_DuplicateDelete_TombstoneStaysAtTheFirstDelete(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + c := newDeleteFixture(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + uri := "at://" + delTestCommunity + "/social.coves.community.post/deldup" + + require.NoError(t, c.HandleEvent(ctx, delPostEvent("create", "deldup", delRevCreate, "bafdeldup", base))) + + deleteEvent := delPostEvent("delete", "deldup", delRevDelete, "", base+1_000_000) + require.NoError(t, c.HandleEvent(ctx, deleteEvent)) + first, _, _ := readDeletedPost(t, db, uri) + require.NotNil(t, first) + + // The rewind duplicate: the connector rewinds its cursor 5s after every + // reconnect, so the IDENTICAL commit — same rev — is guaranteed to be + // redelivered in production. + require.NoError(t, c.HandleEvent(ctx, deleteEvent), + "a redelivered delete must be a silent no-op, not an error the connector logs as a failure") + + // And the other shape of a repeat: a genuinely later delete commit for a + // record that is already tombstoned. This one clears the rev gate (strictly + // newer rev), so only the UPDATE's own `deleted_at IS NULL` guard stands + // between it and a moved timestamp. + require.NoError(t, c.HandleEvent(ctx, delPostEvent("delete", "deldup", delRevLater, "", base+2_000_000))) + + second, _, _ := readDeletedPost(t, db, uri) + assert.Equal(t, first.UTC(), second.UTC(), + "deleted_at must record when the post was FIRST deleted; a repeated delete that moves it "+ + "would keep re-dating the tombstone every time the connector rewinds") +} + +func TestPostConsumer_DuplicateCreate_IndexesExactlyOnce(t *testing.T) { + t.Parallel() + db := testkit.DB(t) + + c := newDeleteFixture(t, db) + ctx := context.Background() + uri := "at://" + delTestCommunity + "/social.coves.community.post/dupcreate" + + // The post sibling of TestVoteConsumer_DuplicateCreate_IncrementsExactlyOnce + // and TestCommentConsumer_DuplicateCreate_CountsExactlyOnce: the same commit + // delivered twice by a cursor rewind. + event := delPostEvent("create", "dupcreate", delRevCreate, "bafdupcreate", time.Now().UnixMicro()) + require.NoError(t, c.HandleEvent(ctx, event)) + require.NoError(t, c.HandleEvent(ctx, event), + "a redelivered post create must be a silent no-op") + + var rows int + require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM posts WHERE uri = $1`, uri).Scan(&rows)) + assert.Equal(t, 1, rows, "a duplicate create must not produce a second post row") +} diff --git a/internal/core/communities/service_provisioning_test.go b/internal/core/communities/service_provisioning_test.go index 584f0ff..c9a2083 100644 --- a/internal/core/communities/service_provisioning_test.go +++ b/internal/core/communities/service_provisioning_test.go @@ -114,6 +114,15 @@ func TestService_CreateProvisionsAResolvableAccount(t *testing.T) { assert.Equal(t, "at://"+community.DID+"/social.coves.community.profile/self", community.RecordURI) assert.Equal(t, community.DID, community.OwnerDID, "V2: a community owns itself") + // Provisioning must also hand back a usable SESSION on the new account, not + // only its password. Every later write into this repo — a post, a moderation + // action — goes out on these credentials after EnsureFreshToken, and the + // refresh token is the only thing that keeps that working past the access + // token's lifetime. A provisioner that stored one and dropped the other would + // look healthy here and fail hours later, on the first refresh. + assert.NotEmpty(t, community.PDSAccessToken, "provisioning must return an access token for the community's repo") + assert.NotEmpty(t, community.PDSRefreshToken, "without a refresh token the community's credentials expire unrecoverably") + record := pdsServer.Login(t, expectedHandle, community.PDSPassword). GetRecord(t, "social.coves.community.profile", "self") assert.Equal(t, name, record.Value["name"]) diff --git a/internal/core/posts/harness_test.go b/internal/core/posts/harness_test.go new file mode 100644 index 0000000..eea7681 --- /dev/null +++ b/internal/core/posts/harness_test.go @@ -0,0 +1,30 @@ +//go:build integration + +package posts_test + +import ( + "os" + "testing" + + "Coves/tests/testkit" +) + +// TestMain sets the infrastructure floor for this package's integration build. +// +// It lives in a tagged file because a TestMain applies to the whole test +// binary: the untagged unit build of this package (blob_transform_test.go, +// embed_validation_test.go, service_get_posts_test.go and friends, all in +// package posts) needs nothing out of process and must not be made to probe +// Postgres and a PDS before it can run. A future test file here must therefore +// NOT declare a second TestMain — with -tags integration both halves compile +// into one binary, and two TestMains do not. +// +// The tests are in package posts_test (external) rather than in posts, because +// they exercise the service against the real repositories in +// internal/db/postgres, which imports posts. In-package that is an import +// cycle; from outside it is an ordinary dependency. This mirrors +// internal/core/communities, whose integration tests are package +// communities_test for the same reason. +func TestMain(m *testing.M) { + os.Exit(testkit.Main(m, testkit.RequirePostgres, testkit.RequirePDS)) +} diff --git a/internal/core/posts/service_writeforward_test.go b/internal/core/posts/service_writeforward_test.go new file mode 100644 index 0000000..73fffe7 --- /dev/null +++ b/internal/core/posts/service_writeforward_test.go @@ -0,0 +1,346 @@ +//go:build integration + +package posts_test + +import ( + "context" + + "testing" + + "Coves/internal/api/middleware" + "Coves/internal/atproto/pds" + "Coves/internal/core/communities" + "Coves/internal/core/posts" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + "github.com/bluesky-social/indigo/atproto/auth/oauth" + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// What creating and deleting a post actually does to the community's repo, and +// who is allowed to do it. +// +// This is the client-write half of the post domain: the half that +// tests/e2e/post_contract_test.go structurally cannot reach. §3.4b of +// docs/TEST_ARCHITECTURE.md records why — RequireAuth accepts only a sealed +// session token, minted nowhere but the browser OAuth callback, so T2 can prove +// that the write endpoints refuse an unauthenticated client and nothing beyond +// it. Authenticated write BEHAVIOUR is therefore proven here, against a real +// PDS, and these are the assertions that used to live in +// tests/integration/post_e2e_test.go's "Write-Forward to PDS" and +// tests/integration/post_delete_test.go's authorization tests. +// +// # WHAT MAKES THE REPO THE INTERESTING PART +// +// A post record does not live in its author's repo. It lives in the COMMUNITY's +// repo, written with the community's own PDS credentials, carrying an `author` +// field that names the human who wrote it (internal/core/posts/service.go step +// 9, and the reason the Jetstream consumer's first security check is +// repoDID == record.community). +// +// That makes two things testable only from the PDS side. First, that the +// service really writes into the community's repository rather than the +// caller's — a post written to the author's repo would be rejected outright by +// the consumer, so the AppView would simply never index it and every test that +// only reads the service's return value would still pass. Second, that deletion +// is authorized against the RECORD's author field rather than against anything +// the caller supplies: DeletePost fetches the record from the PDS specifically +// to read `author` out of it, because the community's credentials would happily +// delete anyone's post. + +const ( + // The instance identity these tests provision communities under. It matches + // the AppView's default (internal/config: INSTANCE_DID defaults to + // did:web:coves.social), and the domain must be one of the PDS' + // PDS_SERVICE_HANDLE_DOMAINS or account creation is refused. + instanceDID = "did:web:coves.social" + instanceDomain = "coves.social" + + postCollection = "social.coves.community.post" +) + +// postFixture is the post service wired the way cmd/server wires it, over a +// real community that owns a real PDS repository. +type postFixture struct { + service posts.Service + pds *testkit.PDS + community *communities.Community + author *testkit.Account +} + +// newPostFixture provisions a community on the test PDS and returns the post +// service pointed at it. +// +// The community is provisioned through communities.CreateCommunity rather than +// seeded into the index, because unlike the subscribe/block write-forwards this +// is a write into the COMMUNITY's repo: it needs an account that exists on the +// PDS and credentials the AppView can use, and both are what provisioning +// produces. The optional collaborators (aggregators, blobs, unfurl, bluesky) +// are nil — every one of them is a branch on the record's contents, and what is +// under test here is where the record lands. +func newPostFixture(t *testing.T) *postFixture { + t.Helper() + + db := testkit.DB(t) + pdsServer := testkit.NewPDS(t) + communityRepo := postgres.NewCommunityRepository(db) + + communityService := communities.NewCommunityServiceWithPDSFactory( + communityRepo, + pdsServer.URL(), + instanceDID, + instanceDomain, + communities.NewPDSAccountProvisioner(instanceDomain, pdsServer.URL()), + testkit.PasswordAuthFactory(pds.NewFromAccessToken), + nil, + ) + + author := pdsServer.CreateAccount(t, testkit.WithHandlePrefix("pa")) + + name := testkit.UniqueIDWithPrefix(t, "pw") + require.LessOrEqualf(t, len("c-"+name), testkit.MaxIDLength, + "the generated community name %q makes a handle label the PDS will refuse", name) + + community, err := communityService.CreateCommunity(context.Background(), communities.CreateCommunityRequest{ + Name: name, + DisplayName: "Write forward", + Description: "a community whose repo receives posts", + Visibility: "public", + CreatedByDID: author.DID, + }) + require.NoError(t, err) + + return &postFixture{ + service: posts.NewPostService( + postgres.NewPostRepository(db), communityService, + nil, nil, nil, nil, pdsServer.URL()), + pds: pdsServer, + community: community, + author: author, + } +} + +// communityAccount returns a session on the community's own repo, which is how +// a test reads back what the service wrote there. +func (f *postFixture) communityAccount(t *testing.T) *testkit.Account { + t.Helper() + return f.pds.Login(t, f.community.Handle, f.community.PDSPassword) +} + +// createPost writes a post as authorDID. The DID goes into the context as well +// as the request: the service re-checks the two against each other +// (defence-in-depth against a bypassed handler), so a test that sets only one +// exercises the mismatch guard by accident. +func (f *postFixture) createPost(t *testing.T, authorDID, title, content string) *posts.CreatePostResponse { + t.Helper() + resp, err := f.service.CreatePost( + middleware.SetTestUserDID(context.Background(), authorDID), + posts.CreatePostRequest{ + Community: f.community.DID, + Title: &title, + Content: &content, + AuthorDID: authorDID, + }) + require.NoError(t, err) + require.NotEmpty(t, resp.URI) + require.NotEmpty(t, resp.CID) + return resp +} + +// sessionFor builds the OAuth session shape DeletePost takes. Only the DID is +// load-bearing: the delete itself goes out on the COMMUNITY's credentials, and +// the session exists to say who is asking. +func sessionFor(t *testing.T, account *testkit.Account, hostURL string) *oauth.ClientSessionData { + t.Helper() + did, err := syntax.ParseDID(account.DID) + require.NoError(t, err) + return &oauth.ClientSessionData{ + AccountDID: did, + SessionID: "post-write-forward-test", + HostURL: hostURL, + AccessToken: account.AccessToken, + } +} + +// rkeyOf returns the record key an AT-URI ends with. +func rkeyOf(t *testing.T, uri string) string { + t.Helper() + parsed, err := syntax.ParseATURI(uri) + require.NoErrorf(t, err, "the service returned an unparseable record URI %q", uri) + rkey := parsed.RecordKey().String() + require.NotEmptyf(t, rkey, "the record URI %q has no record key", uri) + return rkey +} + +func TestService_CreateWritesThePostIntoTheCommunityRepo(t *testing.T) { + t.Parallel() + + f := newPostFixture(t) + resp := f.createPost(t, f.author.DID, "write-forward title", "write-forward body") + + // The authority of the URI is the community, not the author. This is the + // single most consequential fact about a post record's location: the + // consumer rejects any post whose repo DID differs from its community + // field, so a service that wrote to the author's repo would produce posts + // that never index, with no error anywhere on the write path. + rkey := rkeyOf(t, resp.URI) + assert.Equal(t, "at://"+f.community.DID+"/"+postCollection+"/"+rkey, resp.URI) + + record := f.communityAccount(t).GetRecord(t, postCollection, rkey) + + // The CID the service reported is the CID of the record that actually + // committed. Worth asserting rather than merely checking it is non-empty: + // the response's CID is what a client uses to build a strongRef — a vote or + // a comment's parent reference — so a service returning a stale or invented + // one would produce references that resolve to nothing, and nothing on the + // write path would notice. + assert.Equal(t, record.CID, resp.CID, + "the CID returned to the client must be the committed record's") + + assert.Equal(t, postCollection, record.Value["$type"]) + assert.Equal(t, f.community.DID, record.Value["community"], + "the record's community field must match the repo it lives in, or the consumer rejects it as a spoof") + assert.Equal(t, f.author.DID, record.Value["author"], + "posts live in the community's repo but belong to their author, and this field is the only thing that says so") + assert.Equal(t, "write-forward title", record.Value["title"]) + assert.Equal(t, "write-forward body", record.Value["content"]) + assert.NotEmpty(t, record.Value["createdAt"]) +} + +func TestService_DeleteRemovesTheRecordFromTheCommunityRepo(t *testing.T) { + t.Parallel() + + f := newPostFixture(t) + ctx := context.Background() + resp := f.createPost(t, f.author.DID, "to be deleted", "body") + rkey := rkeyOf(t, resp.URI) + + require.NoError(t, f.service.DeletePost(ctx, sessionFor(t, f.author, f.pds.URL()), + posts.DeletePostRequest{URI: resp.URI})) + + community := f.communityAccount(t) + assert.True(t, testkit.IsNotFound(getRecordErr(ctx, community, postCollection, rkey)), + "the post record is still in the community's repo after its author deleted it") + + // KNOWN DEFECT, pinned as it behaves rather than as it is meant to. + // + // DeletePost intends a repeated delete to be idempotent: it checks the + // record fetch for pds.ErrNotFound and returns nil, commented "Post already + // deleted or never existed - idempotent success" (service.go step 7). That + // branch is unreachable against this PDS. com.atproto.repo.getRecord answers + // a missing record with HTTP 400 and "Could not locate record", and + // pds/client.go maps 400 to ErrBadRequest — so the not-found check misses, + // and the delete a client retries after a lost response comes back as an + // opaque failure the handler renders as a 500. + // + // Asserting the intent here would fail the suite over a production bug this + // task is not fixing; asserting nothing would let the bug become invisible. + // So the assertion is the current truth, and it is written to FAIL LOUDLY + // the moment the classification is fixed — at which point this block becomes + // assert.NoError and the comment goes away. + err := f.service.DeletePost(ctx, sessionFor(t, f.author, f.pds.URL()), + posts.DeletePostRequest{URI: resp.URI}) + require.Errorf(t, err, "the idempotent-delete defect appears to be FIXED: "+ + "replace this block with assert.NoError and delete the KNOWN DEFECT comment above it") + assert.Contains(t, err.Error(), "Could not locate record", + "the repeated delete failed for a different reason than the known not-found misclassification") +} + +func TestService_DeleteRefusesEveryoneButTheAuthor(t *testing.T) { + t.Parallel() + + f := newPostFixture(t) + ctx := context.Background() + resp := f.createPost(t, f.author.DID, "the author's post", "body") + rkey := rkeyOf(t, resp.URI) + + // The attacker is a fully legitimate account with a real session. What they + // do not have is the record's author field — and since the delete goes out + // on the COMMUNITY's credentials rather than on theirs, that field is the + // ONLY thing standing between them and deleting someone else's post. + attacker := f.pds.CreateAccount(t, testkit.WithHandlePrefix("atk")) + + err := f.service.DeletePost(ctx, sessionFor(t, attacker, f.pds.URL()), + posts.DeletePostRequest{URI: resp.URI}) + require.ErrorIs(t, err, posts.ErrNotAuthorized, + "a user who is not the post's author must be refused") + + // And refused means refused: the record is still there. + community := f.communityAccount(t) + record := community.GetRecord(t, postCollection, rkey) + assert.Equal(t, f.author.DID, record.Value["author"], + "the rejected delete removed the record anyway") +} + +func TestService_DeleteRejectsMalformedRequests(t *testing.T) { + t.Parallel() + + f := newPostFixture(t) + ctx := context.Background() + session := sessionFor(t, f.author, f.pds.URL()) + + // Everything that must fail before the service touches the network. Each + // case is a client mistake, and each must be answerable without the PDS + // having been asked anything — a validation error, not a 500 from a failed + // fetch of a nonsense URI. + for _, tc := range []struct { + name string + session *oauth.ClientSessionData + uri string + }{ + {name: "no session", session: nil, uri: "at://" + f.community.DID + "/" + postCollection + "/abc"}, + {name: "empty URI", session: session, uri: ""}, + {name: "not an AT-URI", session: session, uri: "invalid-uri-format"}, + {name: "wrong collection", session: session, uri: "at://" + f.community.DID + "/social.coves.community.comment/abc"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + err := f.service.DeletePost(ctx, tc.session, posts.DeletePostRequest{URI: tc.uri}) + require.Error(t, err) + assert.Truef(t, posts.IsValidationError(err), + "expected a validation error the handler can turn into a 400, got: %v", err) + }) + } +} + +func TestService_DeleteReportsAnUnknownCommunityAsNotFound(t *testing.T) { + t.Parallel() + + f := newPostFixture(t) + + // A well-formed URI whose authority is a community the AppView has never + // indexed. The distinction from the validation errors above is what the + // handler does with it — 404 rather than 400 — so the sentinel identity is + // the assertion, not merely that an error came back. + // + // The DID is a literal rather than a generated one: did:plc identifiers are + // 24 base32 characters (a-z, 2-7), which UniqueID does not promise, and a DID + // that fails the FORMAT check would take the validation path above instead of + // the lookup path under test. Uniqueness is not needed — the database is a + // per-test clone in which nothing has ever been indexed. + // + // It is spelled at the full 24 characters deliberately. validateDIDFormat + // (service.go) checks the CHARACTER SET but not the length, so a 23-character + // identifier would pass today and start failing the moment that omission is + // corrected — turning this lookup test into a validation test without anyone + // touching it. Fixing the validator is not this task's business; not depending + // on the gap is. + uri := "at://did:plc:aaaaaaaaneverindexedcomm/" + postCollection + "/abc" + err := f.service.DeletePost(context.Background(), sessionFor(t, f.author, f.pds.URL()), + posts.DeletePostRequest{URI: uri}) + assert.ErrorIs(t, err, posts.ErrCommunityNotFound) +} + +// getRecordErr asks the PDS for a record and returns only the error, so a test +// can assert a record's ABSENCE — Account.GetRecord fails the test on a missing +// record, which is the right default and the wrong tool here. +func getRecordErr(ctx context.Context, account *testkit.Account, collection, rkey string) error { + return account.XRPC().Query(ctx, "com.atproto.repo.getRecord", map[string][]string{ + "repo": {account.DID}, + "collection": {collection}, + "rkey": {rkey}, + }, nil) +} diff --git a/loop_state.md b/loop_state.md index 8dc9279..f0b2633 100644 --- a/loop_state.md +++ b/loop_state.md @@ -48,8 +48,8 @@ Stop the loop when every task is done, or on any blocked task. | 9 | Global-state audit (t.Setenv/os.Setenv/logger/http-default → testkit injection); enable t.Parallel on proven-safe; connection budgets; `-race` clean; drop -p 1 | 3 ⛩ | S | done | (see git log) | PHASE 3 COMPLETE. 343 t.Parallel; audit: 0 convert / 4 sites deliberately-serial / rest safe. 9 internal straggler files migrated (goose now EXTINCT in test code; MigrateSharedDatabase deleted). THREE concurrency bugs -p 1 was masking: [A] template-destruction race (fixed: usePrivateTemplate) [B] legacy firehose 5s-behind-30s-promise, quantified (patched: jetstreamReadBudget, counter machinery deleted, non-timeout errors terminate) [C] Jetstream account/identity events BYPASS wantedCollections → parallel signup storms starve subscribers (measured 2/4 fail at -p 2; -p STAYS 1 with new documented reason). ConcurrencyBudget models both dims + nestedClonePools; -p 1 -parallel 26. Review: Codex good + Opus 3-high (binary-abort class, all fixed incl. fail-open Makefile splice PROVEN closed). make ci GREEN ×2 117/128s (clone tax repaid, beats 124s pre-clone); -race + -shuffle clean; peak 27/200 conns; 3401 tests/0 skips; audit 532 | | 10 | Contract-manifest CI check (WantedCollections ↔ //coves:ingestion-contract markers) + T2 skeleton (serial runner via compose runner; make test-e2e; test-e2e-dev escape hatch) | 4 ⛩ | S | done | (see git log) | THE PIPELINE WORKS: TestPipelineSmoke green in hermetic stack (direct PDS write → Jetstream → container consumers → getProfile, 0.95s de-raced). cmd/contract-manifest (38 tests, MatchFile-based, pending_contracts.txt ratchet w/ task ownership, AST forbidden-imports in marker files); T2 skeleton (newPipeline, contractBudget=45s, per-contract synthetic IPv6 vs the ONE-BUCKET rate limiter — A/B proven 60+40=100); make test-e2e via compose runner 48s cold (lib/ci-stack.sh + runner-ready.sh factored); zero-skip T2 enforcement. Census: 10 collections = task mapping exact. Review: Codex needs-work + Opus 3-high → 8 fixes (manifest bypasses had live probes; smoke de-raced vs profile-backfill reconciliation path). make ci GREEN ×2 ~2:00, 3448/0 | | 11 | Contracts: community (community.profile ingestion + API) — strangler: behavior inventory of community_e2e_test.go (1820 LOC) → down-tier T1s → contract → delete old | 4 | S | done | (see git log) | TEMPLATE PROVEN. 22-behavior inventory; 2,168 LOC deleted, +14 net tests; 2/9 serial firehose files gone. Ingestion contract SELF-REGISTERS the community's PDS repo (stronger than arming — no sync write exists; consumer has NO must-know-first gate, verified). FOUND+FILED prod defect: unverifiable handles → handle.invalid UNIQUE squat → federated communities silently dropped; second symptom pds_url permanently empty → BridgeTrust denies bridged votes (issue extended). STANDING TIER LIMIT (spec §3.4b amended): sealed sessions mint only in browser OAuth — T2 covers auth boundary + reads; authenticated writes proven at T1; test-only mint = phase-5 pre-work. Review: Codex 1 high (update-handler boundary died — restored w/ 8 tests) + Opus audit 17/20 equal-or-stronger, 3 gaps all closed. make ci GREEN ×2 3496/0 @2:12 | -| 12 | Contracts: post (community.post) + post_delete + decompose post god-files | 4 | S | pending | | | -| 13 | Contracts: comment (community.comment) + comment god-files (1821+1443+1229+999 LOC) | 4 | S | pending | | biggest decomposition | +| 12 | Contracts: post (community.post) + post_delete + decompose post god-files | 4 | S | done | (see git log) | post_e2e (662) + post_delete (841) deleted, net −1402 w/ more coverage; serial firehose files 7→5. TWO MORE PROD DEFECTS FILED (tally 4): p1 deleted posts served in full by getComments to anon callers forever (repo missing deleted_at filter); p3 DeletePost idempotency branch unreachable (PDS 400 ≠ ErrNotFound — pinned w/ loud require.Errorf). Spoof negative REWRITTEN intra-repo after Opus traced indigo parallel scheduler (cross-repo FIFO doesn't exist; 5s hold was the real bound; relay would break silently) — mutation-tested. Consumer gates pinned: author-not-found=transient+replay-accepted. Journey flake (2/6 gates, note-[C] starvation) mitigated: 60s/75s ordered waits + starvation-vs-dead tally; real fix task 16. make ci GREEN 3496/0 @2:21 | +| 13 | Contracts: comment (community.comment) + comment god-files (1821+1443+1229+999 LOC) | 4 | S | in-progress | | biggest decomposition | | 14 | Contracts: vote (feed.vote) + user (actor.profile incl. avatar blob path) + subscription (community.subscription) | 4 | S | pending | | vote re-tap idempotency invariant | | 15 | Contracts: blocks (actor.block + community.block) + aggregators (aggregator.service + aggregator.authorization) | 4 | S | pending | | collections revision-1 inventory MISSED — see spec §3.4a | | 16 | Reliability suite (§3.4c: cursor resume, replay-once, rev-gate no-resurrection via Holds, dead-letter, 2-feed overlap) + user_journey rebuild + rm -rf tests/integration | 4 ⛩ | F | pending | | FULL PANEL on the whole Phase-4 output | @@ -262,3 +262,22 @@ Stop the loop when every task is done, or on any blocked task. createdByDid — pinned in tests, unify someday. internal/core/communities tests are package communities_test (external, import cycle) — task 17 must NOT add a second TestMain in package communities. +- **From task 12 (for tasks 13-15)**: endpoint NOT-FOUND SHAPES differ — + community.get 404s; post.get answers 200 + notFound union member + (PendingIfNotFound useless there; probe the field); actor feeds 200+empty + for unknown DID. Spike the shape first. did:plc "nobody" fixtures must be + REAL 24-char base32 literals (UniqueID doesn't emit base32; validator + currently omits the length check — don't lean on it). NEGATIVE BOUNDS + must be INTRA-REPO (Jetstream parallelizes across repos — write the bad + record and the bounding good record into the SAME repo; cross-repo + ordering is topology luck). Unresolvable-handle 404 branch unreachable + under egress block (Phase-5 topology). Contracts leave one retired dead + letter per run on kept stacks (task 16: know before asserting counts). + internal/core/posts tests = package posts_test external (no second + TestMain in package posts). FLAKE LEDGER: TestFullUserJourney_E2E + starvation flake MITIGATED (60s/75s + tally); if it flakes again in + 13-15 → pull task 16 forward, do NOT raise constants again. git stash + is unreliable in this worktree (index merge errors) — back up files + before A/B tests. Serial firehose files remaining: 5 (comment_e2e, + community_avatar_e2e, user_journey_e2e, user_profile_avatar_e2e, + vote_e2e). diff --git a/tests/ci/pending_contracts.txt b/tests/ci/pending_contracts.txt index e70e4c6..d70f207 100644 --- a/tests/ci/pending_contracts.txt +++ b/tests/ci/pending_contracts.txt @@ -19,7 +19,6 @@ # Task 20 empties this file and flips cmd/contract-manifest to # -allow-pending=false, after which "has a contract" is the only passing state. -social.coves.community.post # task 12: post ingestion contract, with the post god-file decomposition social.coves.community.comment # task 13: comment ingestion contract, with the comment god-file decomposition social.coves.feed.vote # task 14: vote ingestion contract (the re-tap idempotency invariant lives here) # diff --git a/tests/e2e/post_contract_test.go b/tests/e2e/post_contract_test.go new file mode 100644 index 0000000..5ddfcf7 --- /dev/null +++ b/tests/e2e/post_contract_test.go @@ -0,0 +1,592 @@ +//go:build e2e + +package e2e + +import ( + "context" + "fmt" + "net/http" + "net/url" + "strings" + "testing" + "time" + + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The post domain's pipeline contracts: the ingestion proof for +// social.coves.community.post, and the client-facing surface a third-party +// client actually reaches. +// +// # POSTS LIVE IN THE COMMUNITY'S REPO, WHICH IS MOST OF THE SETUP +// +// A post record is not written to its author's repository. It is written to the +// COMMUNITY's, with the community's own PDS credentials, carrying an `author` +// field naming the human who wrote it. That single fact drives everything below: +// +// - The consumer's first security check is repoDID == record.community +// (post_consumer.go validatePostEvent). A post in any other repo is rejected +// as a spoof, permanently. +// - The community must be INDEXED before the post arrives, or the post is +// rejected — transiently, so a redrive can succeed once the community lands. +// - The author must be indexed too, and identities enter the index only +// through social.coves.actor.signup (contracts_test.go's IndexedAccount says +// why). +// +// So an ingestion contract for posts needs three things standing before it can +// write a single record: a signed-up author, a community whose repo the AppView +// learned about from the firehose, and a session on that community. The first +// two are pipeline proofs of their own domains, which is why this file waits for +// each of them explicitly rather than assuming — a post that never appears +// because its community never indexed would otherwise be reported as a post +// pipeline failure. +// +// # NOTHING BUT THE FIREHOSE CAN PUT A POST IN THE INDEX (checked, see contracts_test.go) +// +// The package doc's reconciliation hazard — code that reads the PDS by itself +// and can satisfy a wait with every consumer dead — comes up empty for posts, +// and more cleanly than it did for communities. The search, recorded so the next +// reader need not repeat it: +// +// - posts.CreatePost writes NOTHING to Postgres. Unlike communities.CreateCommunity +// (the synchronous client path §3.4 warns about), it validates, forwards the +// record to the community's PDS, and returns the URI — "AppView will index +// via Jetstream consumer", service.go. So even the client path is honest here. +// - posts.DeletePost also writes nothing. It does fetch the record from the PDS, +// but only to read `author` out of it for the authorization check; the result +// is discarded. +// - The read paths (posts.GetPosts, timeline, discover, communityFeeds, +// comments) are SELECT-only. None constructs a PDS client. +// - The only INSERT INTO posts reachable in a running server is +// post_consumer.go's, inside the rev-gated transaction. postgres.postRepo's +// own Create and SoftDelete have no non-test callers at all. +// +// A single create → visible observation is therefore already honest, with no +// arming write of the kind the actor.profile contract needs. +// +// # WHAT THE DELETE ASSERTION IS MADE AGAINST, AND WHY IT IS NOT getComments +// +// deletePost SOFT-deletes: `UPDATE posts SET deleted_at = NOW()`, keeping the row +// so comment threads keep their parent and the rev gate has somewhere to hang a +// tombstone. The row's survival is asserted at T1 +// (internal/atproto/jetstream/post_delete_test.go); what belongs here is that the +// deletion is OBSERVABLE — social.coves.community.post.get stops serving the post +// and keeps not serving it. +// +// It is worth saying which endpoint that is measured on, because a second one +// disagrees. social.coves.community.comment.getComments?post= hydrates its +// post context through postRepo.GetByURI, which has no `deleted_at IS NULL` +// filter — so a deleted post is still served in full (title, content, author) by +// the thread endpoint while post.get correctly reports it gone. That is a +// content-visibility defect, reported rather than asserted here: pinning it would +// cement it, and asserting the intended behaviour would fail this suite over a +// bug this task is not fixing. +// +// # THE ONE SEAM NO SINGLE TEST SPANS, AND WHY THAT IS ACCEPTABLE +// +// The file this contract replaces had a test — TestPostE2E_DeleteWithJetstream — +// that ran the whole arc in one process: call the SERVICE's delete, watch the +// firehose, assert the row was soft-deleted. That arc is unsayable at T2 now. The +// service's delete is reached through social.coves.community.post.delete, which +// is behind RequireAuth, and §3.4b's known limitation is that nothing outside the +// browser OAuth callback mints a credential RequireAuth accepts. So the tier can +// prove the endpoint REFUSES an unauthenticated caller and no more. +// +// What covers the seam instead is composition, and the join is tighter than it +// first looks because the two halves meet at a protocol guarantee rather than at +// an assumption: +// +// - internal/core/posts/service_writeforward_test.go proves the service's +// delete removes the record from the community's repo, against a real PDS. +// A repo mutation on a PDS IS a commit on that PDS' firehose — that is what +// a PDS is — so the service's delete necessarily emits a delete commit. +// - TestPostIngestion below proves that a delete commit on exactly that +// collection, in exactly that repo, reaching the AppView's own consumers, +// soft-deletes the post and keeps it gone. +// +// The honest caveat: no single test observes both halves of one delete, so a +// defect that lived precisely in the handoff — the service deleting a DIFFERENT +// rkey than the one it reports, say — would need both tests to be wrong in the +// same direction to escape. The gap closes when the Phase-5 test-only session +// mint lands and the API contract can drive an authenticated delete end to end. +const postCollection = "social.coves.community.post" + +// postView is the slice of social.coves.community.post.get's postView member +// that the contracts observe. As elsewhere in this package, modelling only the +// asserted fields keeps a new lexicon field from breaking every contract that +// reads a post. +type postView struct { + URI string `json:"uri"` + CID string `json:"cid"` + RKey string `json:"rkey"` + Author identityRef `json:"author"` + Community identityRef `json:"community"` + Record map[string]any `json:"record"` + Stats postStats `json:"stats"` + CreatedAt time.Time `json:"createdAt"` + IndexedAt time.Time `json:"indexedAt"` + EditedAt *time.Time `json:"editedAt,omitempty"` + + // NotFound is the discriminator of the notFoundPost union member, which + // shares the array with postView. The endpoint answers 200 either way (a + // missing post is not an error at this endpoint, unlike community.get), so + // this field — not a status code — is how a contract tells them apart. + NotFound bool `json:"notFound"` +} + +// identityRef is the author/community reference a post view carries. +type identityRef struct { + DID string `json:"did"` + Handle string `json:"handle"` + Name string `json:"name"` +} + +type postStats struct { + Upvotes int `json:"upvotes"` + Downvotes int `json:"downvotes"` + Score int `json:"score"` + CommentCount int `json:"commentCount"` +} + +// Posts reads post views from the AppView, in the order the URIs were asked for. +// +// The endpoint takes a REPEATED `uris` parameter rather than a single `uri`, and +// answers with one union member per requested URI, so the result is positional: +// index i is the answer about uris[i], found or not. +func (p *pipeline) Posts(ctx context.Context, uris ...string) ([]postView, error) { + var out struct { + Posts []postView `json:"posts"` + } + params := url.Values{} + for _, uri := range uris { + params.Add("uris", uri) + } + if err := p.AppView.Query(ctx, "social.coves.community.post.get", params, &out); err != nil { + return nil, err + } + return out.Posts, nil +} + +// Post reads one post view. A post that is not indexed comes back as a +// notFoundPost member, NOT as an error — see postView.NotFound. +func (p *pipeline) Post(ctx context.Context, uri string) (postView, error) { + views, err := p.Posts(ctx, uri) + if err != nil { + return postView{}, err + } + if len(views) != 1 { + // Returned rather than fataled, so that inside a probe it is a TERMINAL + // error (§3.3): the endpoint's answer is positional, and a wait that + // retried through a broken one would time out blaming the pipeline for a + // response-shape bug. + return postView{}, fmt.Errorf( + "social.coves.community.post.get answered with %d results for 1 requested URI: "+ + "the response is positional and must carry one union member per requested URI", + len(views)) + } + return views[0], nil +} + +// postURI renders the AT-URI a post record has once it is committed: the +// COMMUNITY's DID is the authority, which is the whole shape of this domain. +func postURI(communityDID, rkey string) string { + return "at://" + communityDID + "/" + postCollection + "/" + rkey +} + +// postRecord builds a social.coves.community.post record in the shape +// internal/core/posts writes it (service.go step 9), so the consumer parses +// exactly what production hands it. +func postRecord(communityDID, authorDID, title, content string) map[string]any { + return map[string]any{ + "$type": postCollection, + "community": communityDID, + "author": authorDID, + "title": title, + "content": content, + "createdAt": time.Now().UTC().Format(time.RFC3339), + } +} + +// indexedCommunity provisions a community's repo, writes its profile, and waits +// for the AppView to have learned about it from the firehose. +// +// Posts, comments and votes all need a community that is INDEXED, not merely +// provisioned — the post consumer refuses a post whose community it has never +// seen — and that wait is a step every one of tasks 12-15 would otherwise +// hand-roll. Left in this file rather than pushed into the community contract's: +// provisionCommunityRepo is the community domain's fixture, and this is what the +// domains hanging off it need on top. +func indexedCommunity(t *testing.T, p *pipeline, prefix, creatorDID string) provisionedCommunity { + t.Helper() + + community := provisionCommunityRepo(t, p, prefix) + community.PutRecord(t, communityProfileCollection, "self", + communityProfile(community, creatorDID, "host "+community.Name, "a community to hang records on", "public")) + + p.Await(t, "the community hosting these records to be indexed", func() (bool, error) { + _, err := p.Community(context.Background(), community.DID) + return testkit.PendingIfNotFound(err) + }) + return community +} + +// TestPostIngestion is the pipeline proof for posts. +// +// coves:ingestion-contract social.coves.community.post +// +// Every record below is written straight into the community's own repo with the +// community's session, and every observation is made through +// social.coves.community.post.get: +// +// spoof → a post claiming ANOTHER community never appears, and stays absent (Holds) +// create → the post appears, carrying the record's own field values +// update → the same URI serves the new values, marked edited +// delete → the post is gone, and STAYS gone (Holds, §3.4a) +// +// # HOW THE NEGATIVE IS BOUNDED, AND WHY THE SPOOF IS SHAPED THE WAY IT IS +// +// "The spoofed post never appears" is the one assertion here that cannot be +// proven by waiting — waiting only ever shows that it has not appeared YET. It +// is bounded instead by a later event: the spoof is written FIRST and the real +// post SECOND, INTO THE SAME REPO, so when the real post is visible the spoof's +// commit has necessarily already been through the same consumer. Only then is +// its absence meaningful, and Holds keeps watching in case a redrive changes its +// mind. +// +// Same repo is the load-bearing word, and it is why the spoof is a community +// forging a post for a DIFFERENT community rather than the more obvious shape (a +// USER writing a post claiming a community). Both trip the identical check — +// post_consumer.go's repoDID != record.community — but only the same-repo +// version has an ordering guarantee behind it: +// +// - the PDS sequencer assigns one monotonic order to a repo's own commits; +// - Jetstream serializes per repo and PARALLELIZES ACROSS repos; +// - the connector hands one feed's events to the consumer sequentially. +// +// So two commits in one repo cannot overtake each other anywhere on the path, +// while two commits in DIFFERENT repos have no such guarantee — a cross-repo +// spoof would be bounded only by the 5-second Holds window and by the firehose +// happening to be idle, and Phase 5's relay topology would break even that, +// silently, leaving a test that still passes and no longer proves anything. +// Please do not "simplify" this back to writing the spoof into the author's repo. +// +// The victim community is real and indexed, not a fabricated DID, so the +// rejection can only be attributed to the repo mismatch: had it been a DID the +// AppView has never seen, the community-not-found gate would reject the record +// first and the ownership check would never be reached. +func TestPostIngestion(t *testing.T) { + p := newPipeline(t) + + author := p.IndexedAccount(t, "pi") + community := indexedCommunity(t, p, "p", author.DID) + victim := indexedCommunity(t, p, "v", author.DID) + + created := "created " + testkit.UniqueID(t) + updated := "updated " + testkit.UniqueID(t) + + // ---- the spoof, written first so the create below bounds it ------------- + // This community's repo, forging a post that claims to belong to the victim + // community. The consumer builds a post's URI from the REPO the commit + // arrived in, so an indexed spoof would land under THIS community's DID — + // which is the URI checked below. + spoofRKey := testkit.TID() + community.PutRecord(t, postCollection, spoofRKey, + postRecord(victim.DID, author.DID, "spoofed", "a post forged for a community that does not own this repo")) + spoofURI := postURI(community.DID, spoofRKey) + + // ---- create ------------------------------------------------------------- + rkey := testkit.TID() + uri := postURI(community.DID, rkey) + record := community.PutRecord(t, postCollection, rkey, + postRecord(community.DID, author.DID, created, "written straight into the community's repo")) + + observe := func(description string, accept func(postView) bool) postView { + t.Helper() + var observed postView + p.Await(t, description, func() (bool, error) { + view, err := p.Post(context.Background(), uri) + if err != nil { + return false, err + } + observed = view + return !view.NotFound && accept(view), nil + }) + return observed + } + + view := observe("the directly-written post to reach social.coves.community.post.get via the consumers", + func(v postView) bool { return v.Record["title"] == created }) + + require.Equal(t, uri, view.URI, "the AppView served a post under a different URI than the record's") + require.Equal(t, record.CID, view.CID, "the indexed CID must be the commit's, not a re-derived one") + require.Equal(t, rkey, view.RKey) + require.Equal(t, author.DID, view.Author.DID, "the author is the record's author field, not the repo owner") + require.Equal(t, author.Handle, view.Author.Handle, + "the author reference is joined to the users table, so an unindexed author would have failed the join") + require.Equal(t, community.DID, view.Community.DID) + require.Equal(t, community.Handle, view.Community.Handle) + require.Equal(t, community.Name, view.Community.Name) + require.Equal(t, postCollection, view.Record["$type"]) + require.Equal(t, community.DID, view.Record["community"]) + require.Equal(t, author.DID, view.Record["author"]) + require.Equal(t, "written straight into the community's repo", view.Record["content"]) + require.Equal(t, postStats{}, view.Stats, + "a post arrives with no votes and no comments; non-zero stats here would mean the consumer invented them") + require.False(t, view.CreatedAt.IsZero()) + require.False(t, view.IndexedAt.IsZero()) + require.Nil(t, view.EditedAt, "a post that has never been edited must not claim an edit time") + + // ---- the spoof, now bounded --------------------------------------------- + // The create above is visible, and it was committed to this same repo AFTER + // the spoof — so the spoof's commit has already been through the consumer. + spoofAbsent := func() (bool, error) { + v, err := p.Post(context.Background(), spoofURI) + if err != nil { + return false, err + } + return v.NotFound, nil + } + absent, err := spoofAbsent() + require.NoError(t, err) + require.Truef(t, absent, + "a post record in %s claiming to belong to community %s was INDEXED: the consumer's "+ + "repo-ownership check (repoDID == record.community) is the only thing stopping any repo "+ + "from publishing posts as any community, and the shipped binary is not applying it", + community.DID, victim.DID) + p.Holds(t, "the spoofed post to stay unindexed", spoofAbsent) + + // And the victim's own feed is untouched: the forged post must not have been + // filed under the community it named, either. + victimPost, err := p.Post(context.Background(), postURI(victim.DID, spoofRKey)) + require.NoError(t, err) + require.Truef(t, victimPost.NotFound, + "the forged post was indexed under the community it CLAIMED (%s) rather than the repo it "+ + "came from — an even worse outcome than indexing it, since the post would appear to be "+ + "the victim community's own", victim.DID) + + // ---- update ------------------------------------------------------------- + // Same rkey, so this is an update commit rather than a second create — the + // consumer's updatePost path, which loads the stored row, refuses community + // or author reassignment, and folds the changed content in. + community.PutRecord(t, postCollection, rkey, + postRecord(community.DID, author.DID, updated, "edited through the firehose")) + + view = observe("the updated post to reach social.coves.community.post.get", + func(v postView) bool { return v.Record["title"] == updated }) + + require.Equal(t, "edited through the firehose", view.Record["content"], + "the update path must carry every changed field, not only the title") + require.Equal(t, uri, view.URI, "an update must edit the post in place, not create a second one") + require.NotEqual(t, record.CID, view.CID, "the update must index the new commit's CID") + require.NotNil(t, view.EditedAt, + "a content edit must set editedAt — it is what distinguishes the update path from a re-create") + + // ---- delete ------------------------------------------------------------- + // DeleteExistingRecord rather than DeleteRecord: deleting a key that is not + // there answers 200 and emits no commit, so a wrong rkey would turn into a + // timeout blaming the firehose (testkit/pds.go). + community.DeleteExistingRecord(t, postCollection, rkey) + + gone := func() (bool, error) { + v, err := p.Post(context.Background(), uri) + if err != nil { + return false, err + } + return v.NotFound, nil + } + p.Await(t, "the deleted post to disappear from social.coves.community.post.get", gone) + p.Holds(t, "the deleted post to stay deleted", gone) +} + +// TestPostAPIContract covers the client-facing surface of the post endpoints as +// a third-party client meets it: what an unauthenticated caller gets from the +// write endpoints, and what any caller can read back about a post that exists. +// +// It carries NO ingestion marker — markers are for pipeline proofs (§3.4a), and +// this asserts the client path. +// +// The authenticated half of both write endpoints is proven at T1, for the reason +// §3.4b records and TestCommunityAPIContract spells out: nothing but the browser +// OAuth callback mints a session RequireAuth accepts. For posts specifically that +// half is internal/core/posts/service_writeforward_test.go (the record the +// service puts in the community's repo, and who may delete it) plus +// tests/integration/post_handler_test.go (handler validation). What this adds is +// the part neither can see — that the shipped binary really routes these NSIDs, +// really guards them, and really serves an indexed post back. +func TestPostAPIContract(t *testing.T) { + p := newPipeline(t) + + author := p.IndexedAccount(t, "pa") + community := indexedCommunity(t, p, "a", author.DID) + + title := "api contract " + testkit.UniqueID(t) + first := testkit.TID() + second := testkit.TID() + firstURI, secondURI := postURI(community.DID, first), postURI(community.DID, second) + + community.PutRecord(t, postCollection, first, + postRecord(community.DID, author.DID, title, "read back through the client surface")) + community.PutRecord(t, postCollection, second, + postRecord(community.DID, author.DID, title+" (second)", "the batch's other member")) + + p.Await(t, "both posts to be indexed before the client surface is exercised", func() (bool, error) { + views, err := p.Posts(context.Background(), firstURI, secondURI) + if err != nil { + return false, err + } + return !views[0].NotFound && !views[1].NotFound, nil + }) + + ctx, cancel := context.WithTimeout(context.Background(), contractBudget) + defer cancel() + + t.Run("the write endpoints refuse an unauthenticated client", func(t *testing.T) { + // One request each, no polling: this is the auth boundary of the shipped + // router, and the answer does not become true later. + // + // Both post NSIDs that RegisterPostRoutes puts behind RequireAuth are + // listed. Asserting it HERE rather than only in the handler tests is the + // point: a handler test proves the handler refuses an unauthenticated + // call, and structurally cannot see a route registered without the + // middleware in front of it. Only a request to the running router can. + // + // post.create is additionally the one Coves route behind DualAuth (OAuth + // users OR service-JWT aggregators), so it has two ways to be let in and + // a correspondingly better chance of a wiring change opening it. + for _, endpoint := range []struct { + nsid string + input map[string]any + }{ + {"social.coves.community.post.create", map[string]any{ + "community": community.DID, "title": "nope", "content": "nope"}}, + {"social.coves.community.post.delete", map[string]any{"uri": firstURI}}, + } { + err := p.AppView.Procedure(ctx, endpoint.nsid, endpoint.input, nil) + require.Truef(t, testkit.IsStatus(err, http.StatusUnauthorized), + "%s must answer 401 to a client with no session, answered: %v", endpoint.nsid, err) + } + }) + + t.Run("a batch read answers positionally, found and not-found alike", func(t *testing.T) { + // The endpoint's contract is that result i is the answer about uris[i], + // and that a valid URI nobody has indexed is a notFoundPost member inside + // a 200 — NOT an error, and NOT a shorter array. A client hydrating a feed + // skeleton zips these against its own list, so a compacted response would + // silently misattribute every post after the gap. + missing := postURI(community.DID, testkit.TID()) + + views, err := p.Posts(ctx, secondURI, missing, firstURI) + require.NoError(t, err, "an unresolvable URI in the batch must not fail the whole request") + require.Len(t, views, 3, "the answer must have one member per requested URI") + + require.False(t, views[0].NotFound) + require.Equal(t, secondURI, views[0].URI, "results must come back in request order") + require.True(t, views[1].NotFound, "a valid but unindexed URI must be a notFoundPost member") + require.Equal(t, missing, views[1].URI, "a notFoundPost must echo the URI it is about") + require.False(t, views[2].NotFound) + require.Equal(t, firstURI, views[2].URI) + }) + + t.Run("a handle-based URI is refused rather than resolved", func(t *testing.T) { + // Handles are mutable, so a handle-authority URI would break on rename or, + // worse, resolve to whoever holds the handle next. The service rejects the + // whole request instead of degrading it to a silent notFound, which is the + // difference between a client learning it has a bug and a client showing + // an empty post. + err := p.AppView.Query(ctx, "social.coves.community.post.get", + url.Values{"uris": {"at://" + community.Handle + "/" + postCollection + "/" + first}}, nil) + require.Truef(t, testkit.IsStatus(err, http.StatusBadRequest), + "a handle-authority URI must be a 400, answered: %v", err) + }) + + t.Run("the batch is bounded", func(t *testing.T) { + // An unbounded batch endpoint is an amplification lever: one request, N + // joins. The lexicon's limit is 25 and the handler enforces it before the + // service is called. + uris := make([]string, 0, 26) + for i := 0; i < 26; i++ { + uris = append(uris, postURI(community.DID, testkit.TID())) + } + _, err := p.Posts(ctx, uris...) + require.Truef(t, testkit.IsStatus(err, http.StatusBadRequest), + "asking for 26 URIs must be a 400, answered: %v", err) + }) + + t.Run("the post is served in its author's feed, by DID and by handle", func(t *testing.T) { + // social.coves.actor.getPosts is the other read surface a post reaches a + // client through, and it is a different query against different joins — + // so an indexed post can be visible on post.get and missing here. + // + // Both identifier forms, because they take different paths through the + // handler: a DID goes straight to the query, while a handle is resolved + // first (get_posts.go resolveActor → users.ResolveHandleToDID). The + // handle case doubles as proof that resolution is served from the + // AppView's own index — the stack is egress-blocked, so a lookup that + // escaped to DNS could not have answered at all. + for _, actor := range []string{author.DID, author.Handle} { + var feed struct { + Feed []struct { + Post postView `json:"post"` + } `json:"feed"` + } + require.NoErrorf(t, p.AppView.Query(ctx, "social.coves.actor.getPosts", + url.Values{"actor": {actor}, "limit": {"25"}}, &feed), + "social.coves.actor.getPosts rejected identifier %q", actor) + + seen := make([]string, 0, len(feed.Feed)) + var found bool + for _, item := range feed.Feed { + seen = append(seen, item.Post.URI) + if item.Post.URI == firstURI { + found = true + require.Equal(t, title, item.Post.Record["title"]) + require.Equal(t, community.DID, item.Post.Community.DID) + } + } + require.Truef(t, found, "the post %s was not in the %d posts returned for author %q: %s", + firstURI, len(feed.Feed), actor, strings.Join(seen, ", ")) + } + }) + + t.Run("an unknown author's feed is empty rather than an error", func(t *testing.T) { + // A DID is passed straight through with no existence check + // (get_posts.go resolveActor), so an unknown one answers 200 with an + // empty feed — indistinguishable, by design, from a real account that + // has not posted. Worth pinning because the alternative reading is + // natural and wrong: a client cannot use this endpoint to ask whether an + // actor exists. + // + // The DID is a literal rather than a generated one: the endpoint + // validates did:plc identifiers as base32 (a-z and 2-7), which UniqueID + // does not promise, and a malformed one would take the 400 path instead + // of the lookup path under test. Nothing indexes it, on a fresh stack or + // a kept one, so it needs no run-scoping. + // + // Spelled at the full 24 characters of a real did:plc even though the + // validator only checks the character set: a shorter one passes today and + // would start failing the moment that omission is corrected, quietly + // converting this into a validation test. + var empty struct { + Feed []struct{} `json:"feed"` + } + require.NoError(t, p.AppView.Query(ctx, "social.coves.actor.getPosts", + url.Values{"actor": {"did:plc:aaaaaaaaneverindexedactr"}}, &empty), + "a well-formed DID nobody has indexed is an empty feed, not an error") + assert.Empty(t, empty.Feed) + + // NOT asserted here: the handler's 404 ActorNotFound branch, which is the + // answer to an unresolvable HANDLE. It cannot be reached honestly in this + // stack. A handle missing from the AppView's index falls through to + // external DNS/HTTPS resolution, and the hermetic network is egress- + // blocked by design (§3.7), so the lookup fails with "server misbehaving" + // and the handler renders that as a 400 resolution failure — the correct + // answer to a broken resolver, and not the case anyone means to test. + // Distinguishing a nonexistent handle from a broken resolver needs a + // resolver that can answer, which is the second-PDS-and-relay topology of + // Phase 5. The reachable half of the handle path — a handle the index DOES + // know — is asserted in the feed subtest above. + }) +} diff --git a/tests/integration/helpers.go b/tests/integration/helpers.go index 1a955c9..126fd6d 100644 --- a/tests/integration/helpers.go +++ b/tests/integration/helpers.go @@ -81,49 +81,6 @@ func contains(s, substr string) bool { return strings.Contains(s, substr) } -// authenticateWithPDS authenticates with PDS to get access token and DID -// Used for setting up test environments that need PDS credentials -func authenticateWithPDS(pdsURL, handle, password string) (string, string, error) { - // Call com.atproto.server.createSession - sessionReq := map[string]string{ - "identifier": handle, - "password": password, - } - - reqBody, marshalErr := json.Marshal(sessionReq) - if marshalErr != nil { - return "", "", fmt.Errorf("failed to marshal session request: %w", marshalErr) - } - resp, err := http.Post( - pdsURL+"/xrpc/com.atproto.server.createSession", - "application/json", - bytes.NewBuffer(reqBody), - ) - if err != nil { - return "", "", fmt.Errorf("failed to create session: %w", err) - } - defer func() { _ = resp.Body.Close() }() - - if resp.StatusCode != http.StatusOK { - body, readErr := io.ReadAll(resp.Body) - if readErr != nil { - return "", "", fmt.Errorf("PDS auth failed (status %d, failed to read body: %w)", resp.StatusCode, readErr) - } - return "", "", fmt.Errorf("PDS auth failed (status %d): %s", resp.StatusCode, string(body)) - } - - var sessionResp struct { - AccessJwt string `json:"accessJwt"` - DID string `json:"did"` - } - - if err := json.NewDecoder(resp.Body).Decode(&sessionResp); err != nil { - return "", "", fmt.Errorf("failed to decode session response: %w", err) - } - - return sessionResp.AccessJwt, sessionResp.DID, nil -} - // jetstreamReadBudget is how long the hand-rolled subscribeToJetstream* helpers // in this package wait, in total, for the event they are looking for. // diff --git a/tests/integration/post_delete_test.go b/tests/integration/post_delete_test.go deleted file mode 100644 index ea3c17b..0000000 --- a/tests/integration/post_delete_test.go +++ /dev/null @@ -1,841 +0,0 @@ -//go:build integration - -package integration - -// SERIAL BY DESIGN — do not add t.Parallel() to this file. -// -// Its tests drive the Jetstream firehose through the hand-rolled -// subscribeToJetstream* helpers below rather than testkit's cursor-gated -// subscriber. Those helpers subscribe to one shared stream and match on the -// first event of a collection, so a concurrent test writing the same -// collection is delivered to them too and either steals the match or trips -// their timeout. Per-test database clones do not isolate a shared websocket. -// -// docs/TEST_ARCHITECTURE.md §3.3 ("Parallelism is earned, not assumed"). - -import ( - "Coves/internal/api/middleware" - "Coves/internal/atproto/identity" - "Coves/internal/atproto/jetstream" - "Coves/internal/core/communities" - "Coves/internal/core/posts" - "Coves/internal/core/users" - "Coves/internal/db/postgres" - "Coves/tests/testkit" - "context" - "errors" - "fmt" - "net" - "net/http" - "os" - "strings" - "testing" - "time" - - "github.com/gorilla/websocket" - - oauthlib "github.com/bluesky-social/indigo/atproto/auth/oauth" - "github.com/bluesky-social/indigo/atproto/syntax" -) - -// TestPostDeletion_JetstreamConsumer tests that the Jetstream consumer -// correctly handles post deletion events by soft-deleting posts in the AppView database. -func TestPostDeletion_JetstreamConsumer(t *testing.T) { - db := testkit.DB(t) - - ctx := context.Background() - - // Setup repositories - userRepo := postgres.NewUserRepository(db) - communityRepo := postgres.NewCommunityRepository(db) - postRepo := postgres.NewPostRepository(db) - - // Setup user service for post consumer - identityConfig := identity.DefaultConfig() - identityResolver := identity.NewResolver(db, identityConfig) - userService := users.NewUserService(userRepo, identityResolver, "http://localhost:3001", nil, "") - - // Create test user (author) - author := createTestUser(t, db, "delauthor.test", "did:plc:delauthor123") - - // Create test community - community := &communities.Community{ - DID: "did:plc:deltest123", - Handle: "c-deltest.test.coves.social", - Name: "deltest", - DisplayName: "Delete Test Community", - OwnerDID: "did:plc:deltest123", - CreatedByDID: author.DID, - HostedByDID: "did:web:coves.test", - Visibility: "public", - ModerationType: "moderator", - RecordURI: "at://did:plc:deltest123/social.coves.community.profile/self", - RecordCID: "fakecid123", - PDSAccessToken: "fake_token_for_testing", - PDSRefreshToken: "fake_refresh_token", - } - _, err := communityRepo.Create(ctx, community) - if err != nil { - t.Fatalf("Failed to create test community: %v", err) - } - - consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - - t.Run("Create then delete post via Jetstream", func(t *testing.T) { - rkey := generateTID() - title := "Post to be deleted" - content := "This post will be deleted" - - // Step 1: Create the post via Jetstream event - createEvent := jetstream.JetstreamEvent{ - Did: community.DID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "create", - Collection: "social.coves.community.post", - RKey: rkey, - CID: "bafy2bzacedeltest1", - Record: map[string]interface{}{ - "$type": "social.coves.community.post", - "community": community.DID, - "author": author.DID, - "title": title, - "content": content, - "createdAt": time.Now().Format(time.RFC3339), - }, - }, - } - - err := consumer.HandleEvent(ctx, &createEvent) - if err != nil { - t.Fatalf("Failed to create post: %v", err) - } - - // Verify post was created - postURI := fmt.Sprintf("at://%s/social.coves.community.post/%s", community.DID, rkey) - createdPost, err := postRepo.GetByURI(ctx, postURI) - if err != nil { - t.Fatalf("Post not indexed after create: %v", err) - } - if createdPost.DeletedAt != nil { - t.Fatal("Post should not be deleted initially") - } - - t.Logf("✓ Post created: %s", postURI) - - // Step 2: Delete the post via Jetstream event - deleteEvent := jetstream.JetstreamEvent{ - Did: community.DID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "delete", - Collection: "social.coves.community.post", - RKey: rkey, - }, - } - - err = consumer.HandleEvent(ctx, &deleteEvent) - if err != nil { - t.Fatalf("Failed to delete post: %v", err) - } - - // Step 3: Verify post was soft-deleted - deletedPost, err := postRepo.GetByURI(ctx, postURI) - if err != nil { - t.Fatalf("Post should still exist after soft delete: %v", err) - } - if deletedPost.DeletedAt == nil { - t.Fatal("Post should have deleted_at set after delete") - } - - t.Logf("✓ Post soft-deleted: deleted_at=%v", deletedPost.DeletedAt) - t.Log("✅ Delete flow complete: Create → Delete → Verify soft-deleted") - }) - - t.Run("Delete is idempotent", func(t *testing.T) { - rkey := generateTID() - - // Create a post first - createEvent := jetstream.JetstreamEvent{ - Did: community.DID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "create", - Collection: "social.coves.community.post", - RKey: rkey, - CID: "bafy2bzaceidempotentdel", - Record: map[string]interface{}{ - "$type": "social.coves.community.post", - "community": community.DID, - "author": author.DID, - "title": "Idempotent delete test", - "content": "Testing idempotent deletion", - "createdAt": time.Now().Format(time.RFC3339), - }, - }, - } - err := consumer.HandleEvent(ctx, &createEvent) - if err != nil { - t.Fatalf("Failed to create post: %v", err) - } - - // Delete once - deleteEvent := jetstream.JetstreamEvent{ - Did: community.DID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "delete", - Collection: "social.coves.community.post", - RKey: rkey, - }, - } - err = consumer.HandleEvent(ctx, &deleteEvent) - if err != nil { - t.Fatalf("First delete failed: %v", err) - } - - // Delete again (should be idempotent) - err = consumer.HandleEvent(ctx, &deleteEvent) - if err != nil { - t.Fatalf("Second delete should be idempotent, got error: %v", err) - } - - t.Log("✓ Delete is idempotent - second delete did not fail") - }) - - t.Run("Delete non-existent post is idempotent", func(t *testing.T) { - // Try to delete a post that was never created - deleteEvent := jetstream.JetstreamEvent{ - Did: community.DID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "delete", - Collection: "social.coves.community.post", - RKey: "nonexistent123", - }, - } - - err := consumer.HandleEvent(ctx, &deleteEvent) - if err != nil { - t.Fatalf("Delete of non-existent post should be idempotent, got error: %v", err) - } - - t.Log("✓ Delete of non-existent post is idempotent") - }) -} - -// TestPostDeletion_Authorization tests that only the post author can delete their posts -func TestPostDeletion_Authorization(t *testing.T) { - db := testkit.DB(t) - - ctx := context.Background() - pdsURL := getTestPDSURL() - - // Setup repositories - communityRepo := postgres.NewCommunityRepository(db) - postRepo := postgres.NewPostRepository(db) - - // Create a mock community service for testing - communityService := communities.NewCommunityServiceWithPDSFactory( - communityRepo, - pdsURL, - "did:web:test", - "test.coves.social", - nil, // No provisioner needed for this test - nil, // No PDS factory - nil, // No blob service - ) - - postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL) - - // Create test user (attacker trying to delete another user's post) - attackerID := uniqueTestID() - attackerHandle := fmt.Sprintf("atk%s.local.coves.dev", attackerID) - attackerEmail := fmt.Sprintf("attacker-%s@test.local", attackerID) - attackerToken, attackerDID, err := createPDSAccount(pdsURL, attackerHandle, attackerEmail, "password123") - if err != nil { - t.Skipf("PDS not available: %v", err) - } - - // Setup OAuth session for attacker - parsedDID, err := syntax.ParseDID(attackerDID) - if err != nil { - t.Fatalf("Failed to parse attacker DID: %v", err) - } - attackerSession := &oauthlib.ClientSessionData{ - AccountDID: parsedDID, - AccessToken: attackerToken, - HostURL: pdsURL, - } - - // Create post URI belonging to a DIFFERENT user (the owner) - ownerDID := "did:plc:owner123" - postURI := fmt.Sprintf("at://%s/social.coves.community.post/test123", ownerDID) - - t.Run("Non-author cannot delete post - URI contains wrong DID", func(t *testing.T) { - // The post URI contains ownerDID in the community position - // This should fail because attacker is not the community owner - // and wouldn't have credentials to delete from that repo - - deleteReq := posts.DeletePostRequest{ - URI: postURI, - } - - err := postService.DeletePost(ctx, attackerSession, deleteReq) - - // We expect an error - either NotAuthorized or CommunityNotFound - // since the community doesn't exist in our test DB - if err == nil { - t.Fatal("Expected error when non-author tries to delete post, got nil") - } - - t.Logf("✓ Non-author blocked from deleting post: %v", err) - }) - - t.Run("Invalid URI format returns error", func(t *testing.T) { - deleteReq := posts.DeletePostRequest{ - URI: "invalid-uri-format", - } - - err := postService.DeletePost(ctx, attackerSession, deleteReq) - - if err == nil { - t.Fatal("Expected error for invalid URI, got nil") - } - - // Should be a validation error - if !posts.IsValidationError(err) { - t.Logf("Got error (expected validation): %v", err) - } - - t.Logf("✓ Invalid URI rejected: %v", err) - }) - - t.Run("Empty URI returns error", func(t *testing.T) { - deleteReq := posts.DeletePostRequest{ - URI: "", - } - - err := postService.DeletePost(ctx, attackerSession, deleteReq) - - if err == nil { - t.Fatal("Expected error for empty URI, got nil") - } - - t.Logf("✓ Empty URI rejected: %v", err) - }) - - t.Run("Nil session returns error", func(t *testing.T) { - deleteReq := posts.DeletePostRequest{ - URI: postURI, - } - - err := postService.DeletePost(ctx, nil, deleteReq) - - if err == nil { - t.Fatal("Expected error for nil session, got nil") - } - - t.Logf("✓ Nil session rejected: %v", err) - }) -} - -// TestPostDeletion_ServiceAuthorization tests the author verification logic in the service layer -// This test requires a live PDS to fully test the authorization flow -func TestPostDeletion_ServiceAuthorization_LivePDS(t *testing.T) { - db := testkit.DB(t) - - ctx := context.Background() - pdsURL := getTestPDSURL() - - // Check if PDS is available - healthResp, err := http.Get(pdsURL + "/xrpc/_health") - if err != nil { - t.Skipf("PDS not available: %v", err) - } - _ = healthResp.Body.Close() - - // Setup repositories - communityRepo := postgres.NewCommunityRepository(db) - postRepo := postgres.NewPostRepository(db) - - // Get instance credentials to determine correct domain - instanceHandle := os.Getenv("PDS_INSTANCE_HANDLE") - instancePassword := os.Getenv("PDS_INSTANCE_PASSWORD") - if instanceHandle == "" { - instanceHandle = "testuser123.local.coves.dev" - } - if instancePassword == "" { - instancePassword = "test-password-123" - } - - _, instanceDID, err := authenticateWithPDS(pdsURL, instanceHandle, instancePassword) - if err != nil { - t.Skipf("Failed to authenticate with PDS: %v", err) - } - - var instanceDomain string - if strings.HasPrefix(instanceDID, "did:web:") { - instanceDomain = strings.TrimPrefix(instanceDID, "did:web:") - } else { - instanceDomain = "local.coves.dev" - } - - // Create provisioner for community creation - provisioner := communities.NewPDSAccountProvisioner(instanceDomain, pdsURL) - - communityService := communities.NewCommunityServiceWithPDSFactory( - communityRepo, - pdsURL, - instanceDID, - instanceDomain, - provisioner, - nil, - nil, - ) - - postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL) - - // Create two test users - ownerID := uniqueTestID() - ownerHandle := fmt.Sprintf("pown%s.local.coves.dev", ownerID) - ownerEmail := fmt.Sprintf("postowner-%s@test.local", ownerID) - _, ownerDID, err := createPDSAccount(pdsURL, ownerHandle, ownerEmail, "password123") - if err != nil { - t.Skipf("Failed to create owner account: %v", err) - } - owner := createTestUser(t, db, ownerHandle, ownerDID) - - attackerID := uniqueTestID() - attackerHandle := fmt.Sprintf("patk%s.local.coves.dev", attackerID) - attackerEmail := fmt.Sprintf("postattacker-%s@test.local", attackerID) - attackerToken, attackerDID, err := createPDSAccount(pdsURL, attackerHandle, attackerEmail, "password123") - if err != nil { - t.Skipf("Failed to create attacker account: %v", err) - } - _ = createTestUser(t, db, attackerHandle, attackerDID) - - // Setup attacker session - parsedAttackerDID, _ := syntax.ParseDID(attackerDID) - attackerSession := &oauthlib.ClientSessionData{ - AccountDID: parsedAttackerDID, - AccessToken: attackerToken, - HostURL: pdsURL, - } - - // Create a test community - communityName := fmt.Sprintf("del%s", uniqueTestID()) - community, err := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{ - Name: communityName, - DisplayName: "Delete Auth Test Community", - Description: "Testing post deletion authorization", - CreatedByDID: owner.DID, - Visibility: "public", - }) - if err != nil { - t.Fatalf("Failed to create community: %v", err) - } - - t.Logf("✓ Community created: %s (%s)", community.Name, community.DID) - - // Create a post as the owner - title := "Owner's Post" - content := "This post belongs to the owner" - createResp, err := postService.CreatePost( - middleware.SetTestUserDID(ctx, owner.DID), - posts.CreatePostRequest{ - Community: community.DID, - Title: &title, - Content: &content, - AuthorDID: owner.DID, - }, - ) - if err != nil { - t.Fatalf("Failed to create post: %v", err) - } - - t.Logf("✓ Post created by owner: %s", createResp.URI) - - t.Run("Attacker cannot delete owner's post", func(t *testing.T) { - deleteReq := posts.DeletePostRequest{ - URI: createResp.URI, - } - - err := postService.DeletePost(ctx, attackerSession, deleteReq) - - if err == nil { - t.Fatal("Expected ErrNotAuthorized when attacker tries to delete owner's post") - } - - if !errors.Is(err, posts.ErrNotAuthorized) { - t.Errorf("Expected ErrNotAuthorized, got: %v", err) - } - - t.Logf("✅ Authorization check passed: attacker blocked with %v", err) - }) -} - -// TestPostE2E_DeleteWithJetstream tests post deletion with real PDS and Jetstream -// This is a TRUE E2E test that follows the complete flow: -// 1. Create real community on PDS -// 2. Create real post on PDS -// 3. Subscribe to Jetstream -// 4. Delete post via service (which deletes from PDS) -// 5. Receive delete event from Jetstream -// 6. Verify post is soft-deleted in AppView DB -func TestPostE2E_DeleteWithJetstream(t *testing.T) { - db := testkit.DB(t) - - pdsURL := getTestPDSURL() - healthResp, err := http.Get(pdsURL + "/xrpc/_health") - if err != nil { - t.Skipf("PDS not running at %s: %v", pdsURL, err) - } - _ = healthResp.Body.Close() - - // Check Jetstream is available - pdsHostname := strings.TrimPrefix(pdsURL, "http://") - pdsHostname = strings.TrimPrefix(pdsHostname, "https://") - pdsHostname = strings.Split(pdsHostname, ":")[0] - jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.community.post", pdsHostname) - - testConn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil) - if err != nil { - t.Skipf("Jetstream not running at %s: %v", jetstreamURL, err) - } - _ = testConn.Close() - - ctx := context.Background() - - // Setup repositories - userRepo := postgres.NewUserRepository(db) - communityRepo := postgres.NewCommunityRepository(db) - postRepo := postgres.NewPostRepository(db) - - // Setup identity resolver for user service - identityConfig := identity.DefaultConfig() - plcURL := os.Getenv("PLC_DIRECTORY_URL") - if plcURL == "" { - plcURL = "http://localhost:3002" - } - identityConfig.PLCURL = plcURL - identityResolver := identity.NewResolver(db, identityConfig) - userService := users.NewUserService(userRepo, identityResolver, pdsURL, nil, "") - - // Setup community service with provisioner for real PDS - var instanceDomain string - instanceHandle := os.Getenv("PDS_INSTANCE_HANDLE") - instancePassword := os.Getenv("PDS_INSTANCE_PASSWORD") - if instanceHandle == "" { - instanceHandle = "testuser123.local.coves.dev" - } - if instancePassword == "" { - instancePassword = "test-password-123" - } - - _, instanceDID, err := authenticateWithPDS(pdsURL, instanceHandle, instancePassword) - if err != nil { - t.Skipf("Failed to authenticate with PDS: %v", err) - } - - if strings.HasPrefix(instanceDID, "did:web:") { - instanceDomain = strings.TrimPrefix(instanceDID, "did:web:") - } else { - instanceDomain = "coves.social" - } - - provisioner := communities.NewPDSAccountProvisioner(instanceDomain, pdsURL) - communityService := communities.NewCommunityServiceWithPDSFactory( - communityRepo, - pdsURL, - instanceDID, - instanceDomain, - provisioner, - nil, - nil, - ) - - postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL) - - // Create test user - testID := uniqueTestID() - testUserHandle := fmt.Sprintf("pd%s.local.coves.dev", testID) - testUserEmail := fmt.Sprintf("postdel%s@test.local", testID) - testUserPassword := "test-password-123" - - _, userDID, err := createPDSAccount(pdsURL, testUserHandle, testUserEmail, testUserPassword) - if err != nil { - t.Fatalf("Failed to create test user on PDS: %v", err) - } - testUser := createTestUser(t, db, testUserHandle, userDID) - - // Create test community on real PDS - communityName := fmt.Sprintf("pd%s", testID) - t.Logf("\n📝 Creating community on PDS: %s", communityName) - - community, err := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{ - Name: communityName, - DisplayName: "Post Delete E2E Test Community", - Description: "Testing post deletion E2E flow", - CreatedByDID: testUser.DID, - Visibility: "public", - }) - if err != nil { - t.Fatalf("Failed to create community: %v", err) - } - - t.Logf("✅ Community created: %s (%s)", community.Name, community.DID) - - // Setup Jetstream consumer - postConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - - t.Run("delete post with real Jetstream indexing", func(t *testing.T) { - // Create post via service (writes to real PDS) - createEventChan := make(chan *jetstream.JetstreamEvent, 10) - createDone := make(chan bool) - - go func() { - subscribeErr := subscribeToJetstreamForPostCreate(ctx, jetstreamURL, community.DID, postConsumer, createEventChan, createDone) - if subscribeErr != nil { - t.Logf("Create subscription error: %v", subscribeErr) - } - }() - - time.Sleep(500 * time.Millisecond) - - title := "Post to delete via E2E" - content := "This post will be deleted and we'll verify via Jetstream" - t.Logf("\n📝 Creating post on PDS...") - - createResp, err := postService.CreatePost( - middleware.SetTestUserDID(ctx, testUser.DID), - posts.CreatePostRequest{ - Community: community.DID, - Title: &title, - Content: &content, - AuthorDID: testUser.DID, - }, - ) - if err != nil { - t.Fatalf("Failed to create post: %v", err) - } - - t.Logf("✅ Post created: %s", createResp.URI) - - // Wait for create event from Jetstream - select { - case <-createEventChan: - t.Logf("✅ Create event received from Jetstream") - case <-time.After(30 * time.Second): - t.Fatalf("Timeout waiting for create event") - } - close(createDone) - - // Verify post exists in AppView - createdPost, err := postRepo.GetByURI(ctx, createResp.URI) - if err != nil { - t.Fatalf("Post should exist after create: %v", err) - } - if createdPost.DeletedAt != nil { - t.Fatal("Post should not be deleted initially") - } - - // Now delete the post - t.Logf("\n🗑️ Deleting post via service...") - - deleteEventChan := make(chan *jetstream.JetstreamEvent, 10) - deleteDone := make(chan bool) - - go func() { - subscribeErr := subscribeToJetstreamForPostDelete(ctx, jetstreamURL, community.DID, postConsumer, deleteEventChan, deleteDone) - if subscribeErr != nil { - t.Logf("Delete subscription error: %v", subscribeErr) - } - }() - - time.Sleep(500 * time.Millisecond) - - // Create OAuth session for the post author - parsedDID, _ := syntax.ParseDID(testUser.DID) - // Get fresh token for the user - freshToken, _, err := authenticateWithPDS(pdsURL, testUserHandle, testUserPassword) - if err != nil { - t.Fatalf("Failed to get fresh token: %v", err) - } - session := &oauthlib.ClientSessionData{ - AccountDID: parsedDID, - AccessToken: freshToken, - HostURL: pdsURL, - } - - err = postService.DeletePost(ctx, session, posts.DeletePostRequest{URI: createResp.URI}) - if err != nil { - t.Fatalf("Failed to delete post: %v", err) - } - - t.Logf("✅ Post delete request sent to PDS") - - // Wait for delete event from Jetstream - t.Logf("\n⏳ Waiting for delete event from Jetstream...") - - select { - case event := <-deleteEventChan: - t.Logf("✅ Received delete event from Jetstream!") - t.Logf(" Operation: %s", event.Commit.Operation) - - if event.Commit.Operation != "delete" { - t.Errorf("Expected operation 'delete', got '%s'", event.Commit.Operation) - } - - // Verify post is soft-deleted in AppView - deletedPost, err := postRepo.GetByURI(ctx, createResp.URI) - if err != nil { - t.Fatalf("Failed to get deleted post: %v", err) - } - - if deletedPost.DeletedAt == nil { - t.Errorf("Expected post to be soft-deleted (deleted_at should be set)") - } else { - t.Logf("✅ Post soft-deleted in AppView at: %v", *deletedPost.DeletedAt) - } - - close(deleteDone) - - case <-time.After(30 * time.Second): - t.Fatalf("Timeout: No delete event received within 30 seconds") - } - - t.Logf("\n✅ TRUE E2E POST DELETE FLOW COMPLETE:") - t.Logf(" Client → Service → PDS DeleteRecord → Jetstream → Consumer → AppView ✓") - }) -} - -// subscribeToJetstreamForPostCreate subscribes for post create events -func subscribeToJetstreamForPostCreate( - ctx context.Context, - jetstreamURL string, - targetDID string, - consumer *jetstream.PostEventConsumer, - eventChan chan<- *jetstream.JetstreamEvent, - done <-chan bool, -) error { - conn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil) - if err != nil { - return fmt.Errorf("failed to connect to Jetstream: %w", err) - } - defer func() { _ = conn.Close() }() - - // ONE deadline for the whole subscription, not one per read: the - // budget is what the caller is willing to wait in total, and a - // per-read deadline would let a busy stream extend it indefinitely. - readDeadline := time.Now().Add(jetstreamReadBudget) - - for { - select { - case <-done: - return nil - case <-ctx.Done(): - return ctx.Err() - default: - if err := conn.SetReadDeadline(readDeadline); err != nil { - return fmt.Errorf("failed to set read deadline: %w", err) - } - - var event jetstream.JetstreamEvent - err := conn.ReadJSON(&event) - if err != nil { - if websocket.IsCloseError(err, websocket.CloseNormalClosure) { - return fmt.Errorf("Jetstream closed the subscription before the event arrived") - } - if netErr, ok := err.(net.Error); ok && netErr.Timeout() { - // The deadline is the whole budget, so its expiry is the answer: - // no matching event arrived. Reading on would be reading a - // connection gorilla has already marked failed. - return fmt.Errorf("no matching event within %s", jetstreamReadBudget) - } - return fmt.Errorf("failed to read Jetstream message: %w", err) - } - - if event.Did == targetDID && event.Kind == "commit" && - event.Commit != nil && event.Commit.Collection == "social.coves.community.post" && - event.Commit.Operation == "create" { - - if err := consumer.HandleEvent(ctx, &event); err != nil { - return fmt.Errorf("failed to process event: %w", err) - } - - select { - case eventChan <- &event: - return nil - case <-time.After(1 * time.Second): - return fmt.Errorf("timeout sending event to channel") - } - } - } - } -} - -// subscribeToJetstreamForPostDelete subscribes for post delete events -func subscribeToJetstreamForPostDelete( - ctx context.Context, - jetstreamURL string, - targetDID string, - consumer *jetstream.PostEventConsumer, - eventChan chan<- *jetstream.JetstreamEvent, - done <-chan bool, -) error { - conn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil) - if err != nil { - return fmt.Errorf("failed to connect to Jetstream: %w", err) - } - defer func() { _ = conn.Close() }() - - // ONE deadline for the whole subscription, not one per read: the - // budget is what the caller is willing to wait in total, and a - // per-read deadline would let a busy stream extend it indefinitely. - readDeadline := time.Now().Add(jetstreamReadBudget) - - for { - select { - case <-done: - return nil - case <-ctx.Done(): - return ctx.Err() - default: - if err := conn.SetReadDeadline(readDeadline); err != nil { - return fmt.Errorf("failed to set read deadline: %w", err) - } - - var event jetstream.JetstreamEvent - err := conn.ReadJSON(&event) - if err != nil { - if websocket.IsCloseError(err, websocket.CloseNormalClosure) { - return fmt.Errorf("Jetstream closed the subscription before the event arrived") - } - if netErr, ok := err.(net.Error); ok && netErr.Timeout() { - // The deadline is the whole budget, so its expiry is the answer: - // no matching event arrived. Reading on would be reading a - // connection gorilla has already marked failed. - return fmt.Errorf("no matching event within %s", jetstreamReadBudget) - } - return fmt.Errorf("failed to read Jetstream message: %w", err) - } - - if event.Did == targetDID && event.Kind == "commit" && - event.Commit != nil && event.Commit.Collection == "social.coves.community.post" && - event.Commit.Operation == "delete" { - - if err := consumer.HandleEvent(ctx, &event); err != nil { - return fmt.Errorf("failed to process event: %w", err) - } - - select { - case eventChan <- &event: - return nil - case <-time.After(1 * time.Second): - return fmt.Errorf("timeout sending event to channel") - } - } - } - } -} diff --git a/tests/integration/post_e2e_test.go b/tests/integration/post_e2e_test.go deleted file mode 100644 index 22cf0ba..0000000 --- a/tests/integration/post_e2e_test.go +++ /dev/null @@ -1,662 +0,0 @@ -//go:build integration - -package integration - -// SERIAL BY DESIGN — do not add t.Parallel() to this file. -// -// Its tests drive the Jetstream firehose through the hand-rolled -// subscribeToJetstream* helpers below rather than testkit's cursor-gated -// subscriber. Those helpers subscribe to one shared stream and match on the -// first event of a collection, so a concurrent test writing the same -// collection is delivered to them too and either steals the match or trips -// their timeout. Per-test database clones do not isolate a shared websocket. -// -// docs/TEST_ARCHITECTURE.md §3.3 ("Parallelism is earned, not assumed"). - -import ( - "Coves/internal/api/handlers/post" - "Coves/internal/atproto/identity" - "Coves/internal/atproto/jetstream" - "Coves/internal/core/communities" - "Coves/internal/core/posts" - "Coves/internal/core/users" - "Coves/internal/db/postgres" - "Coves/tests/testkit" - "bytes" - "context" - "encoding/json" - "fmt" - "net" - "net/http" - "net/http/httptest" - "os" - "strings" - "testing" - "time" - - "github.com/gorilla/websocket" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -// TestPostCreation_E2E_WithJetstream tests the full post creation flow: -// XRPC endpoint → AppView Service → PDS write → Jetstream consumer → DB indexing -// -// This is a TRUE E2E test that simulates what happens in production: -// 1. Client calls POST /xrpc/social.coves.community.post.create with auth token -// 2. Handler validates and calls PostService.CreatePost() -// 3. Service writes post to community's PDS repository -// 4. PDS broadcasts event to firehose/Jetstream -// 5. Jetstream consumer receives event and indexes post in AppView DB -// 6. Post is now queryable from AppView -// -// NOTE: This test simulates the Jetstream event (step 4-5) since we don't have -// a live PDS/Jetstream in test environment. For true live testing, use TestPostCreation_E2E_LivePDS. -func TestPostCreation_E2E_WithJetstream(t *testing.T) { - db := testkit.DB(t) - - // Setup repositories - userRepo := postgres.NewUserRepository(db) - communityRepo := postgres.NewCommunityRepository(db) - postRepo := postgres.NewPostRepository(db) - - // Setup user service for post consumer - identityConfig := identity.DefaultConfig() - identityResolver := identity.NewResolver(db, identityConfig) - userService := users.NewUserService(userRepo, identityResolver, "http://localhost:3001", nil, "") - - // Create test user (author) - author := createTestUser(t, db, "alice.test", "did:plc:alice123") - - // Create test community with fake PDS credentials - // In real E2E, this would be a real community provisioned on PDS - community := &communities.Community{ - DID: "did:plc:gaming123", - Handle: "c-gaming.test.coves.social", - Name: "gaming", - DisplayName: "Gaming Community", - OwnerDID: "did:plc:gaming123", - CreatedByDID: author.DID, - HostedByDID: "did:web:coves.test", - Visibility: "public", - ModerationType: "moderator", - RecordURI: "at://did:plc:gaming123/social.coves.community.profile/self", - RecordCID: "fakecid123", - PDSAccessToken: "fake_token_for_testing", - PDSRefreshToken: "fake_refresh_token", - } - _, err := communityRepo.Create(context.Background(), community) - if err != nil { - t.Fatalf("Failed to create test community: %v", err) - } - - t.Run("Full E2E flow - XRPC to DB via Jetstream", func(t *testing.T) { - ctx := context.Background() - - // STEP 1: Simulate what the XRPC handler would receive - // In real flow, this comes from client with OAuth bearer token - title := "My First Post" - content := "This is a test post!" - postReq := posts.CreatePostRequest{ - Title: &title, - Content: &content, - // Community and AuthorDID set by handler from request context - } - - // STEP 2: Simulate Jetstream consumer receiving the post CREATE event - // In real production, this event comes from PDS via Jetstream WebSocket - // For this test, we simulate the event that would be broadcast after PDS write - - // Generate a realistic rkey (TID - timestamp identifier) - rkey := generateTID() - - // Build the post record as it would appear in Jetstream - jetstreamEvent := jetstream.JetstreamEvent{ - Did: community.DID, // Repo owner (community) - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "create", - Collection: "social.coves.community.post", - RKey: rkey, - CID: "bafy2bzaceabc123def456", // Fake CID - Record: map[string]interface{}{ - "$type": "social.coves.community.post", - "community": community.DID, - "author": author.DID, - "title": *postReq.Title, - "content": *postReq.Content, - "createdAt": time.Now().Format(time.RFC3339), - }, - }, - } - - // STEP 3: Process event through Jetstream consumer - consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - err := consumer.HandleEvent(ctx, &jetstreamEvent) - if err != nil { - t.Fatalf("Jetstream consumer failed to process event: %v", err) - } - - // STEP 4: Verify post was indexed in AppView database - expectedURI := fmt.Sprintf("at://%s/social.coves.community.post/%s", community.DID, rkey) - indexedPost, err := postRepo.GetByURI(ctx, expectedURI) - if err != nil { - t.Fatalf("Post not indexed in AppView: %v", err) - } - - // STEP 5: Verify all fields are correct - if indexedPost.URI != expectedURI { - t.Errorf("Expected URI %s, got %s", expectedURI, indexedPost.URI) - } - if indexedPost.AuthorDID != author.DID { - t.Errorf("Expected author %s, got %s", author.DID, indexedPost.AuthorDID) - } - if indexedPost.CommunityDID != community.DID { - t.Errorf("Expected community %s, got %s", community.DID, indexedPost.CommunityDID) - } - if indexedPost.Title == nil || *indexedPost.Title != title { - t.Errorf("Expected title '%s', got %v", title, indexedPost.Title) - } - if indexedPost.Content == nil || *indexedPost.Content != content { - t.Errorf("Expected content '%s', got %v", content, indexedPost.Content) - } - - // Verify stats initialized correctly - if indexedPost.UpvoteCount != 0 { - t.Errorf("Expected upvote_count 0, got %d", indexedPost.UpvoteCount) - } - if indexedPost.DownvoteCount != 0 { - t.Errorf("Expected downvote_count 0, got %d", indexedPost.DownvoteCount) - } - if indexedPost.Score != 0 { - t.Errorf("Expected score 0, got %d", indexedPost.Score) - } - - t.Logf("✓ E2E test passed! Post indexed with URI: %s", indexedPost.URI) - }) - - t.Run("Consumer validates repository ownership (security)", func(t *testing.T) { - ctx := context.Background() - - // SECURITY TEST: Try to create a post that claims to be from the community - // but actually comes from a user's repository - // This should be REJECTED by the consumer - - maliciousEvent := jetstream.JetstreamEvent{ - Did: author.DID, // Event from user's repo (NOT community repo) - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "create", - Collection: "social.coves.community.post", - RKey: generateTID(), - CID: "bafy2bzacefake", - Record: map[string]interface{}{ - "$type": "social.coves.community.post", - "community": community.DID, // Claims to be for this community - "author": author.DID, - "title": "Fake Post", - "content": "This is a malicious post attempt", - "createdAt": time.Now().Format(time.RFC3339), - }, - }, - } - - consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - err := consumer.HandleEvent(ctx, &maliciousEvent) - - // Should get security error - if err == nil { - t.Fatal("Expected security error for post from wrong repository, got nil") - } - - if !contains(err.Error(), "repository DID") || !contains(err.Error(), "doesn't match") { - t.Errorf("Expected repository mismatch error, got: %v", err) - } - - t.Logf("✓ Security validation passed: %v", err) - }) - - t.Run("Idempotent indexing - duplicate events", func(t *testing.T) { - ctx := context.Background() - - // Simulate the same Jetstream event arriving twice - // This can happen during Jetstream replays or network retries - rkey := generateTID() - event := jetstream.JetstreamEvent{ - Did: community.DID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "create", - Collection: "social.coves.community.post", - RKey: rkey, - CID: "bafy2bzaceidempotent", - Record: map[string]interface{}{ - "$type": "social.coves.community.post", - "community": community.DID, - "author": author.DID, - "title": "Duplicate Test", - "content": "Testing idempotency", - "createdAt": time.Now().Format(time.RFC3339), - }, - }, - } - - consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - - // First event - should succeed - err := consumer.HandleEvent(ctx, &event) - if err != nil { - t.Fatalf("First event failed: %v", err) - } - - // Second event (duplicate) - should be handled gracefully - err = consumer.HandleEvent(ctx, &event) - if err != nil { - t.Fatalf("Duplicate event should be handled gracefully, got error: %v", err) - } - - // Verify only one post in database - uri := fmt.Sprintf("at://%s/social.coves.community.post/%s", community.DID, rkey) - post, err := postRepo.GetByURI(ctx, uri) - if err != nil { - t.Fatalf("Post not found: %v", err) - } - - if post.URI != uri { - t.Error("Post URI mismatch - possible duplicate indexing") - } - - t.Logf("✓ Idempotency test passed") - }) - - t.Run("Handles orphaned posts (unknown community)", func(t *testing.T) { - ctx := context.Background() - - // Post references a community that doesn't exist in AppView yet - // This can happen if Jetstream delivers post event before community profile event - unknownCommunityDID := "did:plc:unknown999" - - event := jetstream.JetstreamEvent{ - Did: unknownCommunityDID, - Kind: "commit", - Commit: &jetstream.CommitEvent{ - Operation: "create", - Collection: "social.coves.community.post", - RKey: generateTID(), - CID: "bafy2bzaceorphaned", - Record: map[string]interface{}{ - "$type": "social.coves.community.post", - "community": unknownCommunityDID, - "author": author.DID, - "title": "Orphaned Post", - "content": "Community not indexed yet", - "createdAt": time.Now().Format(time.RFC3339), - }, - }, - } - - consumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - - // Should log warning but NOT fail (eventual consistency) - // Note: This will fail due to foreign key constraint in current schema - // In production, you might want to handle this differently (defer indexing, etc.) - err := consumer.HandleEvent(ctx, &event) - - // For now, we expect this to fail due to FK constraint - // In future, we might make FK constraint DEFERRABLE or handle orphaned posts differently - if err == nil { - t.Log("⚠️ Orphaned post was indexed (FK constraint not enforced)") - } else { - t.Logf("✓ Orphaned post rejected by FK constraint (expected): %v", err) - } - }) -} - -// TestPostCreation_E2E_LivePDS tests the COMPLETE end-to-end flow with a live PDS: -// 1. HTTP POST to /xrpc/social.coves.community.post.create (with auth) -// 2. Handler → Service → Write to community's PDS repository -// 3. PDS → Jetstream firehose event -// 4. Jetstream consumer → Index in AppView database -// 5. Verify post appears in database with correct data -// -// This is a TRUE E2E test that requires: -// - Live PDS running at PDS_URL (default: http://localhost:3001) -// - Live Jetstream running at the local dev Jetstream (JETSTREAM_FEEDS default: self=ws://localhost:6008) -// - Test database running -func TestPostCreation_E2E_LivePDS(t *testing.T) { - db := testkit.DB(t) - - // Check if PDS is running - pdsURL := os.Getenv("PDS_URL") - if pdsURL == "" { - pdsURL = "http://localhost:3001" - } - - healthResp, err := http.Get(pdsURL + "/xrpc/_health") - if err != nil { - t.Skipf("PDS not running at %s: %v", pdsURL, err) - } - _ = healthResp.Body.Close() - - // Get instance credentials for authentication - instanceHandle := os.Getenv("PDS_INSTANCE_HANDLE") - instancePassword := os.Getenv("PDS_INSTANCE_PASSWORD") - if instanceHandle == "" { - instanceHandle = "testuser123.local.coves.dev" - } - if instancePassword == "" { - instancePassword = "test-password-123" - } - - t.Logf("🔐 Authenticating with PDS as: %s", instanceHandle) - - // Authenticate to get instance DID (needed for provisioner domain) - _, instanceDID, err := authenticateWithPDS(pdsURL, instanceHandle, instancePassword) - if err != nil { - t.Skipf("Failed to authenticate with PDS (may not be configured): %v", err) - } - - t.Logf("✅ Authenticated - Instance DID: %s", instanceDID) - - // Extract instance domain from DID for community provisioning - var instanceDomain string - if strings.HasPrefix(instanceDID, "did:web:") { - instanceDomain = strings.TrimPrefix(instanceDID, "did:web:") - } else { - // Fallback for did:plc - instanceDomain = "coves.social" - } - - // Setup repositories and services - communityRepo := postgres.NewCommunityRepository(db) - postRepo := postgres.NewPostRepository(db) - - // Setup PDS account provisioner for community creation - provisioner := communities.NewPDSAccountProvisioner(instanceDomain, pdsURL) - - // Setup community service with real PDS provisioner - communityService := communities.NewCommunityServiceWithPDSFactory( - communityRepo, - pdsURL, - instanceDID, - instanceDomain, - provisioner, // Real provisioner for creating communities on PDS - nil, // No PDS factory needed - no subscribe/block in this test - nil, // No blob service for this test - ) - - postService := posts.NewPostService(postRepo, communityService, nil, nil, nil, nil, pdsURL) // nil aggregatorService, blobService, unfurlService, blueskyService for user-only tests - - // Setup OAuth auth middleware for E2E testing - e2eAuth := NewE2EOAuthMiddleware() - - // Setup HTTP handler - createHandler := post.NewCreateHandler(postService) - - ctx := context.Background() - - // Create test user (author) - author := createTestUser(t, db, "e2etestauthor.bsky.social", "did:plc:e2etestauthor123") - - // ==================================================================================== - // Part 1: Write-Forward to PDS - // ==================================================================================== - t.Run("1. Write-Forward to PDS", func(t *testing.T) { - // TRUE E2E: Actually provision a real community on PDS - // This tests the full flow: - // 1. Call com.atproto.server.createAccount on PDS - // 2. PDS generates DID, keys, tokens - // 3. Write community profile to PDS repository - // 4. Store credentials in AppView DB - // 5. Use those credentials to create a post - - // uniqueTestID() (Unix seconds + atomic counter) ensures a unique community name - // across reruns; the % modulo form collides ("handle already taken") on the - // provisioned PDS account, and the local label must stay ≤18 chars. - communityName := fmt.Sprintf("e2e%s", uniqueTestID()) - - t.Logf("\n📝 Provisioning test community on live PDS (name: %s)...", communityName) - community, err := communityService.CreateCommunity(ctx, communities.CreateCommunityRequest{ - Name: communityName, - DisplayName: "E2E Test Community", - Description: "Test community for E2E post creation testing", - CreatedByDID: author.DID, - Visibility: "public", - AllowExternalDiscovery: true, - }) - require.NoError(t, err, "Failed to provision community on PDS") - require.NotEmpty(t, community.DID, "Community should have DID from PDS") - require.NotEmpty(t, community.PDSAccessToken, "Community should have access token") - require.NotEmpty(t, community.PDSRefreshToken, "Community should have refresh token") - - t.Logf("✓ Community provisioned: DID=%s, Handle=%s", community.DID, community.Handle) - - // NOTE: Cleanup disabled to allow post-test inspection of indexed data - // Uncomment to enable cleanup after test - // defer func() { - // if err := communityRepo.Delete(ctx, community.DID); err != nil { - // t.Logf("Warning: Failed to cleanup test community: %v", err) - // } - // }() - - // Build HTTP request for post creation - title := "E2E Test Post" - content := "This post was created via full E2E test with live PDS!" - reqBody := map[string]interface{}{ - "community": community.DID, - "title": title, - "content": content, - } - reqJSON, err := json.Marshal(reqBody) - require.NoError(t, err) - - // Create HTTP request - req := httptest.NewRequest("POST", "/xrpc/social.coves.community.post.create", bytes.NewReader(reqJSON)) - req.Header.Set("Content-Type", "application/json") - - // Register the author user with OAuth middleware and get test token - // For Coves API handlers, use Bearer scheme with OAuth middleware - token := e2eAuth.AddUser(author.DID) - req.Header.Set("Authorization", "Bearer "+token) - - // Execute request through auth middleware + handler. - // Capture a Jetstream replay cursor BEFORE the write so the Part 2 subscription - // (opened after the write) cannot miss the resulting firehose commit. - postCreateCursor := jetstreamCursorNow() - rr := httptest.NewRecorder() - handler := e2eAuth.RequireAuth(http.HandlerFunc(createHandler.HandleCreate)) - handler.ServeHTTP(rr, req) - - // Check response - require.Equal(t, http.StatusOK, rr.Code, "Handler should return 200 OK, body: %s", rr.Body.String()) - - // Parse response - var response posts.CreatePostResponse - err = json.NewDecoder(rr.Body).Decode(&response) - require.NoError(t, err, "Failed to parse response") - - t.Logf("✅ Post created on PDS:") - t.Logf(" URI: %s", response.URI) - t.Logf(" CID: %s", response.CID) - - // ==================================================================================== - // Part 2: TRUE E2E - Real Jetstream Firehose Consumer - // ==================================================================================== - // This part tests the ACTUAL production code path in main.go - // including the WebSocket connection and consumer logic - t.Run("2. Real Jetstream Firehose Consumption", func(t *testing.T) { - t.Logf("\n🔄 TRUE E2E: Subscribing to real Jetstream firehose...") - - // Get PDS hostname for Jetstream filtering - pdsHostname := strings.TrimPrefix(pdsURL, "http://") - pdsHostname = strings.TrimPrefix(pdsHostname, "https://") - pdsHostname = strings.Split(pdsHostname, ":")[0] // Remove port - - // Build Jetstream URL with filters for post records - jetstreamURL := fmt.Sprintf("ws://%s:6008/subscribe?wantedCollections=social.coves.community.post", - pdsHostname) - - t.Logf(" Jetstream URL: %s", jetstreamURL) - t.Logf(" Looking for post URI: %s", response.URI) - t.Logf(" Community DID: %s", community.DID) - - // Setup user service (required by post consumer) - userRepo := postgres.NewUserRepository(db) - identityConfig := identity.DefaultConfig() - plcURL := os.Getenv("PLC_DIRECTORY_URL") - if plcURL == "" { - plcURL = "http://localhost:3002" - } - identityConfig.PLCURL = plcURL - identityResolver := identity.NewResolver(db, identityConfig) - userService := users.NewUserService(userRepo, identityResolver, pdsURL, nil, "") - - // Create post consumer (same as main.go) - postConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db) - - // Channels to receive the event - eventChan := make(chan *jetstream.JetstreamEvent, 10) - errorChan := make(chan error, 1) - done := make(chan bool) - - // Start Jetstream WebSocket subscriber in background - // This creates its own WebSocket connection to Jetstream - go func() { - err := subscribeToJetstreamForPost(ctx, withJetstreamCursor(jetstreamURL, postCreateCursor), community.DID, postConsumer, eventChan, errorChan, done) - if err != nil { - errorChan <- err - } - }() - - // Wait for event or timeout - t.Logf("⏳ Waiting for Jetstream event (max 30 seconds)...") - - select { - case event := <-eventChan: - t.Logf("✅ Received real Jetstream event!") - t.Logf(" Event DID: %s", event.Did) - t.Logf(" Collection: %s", event.Commit.Collection) - t.Logf(" Operation: %s", event.Commit.Operation) - t.Logf(" RKey: %s", event.Commit.RKey) - - // Verify it's for our community - assert.Equal(t, community.DID, event.Did, "Event should be from community repo") - - // Verify post was indexed in AppView database - t.Logf("\n🔍 Querying AppView database for indexed post...") - - indexedPost, err := postRepo.GetByURI(ctx, response.URI) - require.NoError(t, err, "Post should be indexed in AppView") - - t.Logf("✅ Post indexed in AppView:") - t.Logf(" URI: %s", indexedPost.URI) - t.Logf(" CID: %s", indexedPost.CID) - t.Logf(" Author DID: %s", indexedPost.AuthorDID) - t.Logf(" Community: %s", indexedPost.CommunityDID) - t.Logf(" Title: %v", indexedPost.Title) - t.Logf(" Content: %v", indexedPost.Content) - - // Verify all fields match what we sent - assert.Equal(t, response.URI, indexedPost.URI, "URI should match") - assert.Equal(t, response.CID, indexedPost.CID, "CID should match") - assert.Equal(t, author.DID, indexedPost.AuthorDID, "Author DID should match") - assert.Equal(t, community.DID, indexedPost.CommunityDID, "Community DID should match") - assert.Equal(t, title, *indexedPost.Title, "Title should match") - assert.Equal(t, content, *indexedPost.Content, "Content should match") - - // Verify stats initialized correctly - assert.Equal(t, 0, indexedPost.UpvoteCount, "Upvote count should be 0") - assert.Equal(t, 0, indexedPost.DownvoteCount, "Downvote count should be 0") - assert.Equal(t, 0, indexedPost.Score, "Score should be 0") - assert.Equal(t, 0, indexedPost.CommentCount, "Comment count should be 0") - - // Verify timestamps - assert.False(t, indexedPost.CreatedAt.IsZero(), "CreatedAt should be set") - assert.False(t, indexedPost.IndexedAt.IsZero(), "IndexedAt should be set") - - // Signal to stop Jetstream consumer - close(done) - - t.Log("\n✅ Part 2 Complete: TRUE E2E - PDS → Jetstream → Consumer → AppView ✓") - - case err := <-errorChan: - t.Fatalf("❌ Jetstream error: %v", err) - - case <-time.After(30 * time.Second): - t.Fatalf("❌ Timeout: No Jetstream event received within 30 seconds") - } - }) - }) -} - -// subscribeToJetstreamForPost subscribes to real Jetstream firehose and processes post events -// This helper creates a WebSocket connection to Jetstream and waits for post events -func subscribeToJetstreamForPost( - ctx context.Context, - jetstreamURL string, - targetDID string, - consumer *jetstream.PostEventConsumer, - eventChan chan<- *jetstream.JetstreamEvent, - errorChan chan<- error, - done <-chan bool, -) error { - conn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil) - if err != nil { - return fmt.Errorf("failed to connect to Jetstream: %w", err) - } - defer func() { _ = conn.Close() }() - - // ONE deadline for the whole subscription, not one per read: the - // budget is what the caller is willing to wait in total, and a - // per-read deadline would let a busy stream extend it indefinitely. - readDeadline := time.Now().Add(jetstreamReadBudget) - - // Read messages until we find our event or receive done signal - for { - select { - case <-done: - return nil - case <-ctx.Done(): - return ctx.Err() - default: - // Set read deadline to avoid blocking forever - if err := conn.SetReadDeadline(readDeadline); err != nil { - return fmt.Errorf("failed to set read deadline: %w", err) - } - - var event jetstream.JetstreamEvent - err := conn.ReadJSON(&event) - if err != nil { - // Check if it's a timeout (expected) - if websocket.IsCloseError(err, websocket.CloseNormalClosure) { - return fmt.Errorf("Jetstream closed the subscription before the event arrived") - } - if netErr, ok := err.(net.Error); ok && netErr.Timeout() { - // The deadline is the whole budget, so its expiry is the answer: - // no matching event arrived. Reading on would be reading a - // connection gorilla has already marked failed. - return fmt.Errorf("no matching event within %s", jetstreamReadBudget) - } - // For other errors, don't retry reading from a broken connection - return fmt.Errorf("failed to read Jetstream message: %w", err) - } - - // Check if this is a post event for the target DID - if event.Did == targetDID && event.Kind == "commit" && - event.Commit != nil && event.Commit.Collection == "social.coves.community.post" { - // Process the event through the consumer - if err := consumer.HandleEvent(ctx, &event); err != nil { - return fmt.Errorf("failed to process event: %w", err) - } - - // Send to channel so test can verify - select { - case eventChan <- &event: - return nil - case <-time.After(1 * time.Second): - return fmt.Errorf("timeout sending event to channel") - } - } - } - } -} diff --git a/tests/integration/user_journey_e2e_test.go b/tests/integration/user_journey_e2e_test.go index 885bf21..5df4a62 100644 --- a/tests/integration/user_journey_e2e_test.go +++ b/tests/integration/user_journey_e2e_test.go @@ -46,6 +46,43 @@ import ( timelineCore "Coves/internal/core/timeline" ) +// # WHY THIS FILE OVERRIDES jetstreamReadBudget +// +// This journey flaked 2 of 6 full-gate runs during task 12 — always the same +// way: part 3's post commit never reached the subscriber inside 30 seconds, and +// every later part failed behind it. It never reproduces standalone. The test +// alone passes, the whole tests/integration package alone passes, and even the +// gate's exact T1 invocation passes; only a full `make ci`, under build load, +// loses the race. +// +// The mechanism is the one recorded in the loop's cross-iteration note [C]: +// Jetstream account and identity events BYPASS wantedCollections, so the +// package's own `-parallel 26` signup storm floods this subscriber's socket with +// events it did not ask for. It has a fixed total read budget, spends it reading +// account events, and never reaches the one post commit it wants. +// +// Two constants below buy headroom and, more importantly, DIAGNOSABILITY: +// +// - journeyReadBudget (60s) replaces the package-wide jetstreamReadBudget for +// this file's two subscribers, doubling the room a storm has to clear. +// - journeyEventTimeout (75s) is every caller-side wait, and it is strictly +// LONGER than the subscriber's budget on purpose. They used to be equal at +// 30s, a photo finish the subscriber could never win — so the failure always +// surfaced as the caller's opaque "Jetstream timeout" instead of the +// subscriber's own account of what it saw. Ordered this way, the subscriber +// always reports first, and it reports how many events it read and how many +// matched, which is what distinguishes a starved subscriber from a pipeline +// that genuinely dropped the commit. +// +// This is mitigation, not a fix. The real repair is rebuilding this journey on +// testkit.Firehose, which re-dials and cursor-gates instead of spending one +// deadline on whatever arrives; that is task 16's scope, and these constants die +// with the file. +const ( + journeyReadBudget = 60 * time.Second + journeyEventTimeout = 75 * time.Second +) + // TestFullUserJourney_E2E tests the complete user experience from signup to interaction: // 1. User A: Signup → Authenticate → Create Community → Create Post // 2. User B: Signup → Authenticate → Subscribe to Community @@ -270,7 +307,7 @@ func TestFullUserJourney_E2E(t *testing.T) { close(done) case err := <-errorChan: t.Fatalf("❌ Jetstream error: %v", err) - case <-time.After(30 * time.Second): + case <-time.After(journeyEventTimeout): close(done) // Check if simulation fallback is allowed (for CI environments) if os.Getenv("ALLOW_SIMULATION_FALLBACK") == "true" { @@ -338,7 +375,7 @@ func TestFullUserJourney_E2E(t *testing.T) { jetstreamFilterURL := fmt.Sprintf("%s?wantedCollections=social.coves.community.post", jetstreamURL) go func() { - err := subscribeToJetstreamForPost(ctx, withJetstreamCursor(jetstreamFilterURL, postCreateCursor), communityDID, postConsumer, eventChan, errorChan, done) + err := subscribeToJetstreamForPost(ctx, withJetstreamCursor(jetstreamFilterURL, postCreateCursor), communityDID, postConsumer, eventChan, done) if err != nil { errorChan <- err } @@ -350,7 +387,7 @@ func TestFullUserJourney_E2E(t *testing.T) { close(done) case err := <-errorChan: t.Fatalf("❌ Jetstream error: %v", err) - case <-time.After(30 * time.Second): + case <-time.After(journeyEventTimeout): close(done) // Check if simulation fallback is allowed (for CI environments) if os.Getenv("ALLOW_SIMULATION_FALLBACK") == "true" { @@ -865,7 +902,14 @@ func subscribeToJetstreamForCommunity( // ONE deadline for the whole subscription, not one per read: the // budget is what the caller is willing to wait in total, and a // per-read deadline would let a busy stream extend it indefinitely. - readDeadline := time.Now().Add(jetstreamReadBudget) + readDeadline := time.Now().Add(journeyReadBudget) + + // Counted so the budget's expiry can say WHY it expired. A subscriber that + // read thousands of events and matched none was starved by the account-event + // storm described at journeyReadBudget; one that read nothing at all is + // looking at a dead firehose. The two need opposite fixes and used to be + // indistinguishable from the failure message. + var eventsSeen, eventsFromTarget int // The gorilla/websocket library panics after 1000 repeated reads on a failed connection @@ -897,7 +941,12 @@ func subscribeToJetstreamForCommunity( // The deadline is the whole budget, so its expiry is the answer: // no matching event arrived. Reading on would be reading a // connection gorilla has already marked failed. - return fmt.Errorf("no matching event within %s", jetstreamReadBudget) + return fmt.Errorf( + "no matching event within %s: read %d event(s), %d from %s, none of them the "+ + "commit this subscription was waiting for (a large count with no match means "+ + "the subscriber was starved by unfiltered account/identity events, not that "+ + "the firehose dropped the commit)", + journeyReadBudget, eventsSeen, eventsFromTarget, targetDID) } // For any other error, return immediately to avoid re-reading from failed connection @@ -905,6 +954,11 @@ func subscribeToJetstreamForCommunity( return fmt.Errorf("failed to read Jetstream message: %w", err) } + eventsSeen++ + if event.Did == targetDID { + eventsFromTarget++ + } + if event.Did == targetDID && event.Kind == "commit" && event.Commit != nil && event.Commit.Collection == "social.coves.community.profile" { if err := consumer.HandleEvent(ctx, &event); err != nil { @@ -965,3 +1019,90 @@ func simulatePostIndexing(t *testing.T, db *sql.DB, consumer *jetstream.PostEven } require.NoError(t, consumer.HandleEvent(ctx, &event)) } + +// subscribeToJetstreamForPost subscribes to the real Jetstream firehose and +// processes post events through the given consumer. +// +// It was defined in post_e2e_test.go until that file was replaced by +// tests/e2e/post_contract_test.go, and moved here because this is its only +// remaining caller. It is not a helper worth generalising: task 16 rebuilds the +// user journey on testkit's cursor-gated subscriber, and this copy dies with the +// last hand-rolled one. +func subscribeToJetstreamForPost( + ctx context.Context, + jetstreamURL string, + targetDID string, + consumer *jetstream.PostEventConsumer, + eventChan chan<- *jetstream.JetstreamEvent, + done <-chan bool, +) error { + conn, _, err := websocket.DefaultDialer.Dial(jetstreamURL, nil) + if err != nil { + return fmt.Errorf("failed to connect to Jetstream: %w", err) + } + defer func() { _ = conn.Close() }() + + // ONE deadline for the whole subscription, not one per read: the + // budget is what the caller is willing to wait in total, and a + // per-read deadline would let a busy stream extend it indefinitely. + readDeadline := time.Now().Add(journeyReadBudget) + + // Counted so the budget's expiry can say WHY it expired. A subscriber that + // read thousands of events and matched none was starved by the account-event + // storm described at journeyReadBudget; one that read nothing at all is + // looking at a dead firehose. The two need opposite fixes and used to be + // indistinguishable from the failure message. + var eventsSeen, eventsFromTarget int + + for { + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + default: + if err := conn.SetReadDeadline(readDeadline); err != nil { + return fmt.Errorf("failed to set read deadline: %w", err) + } + + var event jetstream.JetstreamEvent + err := conn.ReadJSON(&event) + if err != nil { + if websocket.IsCloseError(err, websocket.CloseNormalClosure) { + return fmt.Errorf("Jetstream closed the subscription before the event arrived") + } + if netErr, ok := err.(net.Error); ok && netErr.Timeout() { + // The deadline is the whole budget, so its expiry is the answer: + // no matching event arrived. Reading on would be reading a + // connection gorilla has already marked failed. + return fmt.Errorf( + "no matching event within %s: read %d event(s), %d from %s, none of them the "+ + "commit this subscription was waiting for (a large count with no match means "+ + "the subscriber was starved by unfiltered account/identity events, not that "+ + "the firehose dropped the commit)", + journeyReadBudget, eventsSeen, eventsFromTarget, targetDID) + } + return fmt.Errorf("failed to read Jetstream message: %w", err) + } + + eventsSeen++ + if event.Did == targetDID { + eventsFromTarget++ + } + + if event.Did == targetDID && event.Kind == "commit" && + event.Commit != nil && event.Commit.Collection == "social.coves.community.post" { + if err := consumer.HandleEvent(ctx, &event); err != nil { + return fmt.Errorf("failed to process event: %w", err) + } + + select { + case eventChan <- &event: + return nil + case <-time.After(1 * time.Second): + return fmt.Errorf("timeout sending event to channel") + } + } + } + } +} -- 2.51.2