From 8686c2aa9c0e08ae1ce819124a428c3deccc71b3 Mon Sep 17 00:00:00 2001 From: Bretton Date: Sat, 08 Aug 2026 18:06:16 +0000 Subject: [PATCH] test(review): whole-branch review batch pins — 2 prod blockers + tool data-safety (task 8) Codex whole-branch review found what per-task TDD structurally missed: P1 oauthScopes grants community.post but NOT postv2 (core write path unscoped); P2 DeciderDeps built with nil Admissions → firehose §8 quota silently disabled in prod; P3 editNote removed from a published lexicon (non-additive). Tool data-safety: P4 blob bytes not copied to author repo; P5 langs/tags/crosspost/bridgedStats stripped (irreversible loss); P6 verify adopts a foreign record at the deterministic rkey; P7 crash-between-delete-and-MarkDone strands a migrated row while Complete goes true; P8 census not preflighted; P9 rkey is a non-TID digest in a key:tid collection. Seams: buildDeciderDeps extraction (bug intact), LegacyPost.RawRecord (lossless source), ListResumable. Co-Authored-By: Claude Fable 5 --- cmd/server/oauth_scopes_test.go | 63 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ cmd/server/wiring.go | 23 +++++++++++++++++------ cmd/server/wiring_quota_test.go | 42 ++++++++++++++++++++++++++++++++++++++++++ internal/core/posts/rematerialize.go | 26 +++++++++++++++++++++++--- internal/core/posts/rematerialize_outer_test.go | 112 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/core/posts/rematerialize_rkey_test.go | 70 +++++++++++++++++++++++++++------------------------------------------- internal/core/posts/rematerialize_test.go | 329 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------ internal/db/postgres/rematerialize_ledger.go | 41 +++++++++++++++++++++++++++++++++++++++++ tests/lexicon_editnote_test.go | 66 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 9 file(s) changed, 714 insertion(s)(+), 58 deletion(s)(-) diff --git a/cmd/server/oauth_scopes_test.go b/cmd/server/oauth_scopes_test.go new file mode 100644 --- /dev/null +++ b/cmd/server/oauth_scopes_test.go @@ -0,0 +1,63 @@ +package main + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The OAuth scopes must grant postv2 writes, or the CORE WRITE PATH is refused +// (whole-branch review, P1). +// +// Under author-owned posts, CreatePost / UpdatePost / delete AND the cutover tool +// all write social.coves.community.postv2 into the AUTHOR's repo through the +// author's own OAuth session. A scope-enforcing PDS rejects a write to a +// collection the granted scopes do not name — so a session minted from +// oauthScopes() that lists only community.post cannot write a single postv2, and +// the whole feature fails at the PDS boundary with an authorization error rather +// than anywhere the AppView can see. +// +// The community.post grant must ALSO survive: the tool DELETES the deprecated +// community.post records through these same sessions during the drain, so +// dropping that scope would strand every legacy record undeleteable. + +// scopeFor returns the granted scope entry for a collection, and whether one +// exists. A scope is "repo:?action=...&action=..." (or bare +// "repo:"). +func scopeFor(scopes []string, collection string) (string, bool) { + prefix := "repo:" + collection + for _, s := range scopes { + if s == prefix || strings.HasPrefix(s, prefix+"?") { + return s, true + } + } + return "", false +} + +func TestOAuthScopes_GrantPostV2CreateUpdateDelete(t *testing.T) { + scopes := oauthScopes() + + postv2, ok := scopeFor(scopes, "social.coves.community.postv2") + require.Truef(t, ok, + "oauthScopes() grants no repo:social.coves.community.postv2 scope. CreatePost, UpdatePost, post.delete and the cutover tool all write postv2 through the author's OAuth "+ + "session, so a scope-enforcing PDS refuses the entire write path. Grant postv2 with create+update+delete. Scopes: %v", scopes) + + for _, action := range []string{"action=create", "action=update", "action=delete"} { + assert.Containsf(t, postv2, action, + "the postv2 scope must grant %s: create is the write, update is post.update (§3.4), delete is post.delete — the tool and the write path perform all three", action) + } +} + +func TestOAuthScopes_RetainCommunityPostThroughTheDrain(t *testing.T) { + scopes := oauthScopes() + + post, ok := scopeFor(scopes, "social.coves.community.post") + require.Truef(t, ok, + "oauthScopes() dropped the deprecated community.post grant. The cutover tool DELETES legacy community.post records through these sessions during the drain (§11); "+ + "without the grant every legacy record is stranded undeleteable. Keep it until the post-drain follow-up retires the collection. Scopes: %v", scopes) + + assert.Containsf(t, post, "action=delete", + "the retained community.post grant must include action=delete — the drain's whole job is deleting those records") +} diff --git a/cmd/server/wiring.go b/cmd/server/wiring.go --- a/cmd/server/wiring.go +++ b/cmd/server/wiring.go @@ -437,11 +437,15 @@ // ErrCommunityNotHosted and the backlog query returns nothing. Both are keyed on // STORED CREDENTIALS rather than on communities.hosted_by_did, which is copied // out of a community's own profile record and can therefore be claimed by any // repo on the network. -func (a *application) buildAcceptanceEngine() *posts.AcceptanceEngine { - repoFactory := posts.NewCommunityRepoFactory(a.communityService) - a.communityWriter = posts.NewCommunityRecordWriter(repoFactory, time.Now) - - decider := posts.NewAdmissionEngineDecider(posts.DeciderDeps{ +// buildDeciderDeps assembles everything the production admission decider reads. +// +// Extracted from buildAcceptanceEngine so the wiring itself is testable: the §8 +// firehose quota is only enforced when DeciderDeps.Admissions is wired +// (decider.go applyQuota short-circuits on a nil counter), and a struct literal +// that silently omits the field disables the abuse control in production with no +// error anywhere. wiring_quota_test.go asserts the field is set. +func (a *application) buildDeciderDeps() posts.DeciderDeps { + return posts.DeciderDeps{ Posts: a.postRepo, Communities: a.communityService, Authorizer: a.aggregatorService, @@ -460,7 +464,14 @@ // Resolved ONCE, here, rather than read per decision — and through the // same helper the write path uses, so the two cannot drift into // disagreeing about who is privileged. TrustedAggregatorDIDs: posts.TrustedAggregatorDIDs(), - }) + } +} + +func (a *application) buildAcceptanceEngine() *posts.AcceptanceEngine { + repoFactory := posts.NewCommunityRepoFactory(a.communityService) + a.communityWriter = posts.NewCommunityRecordWriter(repoFactory, time.Now) + + decider := posts.NewAdmissionEngineDecider(a.buildDeciderDeps()) engine := posts.NewAcceptanceEngine( a.admissionRepo, decider, a.communityWriter, diff --git a/cmd/server/wiring_quota_test.go b/cmd/server/wiring_quota_test.go new file mode 100644 --- /dev/null +++ b/cmd/server/wiring_quota_test.go @@ -0,0 +1,42 @@ +package main + +import ( + "testing" + + "Coves/internal/config" + postgresRepo "Coves/internal/db/postgres" + + "github.com/stretchr/testify/require" +) + +// The production decider must be wired with its firehose quota counter, or the §8 +// abuse control is silently OFF (whole-branch review, P2). +// +// decider.go's applyQuota short-circuits — allowing UNLIMITED admissions — when +// DeciderDeps.Admissions is nil. The wiring built the decider without setting +// that field, so on the live instance every direct-PDS-write / remote-author post +// (ActorUser) got unlimited admission: the per-author-per-community submission +// quota that tasks 3 and 5 built exists in code and does nothing in production. +// +// This pins the WIRING, not the decider (which enforces the quota correctly once +// given the counter): buildDeciderDeps must carry the admissions repo the +// ingestion consumer and engine already share, so the same rows that count as +// admitted are the rows the quota meters. + +func TestBuildDeciderDeps_WiresTheFirehoseQuotaCounter(t *testing.T) { + // A non-nil AdmissionRepository is all this needs — the test never queries it, + // it only asserts the wiring PASSES it through. NewAdmissionRepository(nil) + // wraps a nil DB without touching it, so no infrastructure is required. + admissionRepo := postgresRepo.NewAdmissionRepository(nil) + + a := &application{ + admissionRepo: admissionRepo, + cfg: &config.Config{}, + } + + deps := a.buildDeciderDeps() + + require.NotNilf(t, deps.Admissions, + "buildDeciderDeps left DeciderDeps.Admissions nil, so applyQuota short-circuits and the §8 firehose per-author-per-community quota is DISABLED in production: "+ + "any authenticated author can write unlimited postv2 records naming any community and each is admitted. Set Admissions: a.admissionRepo (the same counter the ingestion consumer writes and the engine settles).") +} diff --git a/internal/core/posts/rematerialize.go b/internal/core/posts/rematerialize.go --- a/internal/core/posts/rematerialize.go +++ b/internal/core/posts/rematerialize.go @@ -115,10 +115,22 @@ // written into. Under author-owned posts this field is dropped from the new // record (postV2From), but it is exactly who to re-author under. AuthorDID string - // Record is the decoded legacy body. postV2From drops its `author` field and - // re-stamps the $type; its createdAt is preserved so the re-materialized post - // keeps its original time. + // Record is the decoded legacy body. + // + // DEPRECATED, LOSSY: PostRecord omits published fields — langs, tags, + // crosspostOf, crosspostChain, bridgedStats — so converting through it before + // deleting the old record IRREVERSIBLY drops them (whole-branch review, P5). + // The conversion must run off RawRecord instead; this field is retained only + // until GREEN removes the lossy path. Record PostRecord + + // RawRecord is the legacy record EXACTLY as it stands in the community repo — + // the lossless source the postv2 is built from. The conversion drops only the + // `author` field and re-stamps `$type`; every other field (including langs, + // tags, crosspostOf, crosspostChain, bridgedStats, facets, embed, labels) is + // carried through byte-for-byte, and createdAt is preserved so the + // re-materialized post keeps its original time. + RawRecord map[string]any } // LegacySource enumerates and deletes the deprecated community.post records. @@ -165,6 +177,14 @@ Discover(ctx context.Context, oldURI, authorDID string) (RematerializeLedgerRow, error) // Get reads one row. found is false when the URI has never been discovered. Get(ctx context.Context, oldURI string) (row RematerializeLedgerRow, found bool, err error) + + // ListResumable returns every row in a non-terminal state (not done, not a + // fallback). It is what makes crash-resume drive off the LEDGER rather than + // the source listing (whole-branch review, P7): a record whose delete + // succeeded but whose MarkDone crashed is GONE from the community repo, so a + // re-run's listRecords can never rediscover it — only the ledger row proves it + // is owed a final MarkDone. + ListResumable(ctx context.Context) ([]RematerializeLedgerRow, error) // RecordPostV2Written moves discovered → postv2_written and records the // postv2 coordinates. diff --git a/internal/core/posts/rematerialize_outer_test.go b/internal/core/posts/rematerialize_outer_test.go --- a/internal/core/posts/rematerialize_outer_test.go +++ b/internal/core/posts/rematerialize_outer_test.go @@ -4,6 +4,8 @@ package posts_test import ( "context" + "net/http" + "net/url" "strings" "testing" "time" @@ -106,6 +108,14 @@ Title: strPtr(title), Content: strPtr("words the author is accountable for"), CreatedAt: "2026-01-02T03:04:05Z", }, + RawRecord: map[string]any{ + "$type": postCollection, + "community": communityAcct.DID, + "author": authorAcct.DID, + "title": title, + "content": "words the author is accountable for", + "createdAt": "2026-01-02T03:04:05Z", + }, } // Real author-repo credentials for the author, over the real PDS. @@ -182,3 +192,105 @@ require.Truef(t, testkit.IsNotFound(getRecordErr(ctx, communityAcct, postCollection, oldRkey)), "the old community.post must stay gone across a re-run") } + +// P4 — embed blob BYTES must be copied to the author's repo, not just referenced +// (whole-branch review, P4). +// +// A blob ref names a CID and not a repository, so a reader resolves it against +// the repo it believes owns the record — after the flip, the AUTHOR's. The legacy +// post's images live in the COMMUNITY's blob store; if the tool copies only the +// embed REFERENCE and then deletes the legacy record, the postv2's media resolves +// against an author repo where the bytes never landed (broken image), and the +// community's now-unreferenced blob becomes garbage-collectable — the only copy, +// gone. The bytes must be uploaded into the author's repo, and the old record must +// not be deleted until they are verified present. +func TestRematerialize_OuterContract_CopiesEmbedBlobToAuthorRepo(t *testing.T) { + t.Parallel() + + pdsServer := testkit.NewPDS(t) + communityAcct := pdsServer.CreateAccount(t, testkit.WithHandlePrefix("rbmc")) + authorAcct := pdsServer.CreateAccount(t, testkit.WithHandlePrefix("rbma")) + ctx := context.Background() + + // A real blob uploaded into the COMMUNITY's blob store, referenced by a legacy + // post's images embed — exactly where a pre-flip post's media lives. + blob := communityAcct.UploadBlob(t, []byte("PNGDATA-re-materialization-blob-bytes"), "image/png") + embed := map[string]any{ + "$type": "social.coves.embed.images", + "images": []any{map[string]any{"image": blob, "alt": "a picture"}}, + } + + oldRkey := testkit.TID() + title := "legacy with media " + testkit.UniqueID(t) + seeded := communityAcct.PutRecord(t, postCollection, oldRkey, map[string]any{ + "$type": postCollection, + "community": communityAcct.DID, + "author": authorAcct.DID, + "title": title, + "embed": embed, + "createdAt": "2026-01-02T03:04:05Z", + }) + + legacy := posts.LegacyPost{ + URI: seeded.URI, + CID: seeded.CID, + CommunityDID: communityAcct.DID, + AuthorDID: authorAcct.DID, + Record: posts.PostRecord{ + Type: postCollection, + Community: communityAcct.DID, + Author: authorAcct.DID, + Title: strPtr(title), + Embed: embed, + CreatedAt: "2026-01-02T03:04:05Z", + }, + RawRecord: map[string]any{ + "$type": postCollection, + "community": communityAcct.DID, + "author": authorAcct.DID, + "title": title, + "embed": embed, + "createdAt": "2026-01-02T03:04:05Z", + }, + } + + authorFactory := func(_ context.Context, _ string, _ *oauth.ClientSessionData) (posts.AuthorRepo, error) { + generic, err := pds.NewFromAccessToken(pdsServer.URL(), authorAcct.DID, authorAcct.AccessToken) + require.NoError(t, err) + repo, ok := generic.(posts.AuthorRepo) + require.True(t, ok) + return repo, nil + } + communityGeneric, err := pds.NewFromAccessToken(pdsServer.URL(), communityAcct.DID, communityAcct.AccessToken) + require.NoError(t, err) + communityRepo, ok := communityGeneric.(posts.CommunityRepo) + require.True(t, ok) + writer := posts.NewCommunityRecordWriter( + func(_ context.Context, _ string) (posts.CommunityRepo, error) { return communityRepo, nil }, time.Now) + + source := &realLegacySource{community: communityGeneric, staged: []posts.LegacyPost{legacy}} + ledger := postgres.NewRematerializeLedger(testkit.DB(t)) + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authorFactory, Acceptances: writer} + + _, err = tool.RematerializeOne(ctx, legacy) + require.NoError(t, err) + + // The blob's BYTES must now be fetchable from the AUTHOR's repo. sync.getBlob + // serves a blob only from a repo that actually holds it, so a 200 here proves + // the bytes were copied — a 404 (the current behaviour) proves only the + // reference was, and the post's media is broken the moment the old record goes. + require.Equalf(t, 200, getBlobStatus(t, pdsServer.URL(), authorAcct.DID, blob.CID()), + "the embed blob %s was not copied into the author's repo: the postv2 references a CID whose bytes live only in the community's blob store, "+ + "which is now garbage-collectable and about to be the only copy lost", blob.CID()) +} + +// getBlobStatus fetches a blob from a repo via com.atproto.sync.getBlob and +// returns the HTTP status — 200 if the repo holds the blob, 404 if it does not. +func getBlobStatus(t *testing.T, pdsURL, did, cid string) int { + t.Helper() + req := pdsURL + "/xrpc/com.atproto.sync.getBlob?did=" + url.QueryEscape(did) + "&cid=" + url.QueryEscape(cid) + resp, err := http.Get(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + return resp.StatusCode +} diff --git a/internal/core/posts/rematerialize_rkey_test.go b/internal/core/posts/rematerialize_rkey_test.go --- a/internal/core/posts/rematerialize_rkey_test.go +++ b/internal/core/posts/rematerialize_rkey_test.go @@ -1,12 +1,10 @@ package posts import ( - "crypto/sha256" - "encoding/base32" - "strings" "testing" "time" + "github.com/bluesky-social/indigo/atproto/syntax" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -14,50 +12,37 @@ // The single highest-risk detail in the whole cutover: the postv2 record key the // re-materialization tool writes at (docs/PRD_AUTHOR_OWNED_POSTS.md §11 step 4). // -// A wrong rkey does not fail loudly — it MINTS DUPLICATES. If a re-run computes a -// different key than the first run, createAuthorRecord's converge-by-read never -// fires (the create-only guard is against a DIFFERENT key, which is empty), so a -// second postv2 lands for the same legacy post. Every strongRef built from the -// first run — the acceptance's pinned subject, every comment and vote — points at -// the first record; the second is an orphan duplicate. So this file pins the key -// harder than anything else in the suite: +// It has TWO hard constraints that pull in different directions, and the RED-1 +// digest scheme satisfied only one of them (whole-branch review, P9): // -// 1. It is DETERMINISTIC — two computations of the same old URI are identical. -// 2. It is a PURE FUNCTION OF THE OLD URI ALONE — nothing submission-time -// (fingerprint, dedupe bucket, clock) leaks in, because the migration has -// none of that and a re-run must reproduce the key from the old record only. -// 3. It is NOT SubmissionRkey — the write-path key needs exactly the -// submission-time material this tool lacks, so a tool that reused it would -// draw a fresh key every run and duplicate every post. -// 4. It is the SubjectRkey DIGEST SCHEME applied to the OLD URI: unpadded -// lowercase base32 of the SHA-256 of the URI bytes — total over the legal -// URI space and collision-free, the scheme the write path already trusts. +// 1. STABLE PURE FUNCTION OF THE OLD URI. A wrong or non-deterministic key does +// not fail loudly — it MINTS DUPLICATES: if a re-run computes a different key, +// createAuthorRecord's converge-by-read never fires (the create-only guard is +// against a DIFFERENT, empty key), so a second postv2 lands for one legacy +// post and every strongRef built from the first dangles. So the key must be a +// deterministic function of the OLD URI ALONE — nothing submission-time +// (fingerprint, bucket, clock), which the migration does not have and a re-run +// could not reproduce. In particular it must NOT be SubmissionRkey. // -// The expected value is re-derived here from stdlib rather than by calling -// SubjectRkey, so a bug that changed BOTH the helper and a naive expectation -// together cannot hide: this is an independent check of the derivation. - -// independentRematerializeRkey recomputes the pinned scheme straight from stdlib. -func independentRematerializeRkey(oldURI string) string { - digest := sha256.Sum256([]byte(oldURI)) - return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(digest[:])) -} +// 2. IT MUST BE A VALID TID. The postv2 lexicon declares "key": "tid", so a +// validating PDS rejects any rkey that is not a TID, and feed ordering reads +// the timestamp OUT of the key. The RED-1 scheme, SubjectRkey (a 52-char +// base32 SHA-256 digest), is NOT a TID — syntax.ParseTID rejects it — so a +// conformant PDS would refuse every re-materialized record. The key must be a +// deterministic TID derived purely from the old URI (a hashed timestamp+clock, +// e.g. tidepool's DeterministicTID shape), NOT SubmissionRkey (which mixes in +// submission-time material) and NOT SubjectRkey (which is not a TID at all). -func TestRematerializeRkey_IsTheDigestOfTheOldURI(t *testing.T) { +func TestRematerializeRkey_IsAValidTID(t *testing.T) { oldURI := "at://did:plc:community2222222222222222/social.coves.community.post/3kqijkl2m4c2r" - got := RematerializeRkey(oldURI) + rkey := RematerializeRkey(oldURI) - assert.Equalf(t, independentRematerializeRkey(oldURI), got, - "the re-materialization rkey must be the unpadded lowercase base32 SHA-256 digest of the OLD URI (the SubjectRkey scheme applied to the legacy record). "+ - "A different scheme means a re-run computes a different key, the create-only converge never fires, and a second postv2 is minted for one legacy post") - - // The digest scheme is a fixed 52 characters drawn entirely from the - // rkey-safe lowercase base32 charset — the property that makes it total over - // the legal URI space (§3.2's argument for a digest over a readable transform). - assert.Lenf(t, got, 52, "the digest rkey is a fixed 52 characters for any input; %q is not", got) - assert.Truef(t, got == strings.ToLower(got), "the rkey must be lowercase — an uppercase key is a DIFFERENT key to a PDS that treats rkeys as opaque bytes") - assert.NotContainsf(t, got, "=", "base32 padding '=' is outside the atProto record-key charset and must be dropped") + parsed, err := syntax.ParseTID(rkey) + require.NoErrorf(t, err, + "the re-materialization rkey %q is not a valid TID. The postv2 lexicon declares key:tid, so a validating PDS rejects a non-TID rkey and feed ordering cannot read a timestamp out of it. "+ + "The RED-1 SubjectRkey digest scheme is a 52-char base32 hash, which is not a TID — derive a deterministic TID from the old URI instead", rkey) + assert.Equalf(t, rkey, parsed.String(), "ParseTID must round-trip the rkey unchanged") } func TestRematerializeRkey_IsDeterministic(t *testing.T) { @@ -67,7 +52,7 @@ first := RematerializeRkey(oldURI) second := RematerializeRkey(oldURI) require.Equalf(t, first, second, - "two computations of the re-materialization rkey for the same old URI must be identical, or a crash-resumed run cannot converge on the record its first attempt wrote") + "two computations of the re-materialization rkey for the same old URI must be identical, or a crash-resumed run cannot converge on the record its first attempt wrote — it mints a duplicate") } func TestRematerializeRkey_DependsOnlyOnTheOldURI(t *testing.T) { @@ -91,7 +76,6 @@ // content could collide with a re-materialized post. oldURI := "at://did:plc:community2222222222222222/social.coves.community.post/3kqijkl2m4c2r" communityDID := "did:plc:community2222222222222222" - // A representative SubmissionRkey over unrelated but plausible material. submission := SubmissionRkey(communityDID, "d41d8cd98f00b204e9800998ecf8427e", 0, 5*time.Minute) assert.NotEqualf(t, submission, RematerializeRkey(oldURI), diff --git a/internal/core/posts/rematerialize_test.go b/internal/core/posts/rematerialize_test.go --- a/internal/core/posts/rematerialize_test.go +++ b/internal/core/posts/rematerialize_test.go @@ -4,7 +4,9 @@ package posts_test import ( "context" + "encoding/json" "fmt" + "strings" "sync" "testing" @@ -19,6 +21,51 @@ "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// callLog is an ordered record of the load-bearing calls a run makes, shared by +// the fake factory, author repo, and legacy source, so a test can assert on +// ORDERING that no outcome value reveals — specifically that the credential +// census runs BEFORE any repo mutation (P8). +type callLog struct { + mu sync.Mutex + events []string +} + +func (l *callLog) note(event string) { + if l == nil { + return + } + l.mu.Lock() + defer l.mu.Unlock() + l.events = append(l.events, event) +} + +func (l *callLog) snapshot() []string { + l.mu.Lock() + defer l.mu.Unlock() + return append([]string(nil), l.events...) +} + +// indexOf returns the position of the first event equal to want, or -1. +func indexOf(events []string, want string) int { + for i, e := range events { + if e == want { + return i + } + } + return -1 +} + +// firstMutationIndex is the position of the earliest repo mutation in the log — +// a postv2 write, an acceptance write, or a legacy delete. -1 if none happened. +func firstMutationIndex(events []string) int { + for i, e := range events { + if strings.HasPrefix(e, "write:") || strings.HasPrefix(e, "delete:") || strings.HasPrefix(e, "accept:") { + return i + } + } + return -1 +} + // The re-materialization state machine, against a REAL migration-037 ledger and // FAKE repos (docs/PRD_AUTHOR_OWNED_POSTS.md §11 the rev-2.8 deploy runbook). // @@ -77,10 +124,17 @@ mu sync.Mutex records map[string]*pds.RecordResponse // rkey -> record putErr error // one-shot injected put failure getCIDAt map[string]string // rkey -> CID GetRecord should report (verify-window override) + blobs map[string]bool // blob CIDs uploaded into this repo (P4) + log *callLog // shared ordering log (P8), nil when unused } func newFakeAuthorRepo(did string) *fakeAuthorRepo { - return &fakeAuthorRepo{did: did, records: map[string]*pds.RecordResponse{}, getCIDAt: map[string]string{}} + return &fakeAuthorRepo{ + did: did, + records: map[string]*pds.RecordResponse{}, + getCIDAt: map[string]string{}, + blobs: map[string]bool{}, + } } func (r *fakeAuthorRepo) recordCount() int { @@ -89,6 +143,17 @@ defer r.mu.Unlock() return len(r.records) } +// writtenBody returns the serialised body the tool wrote at a rkey, so a test can +// assert on the postv2 record's fields (P5/P6). +func (r *fakeAuthorRepo) writtenBody(rkey string) map[string]any { + r.mu.Lock() + defer r.mu.Unlock() + if rec, ok := r.records[rkey]; ok { + return rec.Value + } + return nil +} + func (r *fakeAuthorRepo) GetRecord(_ context.Context, collection, rkey string) (*pds.RecordResponse, error) { r.mu.Lock() defer r.mu.Unlock() @@ -121,9 +186,17 @@ // it by reading the standing record back rather than minting a second. return nil, pds.ErrSwapConflict } + r.log.note("write:" + r.did) + uri := "at://" + r.did + "/" + collection + "/" + rkey cid := deterministicCID(rkey) - body, _ := record.(map[string]any) + // The record is captured as its JSON shape, so a test inspects exactly what + // the tool serialised — whether it passed a struct or a map — which is what + // the field-preservation pin (P5) reads. + var body map[string]any + if raw, err := json.Marshal(record); err == nil { + _ = json.Unmarshal(raw, &body) + } r.records[rkey] = &pds.RecordResponse{URI: uri, CID: cid, Value: body} return &pds.RecordCommit{URI: uri, CID: cid, CommitRev: "3krematputxxx"}, nil } @@ -135,27 +208,53 @@ delete(r.records, rkey) return nil } +// UploadBlob records that a blob's bytes were copied into THIS repo — the P4 +// property that the postv2's media resolves against the author, not the +// community whose repo the old record is being deleted from. func (r *fakeAuthorRepo) UploadBlob(_ context.Context, data []byte, mimeType string) (*blobs.BlobRef, error) { - return nil, fmt.Errorf("fakeAuthorRepo.UploadBlob: not used by re-materialization") + r.mu.Lock() + defer r.mu.Unlock() + r.log.note("blob:" + r.did) + // A CID derived from the bytes, so the test can match the ref the tool then + // embeds against what actually landed here. + cid := blobCIDFor(data) + r.blobs[cid] = true + return &blobs.BlobRef{}, nil +} + +func (r *fakeAuthorRepo) hasBlob(cid string) bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.blobs[cid] } func (r *fakeAuthorRepo) DID() string { return r.did } -// seedStanding puts a postv2 record straight into the repo, so a resume test can -// stage "the first run already wrote this". +// seedStanding puts an empty postv2 record straight into the repo, so a resume +// test can stage "the first run already wrote this". func (r *fakeAuthorRepo) seedStanding(collection, rkey string) { + r.seedStandingBody(collection, rkey, map[string]any{}) +} + +// seedStandingBody stages a record with a specific body already standing at a +// rkey, so the body-verify pin (P6) can put a DIFFERENT record at the target key. +func (r *fakeAuthorRepo) seedStandingBody(collection, rkey string, body map[string]any) { r.mu.Lock() defer r.mu.Unlock() uri := "at://" + r.did + "/" + collection + "/" + rkey - r.records[rkey] = &pds.RecordResponse{URI: uri, CID: deterministicCID(rkey), Value: map[string]any{}} + r.records[rkey] = &pds.RecordResponse{URI: uri, CID: deterministicCID(rkey), Value: body} } +// blobCIDFor is a deterministic stand-in CID for a blob's bytes. +func blobCIDFor(data []byte) string { return "bafkreiblob" + fmt.Sprintf("%x", len(data)) } + // fakeAuthorFactory hands out fake author repos by DID, and answers // ErrNoAuthorCredentials for the DIDs marked as unrestorable — exactly what the // production factory does for an aggregator whose stored session is gone. type fakeAuthorFactory struct { repos map[string]*fakeAuthorRepo noCreds map[string]bool + log *callLog // shared ordering log (P8), nil when unused } func newFakeAuthorFactory() *fakeAuthorFactory { @@ -167,12 +266,17 @@ if r, ok := f.repos[did]; ok { return r } r := newFakeAuthorRepo(did) + r.log = f.log f.repos[did] = r return r } func (f *fakeAuthorFactory) factory() posts.AuthorRepoFactory { return func(_ context.Context, authorDID string, _ *oauth.ClientSessionData) (posts.AuthorRepo, error) { + // Every credential resolution is logged, mutating or not, so the census- + // first pin (P8) can assert the tool resolves EVERY author before it + // mutates ANY repo. + f.log.note("resolve:" + authorDID) if f.noCreds[authorDID] { return nil, fmt.Errorf("resuming the stored session of %s: %w", authorDID, posts.ErrNoAuthorCredentials) } @@ -180,6 +284,7 @@ r, ok := f.repos[authorDID] if !ok { return nil, fmt.Errorf("opening the repository of %s: %w", authorDID, posts.ErrNoAuthorCredentials) } + r.log = f.log return r, nil } } @@ -193,12 +298,15 @@ mu sync.Mutex acceptanceCmds []posts.CommunityWriteCommand writeErr error // one-shot injected failure otherCalled []string + log *callLog // shared ordering log (P8), nil when unused } func (s *spyAcceptanceWriter) WriteAcceptance(_ context.Context, cmd posts.CommunityWriteCommand) (posts.CommunityWriteResult, error) { s.mu.Lock() defer s.mu.Unlock() + s.log.note("accept:" + cmd.CommunityDID) + if s.writeErr != nil { err := s.writeErr s.writeErr = nil @@ -263,6 +371,7 @@ mu sync.Mutex posts []posts.LegacyPost deleted map[string]int deleteErr map[string]error + log *callLog // shared ordering log (P8), nil when unused } func newFakeLegacySource(ps ...posts.LegacyPost) *fakeLegacySource { @@ -278,6 +387,7 @@ func (s *fakeLegacySource) DeleteLegacyPost(_ context.Context, legacy posts.LegacyPost) error { s.mu.Lock() defer s.mu.Unlock() + s.log.note("delete:" + legacy.URI) s.deleted[legacy.URI]++ if err, ok := s.deleteErr[legacy.URI]; ok { delete(s.deleteErr, legacy.URI) @@ -311,6 +421,15 @@ Author: authorDID, Title: strPtr(title), Content: strPtr("words the author is accountable for"), CreatedAt: "2026-01-02T03:04:05Z", + }, + // The lossless source the postv2 must be built from (P5). + RawRecord: map[string]any{ + "$type": "social.coves.community.post", + "community": communityDID, + "author": authorDID, + "title": title, + "content": "words the author is accountable for", + "createdAt": "2026-01-02T03:04:05Z", }, } } @@ -582,3 +701,201 @@ require.Lenf(t, calls, 1, "the acceptance must be written exactly once, through the direct writer") assert.Emptyf(t, writer.otherCalled, "the tool called a moderation writer (%v); re-materialization only ever writes an acceptance, and never re-decides removal", writer.otherCalled) } + +// P5 — the conversion must be LOSSLESS. PostRecord omits published fields, so +// converting through it drops langs/tags/crosspostOf/crosspostChain/bridgedStats +// before the old record is deleted: irreversible loss (whole-branch review, P5). +func TestRematerialize_PreservesEveryPublishedField(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + authors.repo(rematAuthorDID) + writer := &spyAcceptanceWriter{} + + rkey := testkit.TID() + oldURI := "at://" + rematCommunityDID + "/social.coves.community.post/" + rkey + + // A legacy record carrying EVERY published field the postv2 lexicon keeps. + raw := map[string]any{ + "$type": "social.coves.community.post", + "community": rematCommunityDID, + "author": rematAuthorDID, + "title": "rich " + testkit.UniqueID(t), + "content": "body with everything", + "createdAt": "2026-01-02T03:04:05Z", + "langs": []any{"en", "fr"}, + "tags": []any{"golang", "atproto"}, + "facets": []any{map[string]any{"index": map[string]any{"byteStart": float64(0), "byteEnd": float64(4)}}}, + "embed": map[string]any{"$type": "social.coves.embed.external", "external": map[string]any{"uri": "https://example.com"}}, + "labels": map[string]any{"$type": "com.atproto.label.defs#selfLabels", "values": []any{map[string]any{"val": "spoiler"}}}, + "crosspostOf": map[string]any{"uri": "at://did:plc:other/social.coves.community.postv2/abc", "cid": "bafyreicrosspost"}, + "crosspostChain": []any{map[string]any{"uri": "at://did:plc:other/social.coves.community.postv2/abc", "cid": "bafyreicrosspost"}}, + "bridgedStats": map[string]any{"upvotes": float64(42)}, + } + legacy := posts.LegacyPost{ + URI: oldURI, + CID: "bafyreilegacy" + rkey, + CommunityDID: rematCommunityDID, + AuthorDID: rematAuthorDID, + RawRecord: raw, + } + source := newFakeLegacySource(legacy) + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + _, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err) + + body := authors.repo(rematAuthorDID).writtenBody(posts.RematerializeRkey(oldURI)) + require.NotNil(t, body, "the tool wrote no postv2 body") + + // Only two fields change: author is dropped, $type is re-stamped. Everything + // else survives byte-for-byte. + assert.Equalf(t, posts.PostV2Collection, body["$type"], "the $type must be re-stamped to postv2") + _, hasAuthor := body["author"] + assert.Falsef(t, hasAuthor, "the author field must be dropped — the repo signature is the authorship anchor") + + for _, field := range []string{"community", "title", "content", "createdAt", "langs", "tags", "facets", "embed", "labels", "crosspostOf", "crosspostChain", "bridgedStats"} { + assert.Equalf(t, raw[field], body[field], + "the postv2 dropped or altered %q; converting through the lossy PostRecord loses published fields the old record can never be recovered from (P5)", field) + } +} + +// P6 — verify must compare the STANDING record's BODY to the intended conversion, +// not merely its CID. A DIFFERENT record already standing at the deterministic +// rkey would otherwise be adopted as "the post" and the legacy original deleted +// (whole-branch review, P6). +func TestRematerialize_RefusesWhenADifferentRecordStandsAtTheRkey(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + authorRepo := authors.repo(rematAuthorDID) + writer := &spyAcceptanceWriter{} + legacy := legacyPost(t, rematCommunityDID, rematAuthorDID) + source := newFakeLegacySource(legacy) + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + // A DIFFERENT record already stands at the target rkey — its CID is the one a + // fresh write would get (so a CID-only verify passes), but its body is NOT this + // legacy post's conversion. createAuthorRecord converges by read onto it; the + // tool must notice the body is wrong and REFUSE rather than delete the original. + rkey := posts.RematerializeRkey(legacy.URI) + authorRepo.seedStandingBody(posts.PostV2Collection, rkey, map[string]any{ + "$type": posts.PostV2Collection, + "community": rematCommunityDID, + "title": "a completely different post that happens to sit at this key", + "content": "not the legacy post's content", + "createdAt": "2020-01-01T00:00:00Z", + }) + + _, err := tool.RematerializeOne(context.Background(), legacy) + require.Errorf(t, err, + "a different record standing at the deterministic rkey was accepted without a body check; the tool must verify the standing record IS this legacy post's conversion") + + assert.Equalf(t, 0, source.deleteCount(legacy.URI), + "VERIFY BEFORE DELETE: with a foreign record at the rkey, the legacy original must NOT be deleted — deleting it destroys the only copy of the real post") + + row, found, err := ledger.Get(context.Background(), legacy.URI) + require.NoError(t, err) + require.True(t, found) + assert.NotEqualf(t, posts.RematerializeDone, row.State, "a body-mismatched record must never reach done") + assert.NotEqualf(t, posts.RematerializeMigrated, row.State, "a body-mismatched record must never reach the migrated checkpoint") +} + +// P7 — crash-resume must be driven by the LEDGER, not the source listing, and +// Complete must require every non-fallback row done — not merely zero fallbacks +// (whole-branch review, P7). +func TestRematerialize_Run_ReconcilesStrandedMigratedRowFromLedger(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + writer := &spyAcceptanceWriter{} + ctx := context.Background() + + // A row that crashed between DeleteLegacyPost and MarkDone: the postv2 and + // acceptance are done, the old record was ALREADY deleted (so the source's + // listRecords can never return it again), but the ledger is stuck at migrated. + strandedURI := "at://" + rematCommunityDID + "/social.coves.community.post/" + testkit.TID() + newRkey := posts.RematerializeRkey(strandedURI) + newURI := "at://" + rematAuthorDID + "/social.coves.community.postv2/" + newRkey + newCID := deterministicCID(newRkey) + _, err := ledger.Discover(ctx, strandedURI, rematAuthorDID) + require.NoError(t, err) + require.NoError(t, ledger.RecordPostV2Written(ctx, strandedURI, newURI, newCID, newRkey)) + require.NoError(t, ledger.MarkVerified(ctx, strandedURI)) + require.NoError(t, ledger.MarkMigrated(ctx, strandedURI)) + + // The source does NOT list the stranded record — its community.post is gone. + source := newFakeLegacySource() + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + report, err := tool.Run(ctx) + require.NoError(t, err) + + row, found, err := ledger.Get(ctx, strandedURI) + require.NoError(t, err) + require.True(t, found) + assert.Equalf(t, posts.RematerializeDone, row.State, + "a migrated row whose record was already deleted was left stuck: resume reads only the source listing, which can never rediscover a deleted record. Resume must drive off ListResumable (the ledger).") + + // Complete must reflect the TRUE state: false while any non-terminal, non- + // fallback row survives — not merely when fallbacks are zero. + nonTerminal := 0 + for state, n := range report.ByState { + if state != posts.RematerializeDone && !posts.IsFallback(state) { + nonTerminal += n + } + } + if nonTerminal > 0 { + assert.Falsef(t, report.Complete, + "Complete is true with %d non-terminal non-fallback row(s) surviving; Complete=(Fallbacks==0) lets the operator run the irreversible legacy removal over half-migrated posts", nonTerminal) + } else { + assert.Truef(t, report.Complete, "every row is terminal-done or fallback, so the run is complete") + } +} + +// P8 — the credential census runs FIRST: a non-mutating preflight over every +// discovered author before ANY repo is mutated, so the operator learns the +// fallback set before a single record is touched (whole-branch review, P8). +func TestRematerialize_Run_ResolvesAllCredentialsBeforeAnyMutation(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + log := &callLog{} + authors := newFakeAuthorFactory() + authors.log = log + writer := &spyAcceptanceWriter{log: log} + + // A migratable author FIRST, then a no-creds author. Under a mutate-as-you-go + // run, the first record is fully written and deleted BEFORE the second author's + // missing credentials are ever discovered — the exact ordering §11 step 3 + // forbids. + withCreds := "did:plc:hascredshascredshascreds" + noCreds := "did:plc:nocredsnocredsnocredsnoc" + authors.repo(withCreds) + authors.noCreds[noCreds] = true + + first := legacyPost(t, rematCommunityDID, withCreds) + second := legacyPost(t, rematCommunityDID, noCreds) + source := &fakeLegacySource{posts: []posts.LegacyPost{first, second}, deleted: map[string]int{}, deleteErr: map[string]error{}, log: log} + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + _, err := tool.Run(context.Background()) + require.NoError(t, err) + + events := log.snapshot() + resolveNoCreds := indexOf(events, "resolve:"+noCreds) + firstMutation := firstMutationIndex(events) + + require.NotEqualf(t, -1, resolveNoCreds, "the no-creds author was never resolved: %v", events) + require.NotEqualf(t, -1, firstMutation, "no repo was mutated at all, so this test proves nothing: %v", events) + assert.Lessf(t, resolveNoCreds, firstMutation, + "the no-creds author's credentials were checked AFTER the first repo mutation (%v). The census must run FIRST — a non-mutating credential "+ + "preflight over every author before any postv2 write, acceptance, or delete — so the fallback set is known before a record is touched (§11 step 3)", events) +} diff --git a/internal/db/postgres/rematerialize_ledger.go b/internal/db/postgres/rematerialize_ledger.go --- a/internal/db/postgres/rematerialize_ledger.go +++ b/internal/db/postgres/rematerialize_ledger.go @@ -86,6 +86,47 @@ row.Reason = reason.String return row, true, nil } +// ListResumable returns every row still in a non-terminal state — the ledger- +// driven resume set (whole-branch review, P7). A migrated row whose delete +// succeeded but whose MarkDone crashed is GONE from the community repo, so only +// this query — never the source's listRecords — can rediscover it. +func (l *rematerializeLedger) ListResumable(ctx context.Context) ([]posts.RematerializeLedgerRow, error) { + rows, err := l.db.QueryContext(ctx, ` + SELECT old_uri, state, author_did, new_uri, new_cid, new_rkey, reason, created_at, updated_at + FROM post_rematerialization_ledger + WHERE state NOT IN ('done', 'fallback_left_legacy', 'fallback_no_creds') + ORDER BY created_at + `) + if err != nil { + return nil, fmt.Errorf("listing resumable ledger rows: %w", err) + } + defer func() { _ = rows.Close() }() + + var out []posts.RematerializeLedgerRow + for rows.Next() { + var ( + row posts.RematerializeLedgerRow + state string + authorDID sql.NullString + newURI sql.NullString + newCID sql.NullString + newRkey sql.NullString + reason sql.NullString + ) + if err := rows.Scan(&row.OldURI, &state, &authorDID, &newURI, &newCID, &newRkey, &reason, &row.CreatedAt, &row.UpdatedAt); err != nil { + return nil, fmt.Errorf("scanning a resumable ledger row: %w", err) + } + row.State = posts.RematerializeState(state) + row.AuthorDID = authorDID.String + row.NewURI = newURI.String + row.NewCID = newCID.String + row.NewRkey = newRkey.String + row.Reason = reason.String + out = append(out, row) + } + return out, rows.Err() +} + // RecordPostV2Written moves discovered → postv2_written and records the postv2 // coordinates the resume path reads back. func (l *rematerializeLedger) RecordPostV2Written(ctx context.Context, oldURI, newURI, newCID, newRkey string) error { diff --git a/tests/lexicon_editnote_test.go b/tests/lexicon_editnote_test.go new file mode 100644 --- /dev/null +++ b/tests/lexicon_editnote_test.go @@ -0,0 +1,66 @@ +package tests + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// editNote is a PUBLISHED field, and removing it is a non-additive break (whole- +// branch review, P3). +// +// social.coves.community.post.update ships on main declaring an optional +// `editNote` string input. The atProto evolution rules are explicit that a +// published lexicon may only change additively — non-optional fields cannot be +// removed and, for an INPUT schema, dropping even an optional field breaks any +// client that still sends it: a stricter consumer built against the new schema +// rejects the request as carrying an unknown property. The branch deleted +// editNote outright. It must stay declared (a deprecated-optional input is fine); +// retiring it, if ever, is a new-NSID change, not an in-place deletion. +// +// Asserted against the raw schema JSON for the same reason the removedPost shape +// is: an open lexicon means no request sample can prove an optional input +// property is ABSENT — reading the schema source is the only thing that fails +// when the field is missing. + +const postUpdatePath = "../internal/atproto/lexicon/social/coves/community/post/update.json" + +func TestPostUpdate_StillDeclaresEditNote(t *testing.T) { + doc := readLexiconJSON(t, postUpdatePath) + + // defs.main.input.schema.properties.editNote must still exist. + main := mustChild(t, doc, "defs", "main") + input := mustChild(t, main, "input") + schema := mustChild(t, input, "schema") + properties := mustChild(t, schema, "properties") + + editNote, ok := properties["editNote"].(map[string]interface{}) + require.Truef(t, ok, + "social.coves.community.post.update no longer declares the `editNote` input. It ships on main (published), so removing it is a NON-ADDITIVE break: "+ + "a client that still sends editNote is rejected by a stricter consumer. Restore it as a deprecated-optional string input.") + + assert.Equalf(t, "string", editNote["type"], + "editNote must stay a string input to remain compatible with the published shape") + + // It must NOT be required — the whole point is that it is an optional, + // backward-compatible input. + if required, ok := schema["required"].([]interface{}); ok { + for _, r := range required { + assert.NotEqualf(t, "editNote", r, "editNote must remain OPTIONAL; making it required is itself a non-additive break") + } + } +} + +// mustChild descends one level into a lexicon tree, failing if the key is absent +// or not an object. +func mustChild(t *testing.T, node map[string]interface{}, path ...string) map[string]interface{} { + t.Helper() + current := node + for _, key := range path { + next, ok := current[key].(map[string]interface{}) + require.Truef(t, ok, "lexicon path element %q is missing or not an object", key) + current = next + } + return current +} -- tangled.sh