From 2e2853e2f664b3d4783ab07512619abf6d366607 Mon Sep 17 00:00:00 2001 From: Bretton Date: Sat, 08 Aug 2026 17:00:21 +0000 Subject: [PATCH] test(rematerialize): RED — the cutover re-materialization tool contract (task 8) Migration-037 ledger schema, rkey-purity (digest of the old URI, never SubmissionRkey), state-machine + crash-resume at every state, verify- before-delete (CID re-read), no-forge fallback (no-creds → left legacy, census gates completion), never-re-decides (structural + behavioral). Deliverable B retiered T2→T1: the e2e constitution forbids the DB-clone assert the ledger requires; outer proof is real-PDS+Postgres T1, the pipeline leg already covered by author_post_contract for any producer. PRD §11 deploy runbook added (legacy removal + fingerprint retype deferred to a post-prod-drain follow-up — prod still holds community.post at merge time). Rule-7 stubs only. Co-Authored-By: Claude Fable 5 --- docs/PRD_AUTHOR_OWNED_POSTS.md | 49 +++++++++++++++++++++++++++++++++++++++++++++++++ internal/core/posts/rematerialize.go | 241 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/core/posts/rematerialize_outer_test.go | 184 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/core/posts/rematerialize_rkey_test.go | 99 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/core/posts/rematerialize_test.go | 584 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/db/postgres/admission_repo_schema_test.go | 13 ++++++++----- internal/db/postgres/rematerialize_ledger.go | 64 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/db/postgres/rematerialize_ledger_schema_test.go | 191 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 8 file(s) changed, 1420 insertion(s)(+), 5 deletion(s)(-) diff --git a/docs/PRD_AUTHOR_OWNED_POSTS.md b/docs/PRD_AUTHOR_OWNED_POSTS.md --- a/docs/PRD_AUTHOR_OWNED_POSTS.md +++ b/docs/PRD_AUTHOR_OWNED_POSTS.md @@ -708,3 +708,52 @@ of the post record entirely. 4. **Private communities** (Beta): acceptance records leak subject URIs on the public firehose; private-community design must address opaque/hashed subjects. Out of scope for the flip. + +--- + +## 11. Cutover deploy runbook (task-8 plan review, rev 2.8) + +The re-materialization tool ships in the loop; the **prod run and the legacy +removal do not** — they are manual, deploy-sequenced, and gated on facts only +the operator can confirm. Getting the order wrong makes live posts vanish or +relaunders a removed post, so the sequence is fixed: + +1. **Deploy the branch** (tool + migration 037 ledger + docs). Legacy read/ + ingest code stays intact — prod still has `community.post` rows. +2. **Maintenance window: stop writers.** Only inside this window: + `TRUNCATE post_submissions` and deploy the fingerprint retype to + `PostV2Record` (a live-writer retype strands in-flight dedupe reservations — + the retype is deploy-sequenced, never merge-coupled; it also ripples through + `enhanceExternalEmbed`/`postV2From`, budget for that). +3. **Credential census FIRST (gates everything).** The tool enumerates authors + whose repo credentials cannot be restored. The fallback for such a post is + to **leave it as `community.post`** (ledger terminal `fallback_left_legacy`), + logged, never deleted — and NEVER an admin-forged postv2 (forging authorship + reintroduces the §2 impersonation liability the whole flip removes). The run + refuses to report "complete" while any fallback row survives. +4. **Run the tool** (resumable, verify-before-delete, idempotent). Per record: + postv2 into the author repo (deterministic rkey = pure function of the OLD + record, NOT `SubmissionRkey`) → acceptance via `WriteAcceptance` DIRECT (NOT + the engine — the engine re-decides and could reject a since-banned author's + live post, rewriting history) pinning the NEW CID → verify both + CID → + ledger `migrated` → delete old → `done`. `migrated` and `done` are distinct + states so a crash after checkpoint retries only the delete. +5. **Truncate + firehose re-index `posts`** (§5.1). The tool writes repo records + + ledger only; the consumer re-indexes the fresh postv2/acceptance. +6. **POST-DRAIN FOLLOW-UP (separate branch, gated on + `SELECT count(*) FROM posts WHERE split_part(uri,'/',4)='social.coves.community.post'` + = 0 AND fallbacks = 0):** only then remove the legacy surfaces — + `consumerWantedCollections` community.post entry, the post_consumer legacy + branch, `visiblePostsJoin`'s NULL-non-postv2-visible branch, `blobOwnerOf` + community fallback, `removedMarkers` legacy handling + record author-field + synthesis, `legacyPostCollection`, `applyRemoval`'s absent collection guard. + These retire together; removing ingest ahead of read (or either ahead of the + drain) is the vanish/relaunder bug. + +**Deferred, independent of the cutover (file as issues):** orphan +community-admissions sweep on community deletion; `community.post_count` +incrementer wiring onto `countAcceptedPostsForCommunity` (cosmetic — display +already excludes non-accepted; no leak). Neither blocks the merge. + +**Merge hygiene:** never run `go mod tidy` (upgrades transitives into a broken +go-log; go-car is pinned indirect — `make ci` uses `mod download`, safe). diff --git a/internal/core/posts/rematerialize.go b/internal/core/posts/rematerialize.go new file mode 100644 --- /dev/null +++ b/internal/core/posts/rematerialize.go @@ -0,0 +1,241 @@ +package posts + +import ( + "context" + "errors" + "time" +) + +// The re-materialization tool: the cutover step that moves every legacy +// social.coves.community.post record (written into a COMMUNITY's repo under the +// community's credentials) to an author-owned social.coves.community.postv2 in +// the AUTHOR's repo, plus the community's acceptance that pins it +// (docs/PRD_AUTHOR_OWNED_POSTS.md §10.1 step 5 and §11 the rev-2.8 deploy +// runbook). +// +// # THIS FILE IS A RED STUB +// +// Every exported symbol here exists so tests/e2e/rematerialize_contract_test.go +// and the T1 state-machine tests compile and FAIL for the right reason. The +// bodies return the not-implemented sentinel; GREEN fills them in. Nothing in +// this file decides anything yet. +// +// # WHY THE LOGIC LIVES IN PACKAGE posts (not a subpackage) +// +// The tool's whole safety argument is that it REUSES the write path's two +// idempotent primitives rather than re-deriving them: +// +// - createAuthorRecord (service.go) — the converge-by-read create: a re-run at +// the same rkey meets its own first attempt (ErrSwapConflict / ErrNoCommit) +// and reports the standing record instead of minting a second post. This is +// what makes a crash-resumed run a no-op. +// - communityRecordWriter.WriteAcceptance (community_writer.go) — the +// deterministic-rkey acceptance write that SKIPS when the standing record +// already pins the target CID, so a re-run emits no new commit and no new +// CID. +// +// Both are package-private. A subpackage would have to either export them (widen +// the write path's surface for one caller) or reimplement them (the exact +// divergence the design forbids — a second rkey scheme mints duplicates). The +// cost is that package posts grows; the benefit is that the tool cannot drift +// from the write path it is draining into. cmd/rematerialize-posts/main.go stays +// a thin wrapper that only wires the seams below. + +// RematerializeState is one legacy record's position in the ledger state machine +// (migration 037). The happy path is discovered → postv2_written → verified → +// migrated → done; the two fallback states are terminal. +type RematerializeState string + +const ( + // RematerializeDiscovered: the ledger row exists for this old URI and + // nothing has been written yet. + RematerializeDiscovered RematerializeState = "discovered" + + // RematerializePostV2Written: the postv2 record stands in the AUTHOR's repo + // at the deterministic rkey. Its URI/CID/rkey are on the ledger row. + RematerializePostV2Written RematerializeState = "postv2_written" + + // RematerializeVerified: the acceptance stands in the COMMUNITY's repo and + // both records have been read back and confirmed to pin the same CID. + RematerializeVerified RematerializeState = "verified" + + // RematerializeMigrated is the CHECKPOINT BEFORE DELETE, distinct from done + // on purpose: it means "verified safe to delete, and the OLD record is still + // present". A crash after this checkpoint resumes by retrying ONLY the delete + // (§11 step 4). + RematerializeMigrated RematerializeState = "migrated" + + // RematerializeDone: the old community.post record has been deleted. + RematerializeDone RematerializeState = "done" + + // RematerializeFallbackLeftLegacy is the terminal state for a record whose + // author credentials could not be restored: the postv2 is NOT written, the + // old record is NOT deleted, the row is logged. §11 step 3 is explicit that + // the fallback NEVER forges authorship (no admin-signed postv2), because + // forging reintroduces the exact §2 impersonation liability the whole flip + // removes. + RematerializeFallbackLeftLegacy RematerializeState = "fallback_left_legacy" + + // RematerializeFallbackNoCreds is reserved terminal vocabulary for + // distinguishing "the author-repo factory reported no credentials" from a + // human-operator-flagged leave-as-legacy. Cycle 1 pins only the behaviour + // (a no-creds author is left legacy and the run is not complete); which of + // the two strings a given cause writes is a GREEN/owner decision — see the + // ambiguity flag in the RED report. + RematerializeFallbackNoCreds RematerializeState = "fallback_no_creds" +) + +// IsFallback reports whether a state is one of the terminal fallback states, so +// the census can gate "complete" on any of them surviving without enumerating +// each string at every call site. +func IsFallback(state RematerializeState) bool { + return state == RematerializeFallbackLeftLegacy || state == RematerializeFallbackNoCreds +} + +// LegacyPost is one deprecated social.coves.community.post record discovered for +// migration. +type LegacyPost struct { + // URI is the OLD record's AT-URI — at:///social.coves.community.post/. + // It is the ledger's primary key and the material the deterministic postv2 + // rkey is derived from. + URI string + + // CID is the old record's content CID, carried for audit only. Authorship + // and content come from the author's repo once the postv2 is written; the + // old CID is never pinned by the new acceptance. + CID string + + // CommunityDID is the repo the old record lived in — the community whose + // acceptance the tool writes. + CommunityDID string + + // AuthorDID is the record's `author` field: the DID whose repo the postv2 is + // 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 PostRecord +} + +// LegacySource enumerates and deletes the deprecated community.post records. +// +// It is a seam so the T1 state machine runs against an in-memory source while +// the T2 contract and production run it against real community repos on the PDS +// (listRecords over social.coves.community.post, delete via the community's own +// credentials). The DELETE is idempotent by contract — a delete of an +// already-gone record reports success — because it is the resumed step a crash +// after the migrated checkpoint retries. +type LegacySource interface { + ListLegacyPosts(ctx context.Context) ([]LegacyPost, error) + DeleteLegacyPost(ctx context.Context, legacy LegacyPost) error +} + +// RematerializeLedgerRow is one row of the migration-037 ledger. +type RematerializeLedgerRow struct { + OldURI string + State RematerializeState + AuthorDID string + + // NewURI, NewCID, NewRkey identify the postv2 the tool wrote. Populated at + // the postv2_written transition and never recomputed on resume — the resumed + // run reads them back rather than deriving a fresh CID. + NewURI string + NewCID string + NewRkey string + + // Reason is the human-readable note attached to a fallback row. + Reason string + + CreatedAt time.Time + UpdatedAt time.Time +} + +// RematerializeLedger is the migration-037 Postgres table, behind an interface +// so the state-machine tests drive a real ledger while the source and repos are +// faked. +type RematerializeLedger interface { + // Discover upserts the row for oldURI in state discovered, idempotently: a + // re-run finds the existing row (whatever state it is in) rather than + // resetting it. + 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) + + // RecordPostV2Written moves discovered → postv2_written and records the + // postv2 coordinates. + RecordPostV2Written(ctx context.Context, oldURI, newURI, newCID, newRkey string) error + + // MarkVerified moves postv2_written → verified. + MarkVerified(ctx context.Context, oldURI string) error + + // MarkMigrated moves verified → migrated — the checkpoint before delete. + MarkMigrated(ctx context.Context, oldURI string) error + + // MarkDone moves migrated → done, after the old record is deleted. + MarkDone(ctx context.Context, oldURI string) error + + // MarkFallback moves the row to a terminal fallback state with a reason. + MarkFallback(ctx context.Context, oldURI string, state RematerializeState, reason string) error + + // CountByState is the census: how many rows sit in each state, so the run + // can refuse "complete" while any fallback survives. + CountByState(ctx context.Context) (map[RematerializeState]int, error) +} + +// RematerializeReport is the census a run returns. +type RematerializeReport struct { + Discovered int + Done int + Fallbacks int + ByState map[RematerializeState]int + + // Complete is false while any fallback row survives — the gate on the + // separate, manual legacy-removal follow-up (§11 step 6). + Complete bool +} + +// Rematerializer drives the cutover. +// +// It holds a CommunityRecordWriter — the DIRECT acceptance writer — and NOT an +// AcceptanceEngine or an AdmissionDecider, by design: the engine re-decides +// admission, which for a since-banned author would REJECT a post that is live in +// production today, rewriting history (§11 step 4). The tool preserves the +// existing acceptance; it never re-adjudicates one. +type Rematerializer struct { + Source LegacySource + Ledger RematerializeLedger + AuthorRepos AuthorRepoFactory + Acceptances CommunityRecordWriter +} + +// errRematerializeNotImplemented is the RED sentinel every stub body returns. +var errRematerializeNotImplemented = errors.New("rematerialize: not implemented (RED stub)") + +// RematerializeRkey is the postv2 record key the tool writes a legacy record at. +// +// IT IS A PURE, STABLE FUNCTION OF THE OLD RECORD'S URI — the single +// highest-risk detail in the tool (§11 step 4). A re-run must recompute the +// IDENTICAL key so createAuthorRecord converges by read instead of minting a +// second postv2. It therefore CANNOT be SubmissionRkey, which needs the +// submission-time fingerprint and dedupe bucket the migration does not have — a +// re-run would draw a different key and duplicate the post. +// +// The scheme is the SubjectRkey digest scheme applied to the OLD URI: a total, +// collision-free (SHA-256) function that the write path already trusts. +func RematerializeRkey(legacyPostURI string) string { return "" } + +// Run discovers every legacy record and drives each to a terminal state, +// returning the census. +func (r *Rematerializer) Run(ctx context.Context) (RematerializeReport, error) { + return RematerializeReport{}, errRematerializeNotImplemented +} + +// RematerializeOne drives a single legacy record from wherever its ledger row +// stands to a terminal state, and returns the state it reached. +func (r *Rematerializer) RematerializeOne(ctx context.Context, legacy LegacyPost) (RematerializeState, error) { + return "", errRematerializeNotImplemented +} diff --git a/internal/core/posts/rematerialize_outer_test.go b/internal/core/posts/rematerialize_outer_test.go new file mode 100644 --- /dev/null +++ b/internal/core/posts/rematerialize_outer_test.go @@ -0,0 +1,184 @@ +//go:build integration + +package posts_test + +import ( + "context" + "strings" + "testing" + "time" + + "Coves/internal/atproto/pds" + "Coves/internal/core/posts" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + "github.com/bluesky-social/indigo/atproto/auth/oauth" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The re-materialization OUTER contract, against a REAL PDS and a REAL ledger: +// the whole tool driven end to end over records a real PDS minted the CIDs for +// (docs/PRD_AUTHOR_OWNED_POSTS.md §11). +// +// # WHY THIS IS T1 AND NOT T2 (the tier the task named) +// +// The plan asked for this outer proof in tests/e2e, driving the tool's package +// in-process. It cannot live there. The e2e package's own constitution +// (tests/e2e/contracts_test.go, "NEVER open a testkit.DB clone to assert on +// AppView state", and community_contract_test.go, "writing to the AppView's own +// database … the one thing the package doc forbids") bars the DB access the tool +// structurally requires: its ledger IS a table in that database. A prior test +// (error_recovery_test.go) was DELETED for exactly the DB-touching this would +// reintroduce. So the real-infra outer proof belongs where real-infra behaviour +// is sanctioned — the integration tier, over a real PDS and a real Postgres — and +// this file is that proof. The pipeline leg the task also wanted (postv2 → +// pending → accepted via getStatus) is already proven for ANY producer of +// postv2/acceptance records by tests/e2e/author_post_contract_test.go, and the +// tool is just such a producer; re-asserting it here would duplicate that +// coverage while breaking the tier it lives in. See the RED report's scope flag. +// +// What only a REAL PDS can prove, and what this file is therefore for: +// +// - createAuthorRecord's converge-by-read against the PDS's ACTUAL create-only +// guard: a re-run at the deterministic rkey meets ErrSwapConflict / ErrNoCommit +// and reads the standing postv2 back, minting no second record. +// - WriteAcceptance's skip against a REAL standing record: a re-run pins no new +// CID. +// - the CID the acceptance pins is one the PDS minted for the postv2, so a +// strongRef that fails to round-trip fails here rather than in production. +// - the old community.post is REALLY gone from the community repo afterward. + +// realLegacySource lists the staged legacy records and deletes them from the real +// community repo. The delete is idempotent — a not-found is success — because it +// is the step a crash after the migrated checkpoint retries. +type realLegacySource struct { + community pds.Client + staged []posts.LegacyPost +} + +func (s *realLegacySource) ListLegacyPosts(_ context.Context) ([]posts.LegacyPost, error) { + return s.staged, nil +} + +func (s *realLegacySource) DeleteLegacyPost(ctx context.Context, legacy posts.LegacyPost) error { + rkey := legacy.URI[strings.LastIndex(legacy.URI, "/")+1:] + err := s.community.DeleteRecord(ctx, postCollection, rkey) + if err != nil && testkit.IsNotFound(err) { + return nil + } + return err +} + +func TestRematerialize_OuterContract_RealPDS_MovesPostAndIsIdempotent(t *testing.T) { + t.Parallel() + + pdsServer := testkit.NewPDS(t) + communityAcct := pdsServer.CreateAccount(t, testkit.WithHandlePrefix("remc")) + authorAcct := pdsServer.CreateAccount(t, testkit.WithHandlePrefix("rema")) + ctx := context.Background() + + // Seed a REAL deprecated community.post into the community's repo, signed with + // the community's own credentials — exactly how the pre-flip write path put it + // there, `author` field and all. + oldRkey := testkit.TID() + title := "legacy " + testkit.UniqueID(t) + seeded := communityAcct.PutRecord(t, postCollection, oldRkey, 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", + }) + + 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), + Content: strPtr("words the author is accountable for"), + CreatedAt: "2026-01-02T03:04:05Z", + }, + } + + // Real author-repo credentials for the author, over the real PDS. + authorFactory := func(_ context.Context, authorDID string, _ *oauth.ClientSessionData) (posts.AuthorRepo, error) { + require.Equalf(t, authorAcct.DID, authorDID, "the tool asked for a repo other than the post's author") + generic, err := pds.NewFromAccessToken(pdsServer.URL(), authorAcct.DID, authorAcct.AccessToken) + require.NoError(t, err) + repo, ok := generic.(posts.AuthorRepo) + require.True(t, ok, "the PDS client must implement the author-repo write surface") + return repo, nil + } + + // Real community-repo credentials, and the DIRECT acceptance writer over them. + communityGeneric, err := pds.NewFromAccessToken(pdsServer.URL(), communityAcct.DID, communityAcct.AccessToken) + require.NoError(t, err) + communityRepo, ok := communityGeneric.(posts.CommunityRepo) + require.True(t, ok, "the PDS client must implement the community-repo write surface") + 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} + + // ---- run ----------------------------------------------------------------- + state, err := tool.RematerializeOne(ctx, legacy) + require.NoError(t, err) + require.Equal(t, posts.RematerializeDone, state) + + wantRkey := posts.RematerializeRkey(legacy.URI) + newURI := "at://" + authorAcct.DID + "/" + postv2Collection + "/" + wantRkey + + // The postv2 stands in the AUTHOR's repo, with NO author field (authorship is + // the repo now), carrying the original community and title. + postV2 := authorAcct.GetRecord(t, postv2Collection, wantRkey) + newCID := postV2.CID + assert.Equalf(t, communityAcct.DID, postV2.Value["community"], "the postv2 must keep its original community") + assert.Equal(t, title, postV2.Value["title"]) + _, hasAuthor := postV2.Value["author"] + assert.Falsef(t, hasAuthor, "the re-materialized postv2 must NOT carry an author field — the repo signature is the authorship anchor") + + // The community's acceptance stands, pinning the NEW postv2 URI and the CID the + // PDS minted for it. + acceptance := communityAcct.GetRecord(t, posts.AcceptanceCollection, posts.SubjectRkey(newURI)) + assertSubject(t, acceptance, newURI, newCID) + + // The old community.post is REALLY gone. + require.Truef(t, testkit.IsNotFound(getRecordErr(ctx, communityAcct, postCollection, oldRkey)), + "the old community.post must be deleted from the community repo once the postv2 and acceptance are verified") + + row, found, err := ledger.Get(ctx, legacy.URI) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, posts.RematerializeDone, row.State) + assert.Equal(t, newURI, row.NewURI) + assert.Equal(t, newCID, row.NewCID) + + // ---- re-run: pure no-op -------------------------------------------------- + stateAgain, err := tool.RematerializeOne(ctx, legacy) + require.NoError(t, err) + assert.Equal(t, posts.RematerializeDone, stateAgain) + + postV2Again := authorAcct.GetRecord(t, postv2Collection, wantRkey) + assert.Equalf(t, newCID, postV2Again.CID, + "a re-run minted a new postv2 CID; the deterministic rkey must converge on the first record via createAuthorRecord's read-back") + + acceptanceAgain := communityAcct.GetRecord(t, posts.AcceptanceCollection, posts.SubjectRkey(newURI)) + assertSubject(t, acceptanceAgain, newURI, newCID) + assert.Equalf(t, acceptance.CID, acceptanceAgain.CID, + "a re-run rewrote the acceptance record; WriteAcceptance must SKIP when the standing record already pins the target CID") + + require.Truef(t, testkit.IsNotFound(getRecordErr(ctx, communityAcct, postCollection, oldRkey)), + "the old community.post must stay gone across a re-run") +} diff --git a/internal/core/posts/rematerialize_rkey_test.go b/internal/core/posts/rematerialize_rkey_test.go new file mode 100644 --- /dev/null +++ b/internal/core/posts/rematerialize_rkey_test.go @@ -0,0 +1,99 @@ +package posts + +import ( + "crypto/sha256" + "encoding/base32" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// 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: +// +// 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. +// +// 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[:])) +} + +func TestRematerializeRkey_IsTheDigestOfTheOldURI(t *testing.T) { + oldURI := "at://did:plc:community2222222222222222/social.coves.community.post/3kqijkl2m4c2r" + + got := 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") +} + +func TestRematerializeRkey_IsDeterministic(t *testing.T) { + oldURI := "at://did:plc:community2222222222222222/social.coves.community.post/3kqijkl2m4c2r" + + 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") +} + +func TestRematerializeRkey_DependsOnlyOnTheOldURI(t *testing.T) { + // Two DIFFERENT old URIs must give two different keys (no accidental + // constant), and the SAME old URI must give the SAME key no matter what else + // is true of the run — there is no fingerprint, bucket, or clock in the + // material, by construction. + a := "at://did:plc:community2222222222222222/social.coves.community.post/3kqijkl2m4c2r" + b := "at://did:plc:community2222222222222222/social.coves.community.post/3kqijkl2m4c2s" + + assert.NotEqualf(t, RematerializeRkey(a), RematerializeRkey(b), + "two distinct legacy records must re-materialize to two distinct postv2 rkeys, or one would silently overwrite the other in the author's repo") +} + +func TestRematerializeRkey_IsNotSubmissionRkey(t *testing.T) { + // SubmissionRkey needs (community, fingerprint, bucket, window) — submission- + // time material the migration does not have. The re-materialization key is a + // function of the OLD URI ALONE. They must not coincide: if the tool's key + // happened to equal a SubmissionRkey, a later genuine submission of the same + // 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), + "the re-materialization key must be independent of SubmissionRkey — it is derived from the OLD record's URI, not the submission fingerprint the migration lacks") +} diff --git a/internal/core/posts/rematerialize_test.go b/internal/core/posts/rematerialize_test.go new file mode 100644 --- /dev/null +++ b/internal/core/posts/rematerialize_test.go @@ -0,0 +1,584 @@ +//go:build integration + +package posts_test + +import ( + "context" + "fmt" + "sync" + "testing" + + "Coves/internal/atproto/pds" + "Coves/internal/core/blobs" + "Coves/internal/core/posts" + "Coves/internal/db/postgres" + "Coves/tests/testkit" + + "github.com/bluesky-social/indigo/atproto/auth/oauth" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// 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). +// +// # WHAT THIS TIER PROVES, AND WHY THE REPOS ARE FAKED +// +// The tool's three load-bearing safety properties are all ORDERING and +// IDEMPOTENCE properties — they live in the orchestration, not in the PDS: +// +// 1. rkey stability (pinned purely at T0 in rematerialize_rkey_test.go) — a +// re-run computes the SAME postv2 key and createAuthorRecord converges by +// read instead of minting a second post. +// 2. VERIFY BEFORE DELETE — the old community.post record is deleted only after +// the postv2 AND its acceptance are confirmed to pin the same CID. Delete +// one instant too early and a crash loses a live post. +// 3. THE FALLBACK NEVER FORGES — a post whose author credentials cannot be +// restored is left as legacy, never re-authored under a forged signature, +// and the run refuses to report "complete" while any such row survives. +// +// The ledger is real because resume reads it; the repos are faked because the +// faults these tests inject — a delete that fails once, a postv2 that changed CID +// under the verify, an author with no credentials — are precisely the ones a real +// PDS will not produce on demand. The write PRIMITIVES the tool reuses +// (createAuthorRecord's converge-by-read, WriteAcceptance's skip) have their own +// real-PDS coverage; the outer real-stack proof is tests/e2e/rematerialize_ +// contract_test.go. +// +// # THE TOOL DOES NOT RE-DECIDE (mirrors the scriptedDecider trick) +// +// The Rematerializer holds a CommunityRecordWriter — the DIRECT acceptance writer +// — and NOT an AcceptanceEngine or an AdmissionDecider. That is the whole reason +// the acceptance is written by WriteAcceptance here: routing through the engine +// would re-run admission, and a since-banned author's LIVE production post would +// be REJECTED, rewriting history (§11 step 4). service_writeforward_test.go pins +// the same "no re-decision" property on the fast path with a scriptedDecider that +// would refuse; here the property is structural — there is no decider to consult +// — and TestRematerialize_UsesDirectAcceptanceWriter_NeverReDecides makes it a +// behaviour. + +// ---- fakes --------------------------------------------------------------- + +func strPtr(s string) *string { return &s } + +// deterministicCID is the CID a fake author repo assigns a postv2 at a given +// rkey: stable across calls, so a re-run's converge-by-read reads the SAME CID +// back rather than a fresh one. Real PDS CIDs are content-addressed and share +// this property for identical bytes. +func deterministicCID(rkey string) string { return "bafyreipostv2" + rkey } + +// fakeAuthorRepo models the one PDS behaviour the tool's idempotence rests on: +// a create-only put (swapRecord "") of a record that already stands is refused +// with ErrSwapConflict, and the standing record is read back instead. +type fakeAuthorRepo struct { + did string + + 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) +} + +func newFakeAuthorRepo(did string) *fakeAuthorRepo { + return &fakeAuthorRepo{did: did, records: map[string]*pds.RecordResponse{}, getCIDAt: map[string]string{}} +} + +func (r *fakeAuthorRepo) recordCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.records) +} + +func (r *fakeAuthorRepo) GetRecord(_ context.Context, collection, rkey string) (*pds.RecordResponse, error) { + r.mu.Lock() + defer r.mu.Unlock() + rec, ok := r.records[rkey] + if !ok { + return nil, pds.ErrNotFound + } + if override, ok := r.getCIDAt[rkey]; ok { + clone := *rec + clone.CID = override + return &clone, nil + } + return rec, nil +} + +func (r *fakeAuthorRepo) PutRecordWithCommit(_ context.Context, collection, rkey string, record any, swapRecord string) (*pds.RecordCommit, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if r.putErr != nil { + err := r.putErr + r.putErr = nil + return nil, err + } + + if _, exists := r.records[rkey]; exists && swapRecord == "" { + // The create-only guard: the record is already here. This is exactly what + // a re-run of a re-materialized post meets, and createAuthorRecord answers + // it by reading the standing record back rather than minting a second. + return nil, pds.ErrSwapConflict + } + + uri := "at://" + r.did + "/" + collection + "/" + rkey + cid := deterministicCID(rkey) + body, _ := record.(map[string]any) + r.records[rkey] = &pds.RecordResponse{URI: uri, CID: cid, Value: body} + return &pds.RecordCommit{URI: uri, CID: cid, CommitRev: "3krematputxxx"}, nil +} + +func (r *fakeAuthorRepo) DeleteRecord(_ context.Context, collection, rkey string) error { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.records, rkey) + return nil +} + +func (r *fakeAuthorRepo) UploadBlob(_ context.Context, data []byte, mimeType string) (*blobs.BlobRef, error) { + return nil, fmt.Errorf("fakeAuthorRepo.UploadBlob: not used by re-materialization") +} + +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". +func (r *fakeAuthorRepo) seedStanding(collection, rkey string) { + 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{}} +} + +// 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 +} + +func newFakeAuthorFactory() *fakeAuthorFactory { + return &fakeAuthorFactory{repos: map[string]*fakeAuthorRepo{}, noCreds: map[string]bool{}} +} + +func (f *fakeAuthorFactory) repo(did string) *fakeAuthorRepo { + if r, ok := f.repos[did]; ok { + return r + } + r := newFakeAuthorRepo(did) + f.repos[did] = r + return r +} + +func (f *fakeAuthorFactory) factory() posts.AuthorRepoFactory { + return func(_ context.Context, authorDID string, _ *oauth.ClientSessionData) (posts.AuthorRepo, error) { + if f.noCreds[authorDID] { + return nil, fmt.Errorf("resuming the stored session of %s: %w", authorDID, posts.ErrNoAuthorCredentials) + } + r, ok := f.repos[authorDID] + if !ok { + return nil, fmt.Errorf("opening the repository of %s: %w", authorDID, posts.ErrNoAuthorCredentials) + } + return r, nil + } +} + +// spyAcceptanceWriter records every WriteAcceptance and returns a standing +// acceptance that pins whatever CID it was handed. A repeat call reports Skipped, +// the way the real writer does when the acceptance already pins the target CID — +// so a re-run mints no new CID. +type spyAcceptanceWriter struct { + mu sync.Mutex + acceptanceCmds []posts.CommunityWriteCommand + writeErr error // one-shot injected failure + otherCalled []string +} + +func (s *spyAcceptanceWriter) WriteAcceptance(_ context.Context, cmd posts.CommunityWriteCommand) (posts.CommunityWriteResult, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.writeErr != nil { + err := s.writeErr + s.writeErr = nil + return posts.CommunityWriteResult{}, err + } + + repeat := false + for _, prior := range s.acceptanceCmds { + if prior.PostURI == cmd.PostURI && prior.PostCID == cmd.PostCID { + repeat = true + } + } + s.acceptanceCmds = append(s.acceptanceCmds, cmd) + + rkey := posts.SubjectRkey(cmd.PostURI) + return posts.CommunityWriteResult{ + URI: "at://" + cmd.CommunityDID + "/social.coves.community.acceptance/" + rkey, + RKey: rkey, + CID: "bafyreiacceptance" + rkey, + Rev: "3krematacceptxx", + Skipped: repeat, + }, nil +} + +// The other four methods exist only to satisfy CommunityRecordWriter. The tool +// must never call them: a re-materialized post is accepted, not removed, +// restored, repinned, or withdrawn. Each records that it was reached so a test +// can prove it was not. +func (s *spyAcceptanceWriter) WriteRemoval(_ context.Context, _ posts.CommunityRemovalCommand) (posts.CommunityWriteResult, error) { + s.note("WriteRemoval") + return posts.CommunityWriteResult{}, nil +} +func (s *spyAcceptanceWriter) RestoreAcceptance(_ context.Context, _ posts.CommunityWriteCommand) (posts.CommunityWriteResult, error) { + s.note("RestoreAcceptance") + return posts.CommunityWriteResult{}, nil +} +func (s *spyAcceptanceWriter) RepinAcceptance(_ context.Context, _ posts.CommunityWriteCommand) (posts.CommunityWriteResult, error) { + s.note("RepinAcceptance") + return posts.CommunityWriteResult{}, nil +} +func (s *spyAcceptanceWriter) DeleteAcceptance(_ context.Context, _ posts.CommunityAcceptanceDeleteCommand) (posts.CommunityWriteResult, error) { + s.note("DeleteAcceptance") + return posts.CommunityWriteResult{}, nil +} +func (s *spyAcceptanceWriter) note(method string) { + s.mu.Lock() + defer s.mu.Unlock() + s.otherCalled = append(s.otherCalled, method) +} +func (s *spyAcceptanceWriter) calls() []posts.CommunityWriteCommand { + s.mu.Lock() + defer s.mu.Unlock() + return append([]posts.CommunityWriteCommand(nil), s.acceptanceCmds...) +} + +// fakeLegacySource yields staged legacy records and records deletes. It can fail +// the FIRST delete of a given URI once, so a crash-after-checkpoint is +// reproducible, and it treats a delete of an already-gone record as success — +// the idempotence the resumed delete leans on. +type fakeLegacySource struct { + mu sync.Mutex + posts []posts.LegacyPost + deleted map[string]int + deleteErr map[string]error +} + +func newFakeLegacySource(ps ...posts.LegacyPost) *fakeLegacySource { + return &fakeLegacySource{posts: ps, deleted: map[string]int{}, deleteErr: map[string]error{}} +} + +func (s *fakeLegacySource) ListLegacyPosts(_ context.Context) ([]posts.LegacyPost, error) { + s.mu.Lock() + defer s.mu.Unlock() + return append([]posts.LegacyPost(nil), s.posts...), nil +} + +func (s *fakeLegacySource) DeleteLegacyPost(_ context.Context, legacy posts.LegacyPost) error { + s.mu.Lock() + defer s.mu.Unlock() + s.deleted[legacy.URI]++ + if err, ok := s.deleteErr[legacy.URI]; ok { + delete(s.deleteErr, legacy.URI) + return err + } + return nil +} + +func (s *fakeLegacySource) deleteCount(uri string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.deleted[uri] +} + +// legacyPost stages one deprecated community.post, keyed by a unique rkey so +// parallel tests never collide on the ledger's old_uri primary key. +func legacyPost(t *testing.T, communityDID, authorDID string) posts.LegacyPost { + t.Helper() + rkey := testkit.TID() + oldURI := "at://" + communityDID + "/social.coves.community.post/" + rkey + title := "legacy " + testkit.UniqueID(t) + return posts.LegacyPost{ + URI: oldURI, + CID: "bafyreilegacy" + rkey, + CommunityDID: communityDID, + AuthorDID: authorDID, + Record: posts.PostRecord{ + Type: "social.coves.community.post", + Community: communityDID, + Author: authorDID, + Title: strPtr(title), + Content: strPtr("words the author is accountable for"), + CreatedAt: "2026-01-02T03:04:05Z", + }, + } +} + +const ( + rematCommunityDID = "did:plc:cccccccccccccccccccccccc" + rematAuthorDID = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" +) + +// ---- tests --------------------------------------------------------------- + +func TestRematerialize_HappyPath_WalksToDoneVerifyBeforeDelete(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + 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} + + state, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err) + require.Equalf(t, posts.RematerializeDone, state, "a fresh legacy record must walk all the way to done") + + // The postv2 was written into the AUTHOR's repo at the deterministic rkey. + authorRepo := authors.repo(rematAuthorDID) + wantRkey := posts.RematerializeRkey(legacy.URI) + newURI := "at://" + rematAuthorDID + "/social.coves.community.postv2/" + wantRkey + assert.Equalf(t, 1, authorRepo.recordCount(), "exactly one postv2 must have been written") + + // The acceptance was written DIRECT, into the community's repo, pinning the + // NEW postv2 CID — not the old community.post CID. + calls := writer.calls() + require.Lenf(t, calls, 1, "exactly one acceptance must have been written") + assert.Equal(t, rematCommunityDID, calls[0].CommunityDID) + assert.Equalf(t, newURI, calls[0].PostURI, "the acceptance must pin the NEW postv2 URI in the author's repo") + assert.Equalf(t, deterministicCID(wantRkey), calls[0].PostCID, + "the acceptance must pin the NEW postv2 CID; pinning the old community.post CID would attest to content that no longer exists") + assert.Emptyf(t, writer.otherCalled, "a re-materialized post is accepted only — no removal/restore/repin/withdraw") + + // The old record was deleted — and the ledger proves the delete came AFTER the + // migrated checkpoint (the row is done, which is only reachable through it). + assert.Equalf(t, 1, source.deleteCount(legacy.URI), "the old community.post must be deleted exactly once") + + row, found, err := ledger.Get(context.Background(), legacy.URI) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, posts.RematerializeDone, row.State) + assert.Equalf(t, newURI, row.NewURI, "the ledger must record the postv2 URI it wrote") + assert.Equalf(t, deterministicCID(wantRkey), row.NewCID, "the ledger must record the postv2 CID it pinned") + assert.Equal(t, wantRkey, row.NewRkey) +} + +func TestRematerialize_ReRun_IsAPureNoOp(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + 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} + + first, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err) + require.Equal(t, posts.RematerializeDone, first) + + // Run it again against a record already fully migrated. Nothing new may + // happen: the same rkey converges (no second postv2), the acceptance skips + // (no new CID), the old record is already gone. + second, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err) + assert.Equal(t, posts.RematerializeDone, second) + + assert.Equalf(t, 1, authors.repo(rematAuthorDID).recordCount(), + "a re-run must not mint a second postv2 — the deterministic rkey converges on the first record") + + calls := writer.calls() + require.GreaterOrEqual(t, len(calls), 1) + assert.Truef(t, calls[len(calls)-1].PostCID == deterministicCID(posts.RematerializeRkey(legacy.URI)), + "a re-run's acceptance must still pin the same CID; a fresh CID would dangle every reference to the acceptance") + // The old record was already gone; a re-run's delete (if attempted) is a no-op + // success, never an error, and never resurrects the record. + row, _, err := ledger.Get(context.Background(), legacy.URI) + require.NoError(t, err) + assert.Equal(t, posts.RematerializeDone, row.State) +} + +func TestRematerialize_ResumeAfterDeleteFailure_RetriesOnlyTheDelete(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + authors.repo(rematAuthorDID) + writer := &spyAcceptanceWriter{} + legacy := legacyPost(t, rematCommunityDID, rematAuthorDID) + source := newFakeLegacySource(legacy) + source.deleteErr[legacy.URI] = fmt.Errorf("transient: the community PDS returned 502 on delete") + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + // First pass: everything succeeds up to the delete, which fails once. The row + // must stop at migrated — the checkpoint BEFORE the delete — never done. + _, err := tool.RematerializeOne(context.Background(), legacy) + require.Errorf(t, err, "a failed delete must surface as an error, not be swallowed into a false 'done'") + + row, found, err := ledger.Get(context.Background(), legacy.URI) + require.NoError(t, err) + require.True(t, found) + require.Equalf(t, posts.RematerializeMigrated, row.State, + "a crash on the delete must leave the row at the migrated checkpoint: postv2 and acceptance verified, old record still present") + require.Equalf(t, 1, source.deleteCount(legacy.URI), "the delete was attempted once and failed") + + postV2Before := authors.repo(rematAuthorDID).recordCount() + acceptancesBefore := len(writer.calls()) + + // Resume. Only the delete should do new work; the postv2 must not be rewritten + // and no new acceptance CID may be minted. + state, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err) + assert.Equal(t, posts.RematerializeDone, state) + + assert.Equalf(t, postV2Before, authors.repo(rematAuthorDID).recordCount(), + "resume rewrote the postv2; the migrated checkpoint exists so a crash retries ONLY the delete") + for _, c := range writer.calls()[acceptancesBefore:] { + assert.Equalf(t, deterministicCID(posts.RematerializeRkey(legacy.URI)), c.PostCID, + "resume minted a new acceptance CID; a re-fire must converge on the same pinned CID") + } + assert.Equalf(t, 2, source.deleteCount(legacy.URI), "resume must retry the delete that failed") +} + +func TestRematerialize_CIDMismatch_DoesNotCheckpointOrDelete(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} + + // The verify re-read of the postv2 comes back with a DIFFERENT CID than the + // one the acceptance pinned — a concurrent edit landing in the write→verify + // window. Verification must fail, and the tool must NOT checkpoint and must + // NOT delete: deleting here would destroy the only copy of a post whose new + // acceptance points at content that no longer stands. + wantRkey := posts.RematerializeRkey(legacy.URI) + authorRepo.getCIDAt[wantRkey] = "bafyreianeditlandedmidverify" + + _, err := tool.RematerializeOne(context.Background(), legacy) + require.Errorf(t, err, "a CID mismatch at verify must be an error, not a silent success") + + assert.Equalf(t, 0, source.deleteCount(legacy.URI), + "VERIFY BEFORE DELETE: the old record must not be deleted when the postv2 CID no longer matches the acceptance") + + 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 mismatched record must never reach done") + assert.NotEqualf(t, posts.RematerializeMigrated, row.State, + "a mismatched record must never reach the migrated checkpoint — migrated asserts the delete is safe, and it is not") +} + +func TestRematerialize_NoCredentials_LeavesLegacyNeverForges(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + writer := &spyAcceptanceWriter{} + + // A human author whose repo credentials cannot be restored non-interactively + // (there is no stored session to resume — see the RED report's credential + // finding). The tool must leave the post as legacy, write NOTHING, and delete + // NOTHING — never an admin-forged postv2, which would reintroduce the §2 + // impersonation liability the whole flip exists to remove. + humanDID := "did:plc:humanhumanhumanhumanhum" + authors.noCreds[humanDID] = true + legacy := legacyPost(t, rematCommunityDID, humanDID) + source := newFakeLegacySource(legacy) + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + state, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err, "a no-creds record is an expected terminal outcome, not a run-failing error") + assert.Equalf(t, posts.RematerializeFallbackLeftLegacy, state, + "an author with no restorable credentials must land in fallback_left_legacy") + assert.Truef(t, posts.IsFallback(state), "the terminal state must be a fallback state the census can gate on") + + assert.Emptyf(t, writer.calls(), "no acceptance may be written for a post that was never re-authored") + assert.Equalf(t, 0, source.deleteCount(legacy.URI), + "the old community.post must SURVIVE: with no valid postv2 to replace it, deleting it destroys the post outright") + + row, found, err := ledger.Get(context.Background(), legacy.URI) + require.NoError(t, err) + require.True(t, found) + assert.Equal(t, posts.RematerializeFallbackLeftLegacy, row.State) + assert.Emptyf(t, row.NewURI, "a fallback row wrote no postv2, so it can name none") +} + +func TestRematerialize_Run_CensusGatesCompletionWhileFallbackSurvives(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + + // One migratable aggregator post and one no-creds human post in the same run. + aggregatorDID := "did:plc:aggregatoraggregatoragg" + humanDID := "did:plc:humantwohumantwohumantwo" + authors.repo(aggregatorDID) + authors.noCreds[humanDID] = true + + migratable := legacyPost(t, rematCommunityDID, aggregatorDID) + stranded := legacyPost(t, rematCommunityDID, humanDID) + source := newFakeLegacySource(migratable, stranded) + writer := &spyAcceptanceWriter{} + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + report, err := tool.Run(context.Background()) + require.NoError(t, err) + + // The migratable one completed; the stranded one is a surviving fallback, so + // the run REFUSES to report complete — the gate on the manual legacy-removal + // follow-up (§11 step 6). + assert.Falsef(t, report.Complete, + "the run must not report complete while a fallback row survives; the operator uses Complete to gate the irreversible legacy-removal step") + assert.GreaterOrEqualf(t, report.Fallbacks, 1, "the census must count the surviving fallback") + assert.GreaterOrEqualf(t, report.Done, 1, "the migratable record must still have reached done") + + // The stranded post's old record must be untouched. + assert.Equalf(t, 0, source.deleteCount(stranded.URI), "a fallback post's old record must never be deleted by the run") + assert.Equalf(t, 1, source.deleteCount(migratable.URI), "the migratable post's old record is deleted") +} + +func TestRematerialize_UsesDirectAcceptanceWriter_NeverReDecides(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ledger := postgres.NewRematerializeLedger(db) + authors := newFakeAuthorFactory() + authors.repo(rematAuthorDID) + writer := &spyAcceptanceWriter{} + legacy := legacyPost(t, rematCommunityDID, rematAuthorDID) + source := newFakeLegacySource(legacy) + + // The Rematerializer's community seam is a CommunityRecordWriter — WriteAcceptance + // direct. There is no AdmissionDecider or AcceptanceEngine field to route + // through, so the tool CANNOT re-run admission. If it could, a since-banned + // author's live post would be rejected here, silently deleting it from a + // community it currently sits in. This mirrors service_writeforward_test.go's + // scriptedDecider trick, made structural: the acceptance is written for the + // post's content unconditionally. + tool := &posts.Rematerializer{Source: source, Ledger: ledger, AuthorRepos: authors.factory(), Acceptances: writer} + + state, err := tool.RematerializeOne(context.Background(), legacy) + require.NoError(t, err) + require.Equal(t, posts.RematerializeDone, state) + + calls := writer.calls() + 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) +} diff --git a/internal/db/postgres/admission_repo_schema_test.go b/internal/db/postgres/admission_repo_schema_test.go --- a/internal/db/postgres/admission_repo_schema_test.go +++ b/internal/db/postgres/admission_repo_schema_test.go @@ -319,12 +319,15 @@ require.NoError(t, err, "with fk_author dropped, a federated author's post must index even though no users row exists for them") // The expected-version parameter is the tripwire, and it has now fired - // twice: migration 035 (post_submissions) and 036 (deleted_accounts) both - // sit on top of 034, so both have to come off first. Rolling back - // explicitly, one asserted step at a time, is what keeps the assertions - // below pointed at 034's Down rather than at whatever happens to be newest. + // three times: migration 035 (post_submissions), 036 (deleted_accounts), and + // 037 (the re-materialization ledger) all sit on top of 034, so all three + // have to come off first. Rolling back explicitly, one asserted step at a + // time, is what keeps the assertions below pointed at 034's Down rather than + // at whatever happens to be newest. + require.EqualValues(t, 37, testkit.MigrateDownOne(t, db, 37), + "037 (the re-materialization ledger) sits on top of 034 and must be rolled back first; asserting which migration came off is what stops this test drifting onto a newer one") require.EqualValues(t, 36, testkit.MigrateDownOne(t, db, 36), - "036 sits on top of 034 and must be rolled back first; asserting which migration came off is what stops this test drifting onto a newer one") + "036 sits on top of 034 and must be rolled back next; asserting which migration came off is what stops this test drifting onto a newer one") require.EqualValues(t, 35, testkit.MigrateDownOne(t, db, 35), "035 sits on top of 034 and must be rolled back next; asserting which migration came off is what stops this test drifting onto a newer one") assert.EqualValues(t, 34, testkit.MigrateDownOne(t, db, 34), diff --git a/internal/db/postgres/rematerialize_ledger.go b/internal/db/postgres/rematerialize_ledger.go new file mode 100644 --- /dev/null +++ b/internal/db/postgres/rematerialize_ledger.go @@ -0,0 +1,64 @@ +package postgres + +import ( + "context" + "database/sql" + "errors" + + "Coves/internal/core/posts" +) + +// PostgreSQL storage for the re-materialization ledger (migration 037): one row +// per legacy social.coves.community.post record, tracking its progress through +// the cutover state machine (docs/PRD_AUTHOR_OWNED_POSTS.md §11). +// +// # THIS FILE IS A RED STUB +// +// It exists so the state-machine tests compile and so cmd/rematerialize-posts +// has a concrete ledger to wire. Every method returns the not-implemented +// sentinel; GREEN implements the single-statement upserts and the census query +// against the migration-037 table. + +// rematerializeLedger is the migration-037-backed posts.RematerializeLedger. +type rematerializeLedger struct { + db *sql.DB +} + +// NewRematerializeLedger returns the Postgres re-materialization ledger. +func NewRematerializeLedger(db *sql.DB) posts.RematerializeLedger { + return &rematerializeLedger{db: db} +} + +var errRematerializeLedgerNotImplemented = errors.New("rematerialize ledger: not implemented (RED stub)") + +func (l *rematerializeLedger) Discover(ctx context.Context, oldURI, authorDID string) (posts.RematerializeLedgerRow, error) { + return posts.RematerializeLedgerRow{}, errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) Get(ctx context.Context, oldURI string) (posts.RematerializeLedgerRow, bool, error) { + return posts.RematerializeLedgerRow{}, false, errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) RecordPostV2Written(ctx context.Context, oldURI, newURI, newCID, newRkey string) error { + return errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) MarkVerified(ctx context.Context, oldURI string) error { + return errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) MarkMigrated(ctx context.Context, oldURI string) error { + return errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) MarkDone(ctx context.Context, oldURI string) error { + return errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) MarkFallback(ctx context.Context, oldURI string, state posts.RematerializeState, reason string) error { + return errRematerializeLedgerNotImplemented +} + +func (l *rematerializeLedger) CountByState(ctx context.Context) (map[posts.RematerializeState]int, error) { + return nil, errRematerializeLedgerNotImplemented +} diff --git a/internal/db/postgres/rematerialize_ledger_schema_test.go b/internal/db/postgres/rematerialize_ledger_schema_test.go new file mode 100644 --- /dev/null +++ b/internal/db/postgres/rematerialize_ledger_schema_test.go @@ -0,0 +1,191 @@ +//go:build integration + +package postgres + +import ( + "context" + "strings" + "testing" + + "Coves/tests/testkit" + + _ "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Migration 037's ledger table, and the state vocabulary the whole cutover turns +// on (docs/PRD_AUTHOR_OWNED_POSTS.md §11 the rev-2.8 deploy runbook). +// +// The re-materialization tool moves every legacy social.coves.community.post to +// an author-owned postv2 plus a community acceptance, and then — and ONLY then — +// deletes the old record. Getting the order wrong makes live posts vanish or +// relaunders a removed post, so the tool is resumable and idempotent, and this +// LEDGER is what makes resume possible: one row per legacy record recording how +// far it got. +// +// The distinction this table has to make representable is migrated ≠ done. +// `migrated` means "verified safe to delete, old record STILL PRESENT"; `done` +// means "old record deleted". A crash between them resumes by retrying ONLY the +// delete, and a delete of an already-gone record is success — which is only +// coherent if the checkpoint BEFORE the delete is its own persisted state. +// +// The two fallback states are the credential census (§11 step 3): a record whose +// author credentials cannot be restored is left as legacy, never re-authored +// under a forged signature, and the run refuses to report "complete" while any +// such row survives. + +const rematerializeLedgerTable = "post_rematerialization_ledger" + +func TestRematerializeLedgerTable_Columns(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + + requireTableExists(t, db, rematerializeLedgerTable) + + type columnShape struct { + dataType string + nullable bool + } + want := map[string]columnShape{ + // The OLD community.post URI is the whole key: one ledger row per legacy + // record, so a re-run finds the existing row rather than accumulating a + // second one for the same post. + "old_uri": {"text", false}, + + // The state machine's cursor. NOT NULL — a row with no state cannot be + // resumed from. + "state": {"text", false}, + + // Who the postv2 is re-authored under (the legacy record's `author` + // field). Nullable is acceptable — a fallback row may never resolve one — + // but it is the audit trail for which repo the tool wrote into. + "author_did": {"text", true}, + + // The postv2 coordinates, populated at the postv2_written transition and + // read back (never recomputed) on resume. NULL until then. + "new_uri": {"text", true}, + "new_cid": {"text", true}, + "new_rkey": {"text", true}, + + // The human-readable note on a fallback row. + "reason": {"text", true}, + + "created_at": {"timestamp with time zone", false}, + "updated_at": {"timestamp with time zone", false}, + } + + rows, err := db.QueryContext(ctx, ` + SELECT column_name, data_type, is_nullable + FROM information_schema.columns + WHERE table_schema = current_schema() AND table_name = $1 + `, rematerializeLedgerTable) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + got := map[string]columnShape{} + for rows.Next() { + var name, dataType, isNullable string + require.NoError(t, rows.Scan(&name, &dataType, &isNullable)) + got[name] = columnShape{dataType: dataType, nullable: isNullable == "YES"} + } + require.NoError(t, rows.Err()) + + for name, wantShape := range want { + gotShape, ok := got[name] + if !assert.Truef(t, ok, "%s.%s is missing", rematerializeLedgerTable, name) { + continue + } + assert.Equalf(t, wantShape.dataType, gotShape.dataType, "%s.%s type", rematerializeLedgerTable, name) + assert.Equalf(t, wantShape.nullable, gotShape.nullable, "%s.%s nullability", rematerializeLedgerTable, name) + } + + assert.Equal(t, []string{"old_uri"}, primaryKeyColumns(t, db, rematerializeLedgerTable), + "the OLD community.post URI is the whole key: one row per legacy record, so a re-run updates in place rather than duplicating the row it resumes from") +} + +func TestRematerializeLedgerTable_StateVocabularyIsClosed(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + requireTableExists(t, db, rematerializeLedgerTable) + + // The tool switches on state exhaustively, so the vocabulary is closed in the + // schema rather than by convention: a typo'd state string would otherwise sit + // in the table as an unresumable row nothing would ever advance. + definitions := checkConstraintDefinitions(t, db, rematerializeLedgerTable) + all := strings.Join(valuesOf(definitions), " ") + + for _, state := range []string{ + "discovered", + "postv2_written", + "verified", + "migrated", + "done", + "fallback_left_legacy", + "fallback_no_creds", + } { + assert.Containsf(t, all, state, + "the state CHECK constraint must admit %q; a missing state is one the tool can never persist", state) + } +} + +func TestRematerializeLedgerTable_ValidStatesInsertAndBogusIsRejected(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + ctx := context.Background() + requireTableExists(t, db, rematerializeLedgerTable) + + // migrated and done are BOTH valid and DISTINCT — the checkpoint-before-delete + // property depends on it. A bogus state must be refused at the schema, where + // every writer meets the constraint, not left to repository discipline. + valid := []string{"discovered", "postv2_written", "verified", "migrated", "done", "fallback_left_legacy", "fallback_no_creds"} + for i, state := range valid { + oldURI := "at://did:plc:community2222222222222222/social.coves.community.post/valid" + string(rune('a'+i)) + _, err := db.ExecContext(ctx, ` + INSERT INTO post_rematerialization_ledger (old_uri, state, created_at, updated_at) + VALUES ($1, $2, NOW(), NOW()) + `, oldURI, state) + require.NoErrorf(t, err, "state %q must be a permitted ledger state", state) + } + + _, err := db.ExecContext(ctx, ` + INSERT INTO post_rematerialization_ledger (old_uri, state, created_at, updated_at) + VALUES ($1, $2, NOW(), NOW()) + `, "at://did:plc:community2222222222222222/social.coves.community.post/bogus", "half_migrated") + require.Errorf(t, err, + "an unknown state 'half_migrated' was accepted; the vocabulary must be closed by a CHECK, or a typo lands as a row nothing resumes") +} + +func TestRematerializeLedgerMigration_RollsBack(t *testing.T) { + t.Parallel() + + db := testkit.DB(t) + requireTableExists(t, db, rematerializeLedgerTable) + + // The expected-version tripwire: 037 is the newest migration, so it is the one + // that comes off. Asserting which migration rolled back is what keeps this + // pointed at 037's Down rather than drifting onto a newer one later. + assert.EqualValues(t, 37, testkit.MigrateDownOne(t, db, 37), + "this test asserts on migration 037's Down section; rolling back a different migration would prove nothing about it") + + var count int + require.NoError(t, db.QueryRowContext(context.Background(), ` + SELECT count(*) FROM information_schema.tables + WHERE table_schema = current_schema() AND table_name = $1 + `, rematerializeLedgerTable).Scan(&count)) + assert.Equalf(t, 0, count, "migration 037's Down must drop %s", rematerializeLedgerTable) +} + +// valuesOf returns a map's values, so the state-vocabulary check can scan every +// CHECK definition on the table at once. +func valuesOf(m map[string]string) []string { + out := make([]string, 0, len(m)) + for _, v := range m { + out = append(out, v) + } + return out +} -- tangled.sh