From de383a206b9176192f6fbbb5c9978f99ae8b2fc9 Mon Sep 17 00:00:00 2001 From: Bretton Date: Sat, 08 Aug 2026 18:30:46 +0000 Subject: [PATCH] feat(review): GREEN — whole-branch review batch fixes (task 8) Prod blockers: oauthScopes grants postv2 create/update/delete (retains community.post through the drain); DeciderDeps.Admissions wired (§8 firehose quota now live in prod); editNote restored to update.json. Tool data-safety: blob bytes copied community→author repo before delete; RawRecord lossless source preserves every published field; verify compares standing body + re-reads acceptance strongRef before checkpoint; Run reconciles stranded migrated rows via ListResumable and Complete requires all non-fallback rows done; credential census preflights before any mutation; RematerializeRkey is a deterministic TID (ParseTID-valid) from the old URI. Co-Authored-By: Claude Fable 5 --- cmd/rematerialize-posts/main.go | 11 ++++++++--- cmd/server/wiring.go | 13 +++++++++++++ internal/atproto/lexicon/social/coves/community/post/update.json | 6 ++++++ internal/core/posts/rematerialize.go | 464 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------------- internal/core/posts/service.go | 8 +++++++- 5 file(s) changed, 463 insertion(s)(+), 39 deletion(s)(-) diff --git a/cmd/rematerialize-posts/main.go b/cmd/rematerialize-posts/main.go --- a/cmd/rematerialize-posts/main.go +++ b/cmd/rematerialize-posts/main.go @@ -258,9 +258,12 @@ } return pds.NewFromAccessToken(fresh.PDSURL, fresh.DID, fresh.PDSAccessToken) } -// legacyPostFromEntry decodes one listRecords entry into a LegacyPost, carrying -// the decoded body forward so the tool re-materializes the record's ACTUAL -// content rather than a re-fetch that might have changed under it. +// legacyPostFromEntry decodes one listRecords entry into a LegacyPost. +// +// The author DID is read out of the decoded body, but the LOSSLESS conversion +// runs off RawRecord — entry.Value verbatim — so every published field +// (langs/tags/crosspostOf/crosspostChain/bridgedStats and the rest) is carried +// through to the postv2 rather than dropped by the lossy PostRecord shape (P5). func legacyPostFromEntry(communityDID string, entry pds.RecordEntry) (posts.LegacyPost, error) { raw, err := json.Marshal(entry.Value) if err != nil { @@ -279,6 +282,8 @@ CID: entry.CID, CommunityDID: communityDID, AuthorDID: record.Author, Record: record, + // The lossless source the postv2 is built from (P5): the raw PDS record. + RawRecord: entry.Value, }, nil } diff --git a/cmd/server/wiring.go b/cmd/server/wiring.go --- a/cmd/server/wiring.go +++ b/cmd/server/wiring.go @@ -249,6 +249,14 @@ func oauthScopes() []string { return []string{ "atproto", "blob:*/*", // avatar and image uploads + // The author-owned post collection: CreatePost, UpdatePost (§3.4), + // post.delete AND the cutover tool all write postv2 through the author's + // own OAuth session, so a scope-enforcing PDS refuses the entire write + // path without this grant. + "repo:social.coves.community.postv2?action=create&action=update&action=delete", + // The deprecated collection is RETAINED through the drain: the cutover tool + // deletes legacy community.post records through these same sessions (§11); + // dropping it would strand every legacy record undeleteable. "repo:social.coves.community.post?action=create&action=update&action=delete", "repo:social.coves.community.comment?action=create&action=update&action=delete", "repo:social.coves.community.profile?action=create&action=update&action=delete", @@ -449,6 +457,11 @@ return posts.DeciderDeps{ Posts: a.postRepo, Communities: a.communityService, Authorizer: a.aggregatorService, + // The §8 firehose quota counter. Without it decider.go's applyQuota + // short-circuits and admits UNLIMITED posts — the same admissions repo the + // ingestion consumer writes and the engine settles, so the rows counted as + // admitted are the rows the quota meters. + Admissions: a.admissionRepo, Aggregators: a.aggregatorService, Policy: posts.AdmissionPolicy{ Ledger: postgresRepo.NewSubmissionLedger(a.db), diff --git a/internal/atproto/lexicon/social/coves/community/post/update.json b/internal/atproto/lexicon/social/coves/community/post/update.json --- a/internal/atproto/lexicon/social/coves/community/post/update.json +++ b/internal/atproto/lexicon/social/coves/community/post/update.json @@ -72,6 +72,12 @@ "type": "string", "maxLength": 640, "maxGraphemes": 64 } + }, + "editNote": { + "type": "string", + "maxLength": 3000, + "maxGraphemes": 300, + "description": "DEPRECATED, retained for backward compatibility: an optional note explaining the edit. This field ships on main; removing it is a non-additive break, so it stays declared as an optional input until a new-NSID change retires it." } } } 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 @@ -2,9 +2,18 @@ package posts import ( "context" + "crypto/sha256" + "encoding/binary" + "encoding/json" "errors" "fmt" + "io" + "net/http" "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + + "Coves/internal/core/blobs" ) // The re-materialization tool: the cutover step that moves every legacy @@ -235,41 +244,126 @@ } // 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. +// It has TWO hard constraints that pull against each other (§11 step 4, whole- +// branch review P9): +// +// 1. A PURE, STABLE FUNCTION OF THE OLD URI ALONE. A re-run must recompute the +// IDENTICAL key so createAuthorRecord converges by read instead of minting a +// second postv2 that dangles every strongRef built from the first. Nothing +// submission-time (fingerprint, bucket, clock) may leak in — the migration +// does not have it and a re-run could not reproduce it — so it is NOT +// SubmissionRkey. +// 2. 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. SubjectRkey (a 52-char base32 digest) is NOT a TID, so it +// cannot be reused here. // -// 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. It is -// SubjectRkey verbatim, not a re-derivation, so the tool and the write path -// cannot come to disagree about what key a subject hashes to. -func RematerializeRkey(legacyPostURI string) string { return SubjectRkey(legacyPostURI) } +// Both are satisfied by deriving a DETERMINISTIC TID from the SHA-256 of the old +// URI: the timestamp and clock bits are drawn from the digest, masked to the +// widths a TID has room for, and encoded with syntax.NewTID — the one encoder +// guaranteed to agree with every ParseTID in the network. Same URI in, same TID +// out; different URIs, different TIDs (SHA-256 collision resistance). +func RematerializeRkey(legacyPostURI string) string { + digest := sha256.Sum256([]byte(legacyPostURI)) + + // The TID timestamp is a 53-bit microsecond field; masking the digest to 53 + // bits keeps it non-negative and inside the range NewTID encodes, while + // staying a pure function of the URI. The value is not a real time — the + // migration has none — but it is stable, which is the only property the rkey + // needs and feed ordering can tolerate. + micros := int64(binary.BigEndian.Uint64(digest[0:8]) & ((1 << 53) - 1)) + + // Ten further digest bits become the clock ID, so two URIs whose 53-bit + // timestamps happened to collide still draw distinct keys. + clockID := uint(binary.BigEndian.Uint16(digest[8:10])) & 0x3FF + + return syntax.NewTID(micros, clockID).String() +} // Run discovers every legacy record and drives each to a terminal state, // returning the census. // +// It runs in THREE ordered passes: +// +// 1. THE CREDENTIAL CENSUS (P8). Every discovered author is resolved with NO +// repo mutation, so an author whose credentials cannot be restored is marked +// a fallback BEFORE a single record is written or deleted. Without this, a +// mutate-as-you-go run would fully migrate and delete the early records +// before ever discovering that a later author is stranded — the exact +// ordering §11 step 3 forbids. +// 2. THE SOURCE PASS. Each listed record is driven to a terminal state. +// 3. THE LEDGER RECONCILE (P7). A row whose delete succeeded but whose MarkDone +// crashed is GONE from the community repo, so the source's listRecords can +// never rediscover it — only ListResumable can. Every non-terminal ledger +// row past the postv2 write is finished from the ledger. +// // A per-record error FAILS THE RUN rather than being logged and skipped: the // safety properties are all ordering ones, and continuing past a record the tool // could not verify would let the operator read a "done"-heavy census as // permission to run the irreversible legacy-removal step while a record sits // half-migrated. A no-creds fallback is NOT such an error — it is an expected -// terminal outcome the census counts — so it lets the run continue while still -// holding Complete false. +// terminal outcome the census counts. func (r *Rematerializer) Run(ctx context.Context) (RematerializeReport, error) { legacies, err := r.Source.ListLegacyPosts(ctx) if err != nil { return RematerializeReport{}, fmt.Errorf("enumerating legacy posts: %w", err) } + // Pass 1 — the census. Resolve EVERY not-yet-started author before ANY repo is + // mutated. A row already past discovered had its credentials confirmed on an + // earlier pass and must not be re-marked — the fallback transition is guarded on + // the discovered state, so re-marking a resumed or already-fallen-back row would + // fail; skipping it keeps the whole run idempotent. + for _, legacy := range legacies { + row, err := r.Ledger.Discover(ctx, legacy.URI, legacy.AuthorDID) + if err != nil { + return RematerializeReport{}, err + } + if row.State != RematerializeDiscovered { + continue + } + if _, err := r.AuthorRepos(ctx, legacy.AuthorDID, nil); err != nil { + if errors.Is(err, ErrNoAuthorCredentials) { + reason := fmt.Sprintf("author %s has no restorable repo credentials: %v", legacy.AuthorDID, err) + if markErr := r.Ledger.MarkFallback(ctx, legacy.URI, RematerializeFallbackLeftLegacy, reason); markErr != nil { + return RematerializeReport{}, markErr + } + continue + } + return RematerializeReport{}, fmt.Errorf("preflighting the credentials of %s: %w", legacy.AuthorDID, err) + } + } + + // Pass 2 — the source pass. A record whose census marked it a fallback is left + // untouched by RematerializeOne (it returns early on a terminal row). for _, legacy := range legacies { if _, err := r.RematerializeOne(ctx, legacy); err != nil { return RematerializeReport{}, fmt.Errorf("re-materializing %s: %w", legacy.URI, err) } } + // Pass 3 — the ledger reconcile. Finish any row the source could not present. + resumable, err := r.Ledger.ListResumable(ctx) + if err != nil { + return RematerializeReport{}, fmt.Errorf("listing resumable rows: %w", err) + } + for _, ledgerRow := range resumable { + // A row still at discovered needs the ORIGINAL record's bytes to build its + // postv2, which the ledger does not hold — only a source listing carries + // them. Such a row is genuinely incomplete and keeps Complete false; it is + // left for a pass whose source can present it, not driven off an empty body. + if ledgerRow.State == RematerializeDiscovered { + continue + } + legacy, err := legacyFromLedgerRow(ledgerRow) + if err != nil { + return RematerializeReport{}, err + } + if _, err := r.RematerializeOne(ctx, legacy); err != nil { + return RematerializeReport{}, fmt.Errorf("reconciling %s: %w", ledgerRow.OldURI, err) + } + } + byState, err := r.Ledger.CountByState(ctx) if err != nil { return RematerializeReport{}, fmt.Errorf("taking the census: %w", err) @@ -285,10 +379,11 @@ if IsFallback(state) { report.Fallbacks += n } } - // The gate on the separate, irreversible legacy-removal follow-up (§11 step - // 6): a surviving fallback is a post still living only as a legacy record, so - // the run must not tell the operator the migration is finished. - report.Complete = report.Fallbacks == 0 + // COMPLETE MEANS EVERY ROW REACHED done — not merely that no fallback survives + // (P7). A row stranded in any non-terminal state, or a surviving fallback, both + // leave Done < Discovered, and the operator's irreversible legacy-removal step + // (§11 step 6) must not run while either is true. + report.Complete = report.Done == report.Discovered return report, nil } @@ -299,9 +394,10 @@ // // The steps are guarded on the ledger state each moves FROM, so a resumed run // re-enters at exactly the step its predecessor stopped before and re-does none // of the completed ones. The load-bearing ordering is VERIFY BEFORE DELETE: the -// old record is deleted only after the postv2 and its acceptance are confirmed to -// pin the same CID, and the migrated checkpoint is persisted BEFORE the delete so -// a crash there retries only the delete. +// old record is deleted only after the postv2, its embed blobs, and its +// acceptance are all confirmed present and consistent, and the migrated +// checkpoint is persisted BEFORE the delete so a crash there retries only the +// delete. func (r *Rematerializer) RematerializeOne(ctx context.Context, legacy LegacyPost) (RematerializeState, error) { row, err := r.Ledger.Discover(ctx, legacy.URI, legacy.AuthorDID) if err != nil { @@ -315,10 +411,11 @@ if IsFallback(row.State) { return row.State, nil } - // Step 1 — postv2_written. Write the author-owned postv2 at the deterministic - // rkey. createAuthorRecord is create-only and converges by read, so a resume - // that re-enters here (it will not, but the guard is honest) would find its own - // first attempt rather than mint a second post. + // Step 1 — postv2_written. Copy the embed blobs into the author's repo, build + // the postv2 LOSSLESSLY from the raw legacy record, and write it at the + // deterministic rkey. createAuthorRecord is create-only and converges by read, + // so a resume that re-enters here finds its own first attempt rather than + // minting a second post. if row.State == RematerializeDiscovered { repo, err := r.AuthorRepos(ctx, legacy.AuthorDID, nil) if err != nil { @@ -336,12 +433,56 @@ } return row.State, fmt.Errorf("opening the author repo of %s: %w", legacy.AuthorDID, err) } + // P5 — the conversion is built from the LOSSLESS raw record, dropping only + // the author field and re-stamping $type. Building it through PostRecord + // would silently strip langs/tags/crosspostOf/crosspostChain/bridgedStats, + // which the old record can never be recovered from once it is deleted. + intended, err := postV2Body(legacy) + if err != nil { + return row.State, err + } + + // P4 — the embed's blob BYTES must live in the AUTHOR's repo before the old + // record (and the community's blob store) can go, or the postv2's media + // resolves against a repo that never held it. The bytes are UPLOADED here, + // before the record that references them is written; the PDS only serves an + // uploaded blob once a record pins it, so presence is VERIFIED after the + // write, below. + if err := r.uploadEmbedBlobs(ctx, repo, legacy); err != nil { + return row.State, err + } + rkey := RematerializeRkey(legacy.URI) - newURI, newCID, _, err := createAuthorRecord(ctx, repo, rkey, postV2From(legacy.Record)) + newURI, newCID, converged, err := createAuthorRecord(ctx, repo, rkey, intended) if err != nil { return row.State, fmt.Errorf("writing the postv2 for %s: %w", legacy.URI, err) } + // P6 — a converged write means a record ALREADY stood at the deterministic + // rkey. createAuthorRecord accepts it by CID, but a CID match alone would + // adopt a DIFFERENT record that merely shares the key. Confirm the standing + // record IS this legacy post's conversion before trusting it; otherwise the + // legacy original would be deleted in favour of a foreign record. + if converged { + standing, err := repo.GetRecord(ctx, PostV2Collection, rkey) + if err != nil { + return row.State, fmt.Errorf("reading the converged postv2 for %s: %w", legacy.URI, err) + } + if !sameRecordBody(standing.Value, intended) { + return row.State, fmt.Errorf( + "a different record already stands at %s in the author's repo: its body is not this legacy post's conversion, so re-materializing would adopt a foreign record and delete the real one", + rkey) + } + } + + // P4 — now the postv2 pins the blobs, so the author repo actually serves + // them. Confirm every embed blob is present BEFORE recording the postv2 (well + // before the migrated checkpoint): a blob the author repo does not serve is a + // broken image the moment the community's copy is garbage-collected. + if err := r.verifyEmbedBlobsPresent(ctx, repo, legacy); err != nil { + return row.State, err + } + if err := r.Ledger.RecordPostV2Written(ctx, legacy.URI, newURI, newCID, rkey); err != nil { return row.State, err } @@ -350,18 +491,29 @@ row.NewURI, row.NewCID, row.NewRkey = newURI, newCID, rkey } // Step 2 — verified. Write the community's acceptance DIRECT (never through the - // engine — see the type's doc) pinning the NEW postv2 CID, then RE-READ the - // postv2 and confirm it still pins that CID. Verification reads the standing - // record rather than trusting the write's returned CID, so a concurrent edit - // landing in the write→verify window is caught here — before anything is - // deleted. + // engine — see the type's doc) pinning the NEW postv2 CID, confirm the + // acceptance actually stands against OUR subject, then RE-READ the postv2 and + // confirm it still pins that CID. Both reads happen before anything is deleted. if row.State == RematerializePostV2Written { - if _, err := r.Acceptances.WriteAcceptance(ctx, CommunityWriteCommand{ + res, err := r.Acceptances.WriteAcceptance(ctx, CommunityWriteCommand{ CommunityDID: legacy.CommunityDID, PostURI: row.NewURI, PostCID: row.NewCID, - }); err != nil { + }) + if err != nil { return row.State, fmt.Errorf("writing the acceptance for %s: %w", row.NewURI, err) + } + + // P6 — verify the acceptance's subject strongRef. Its record key is the + // digest of the subject URI, so a matching rkey proves the acceptance is FOR + // our postv2; an empty CID would mean no record stands at all. A write that + // returned neither has not made the community's acceptance real, and deleting + // the legacy record on the strength of it would drop the post out of its + // community. + if res.CID == "" || res.RKey != SubjectRkey(row.NewURI) { + return row.State, fmt.Errorf( + "the acceptance for %s did not stand against the expected subject (got rkey %q cid %q, want rkey %q)", + row.NewURI, res.RKey, res.CID, SubjectRkey(row.NewURI)) } repo, err := r.AuthorRepos(ctx, legacy.AuthorDID, nil) @@ -389,9 +541,9 @@ } row.State = RematerializeVerified } - // Step 3 — migrated. The checkpoint BEFORE the delete: postv2 and acceptance - // verified, old record still present. Persisting it as its own state is what - // lets a crash on the delete retry ONLY the delete. + // Step 3 — migrated. The checkpoint BEFORE the delete: postv2, blobs and + // acceptance verified, old record still present. Persisting it as its own state + // is what lets a crash on the delete retry ONLY the delete. if row.State == RematerializeVerified { if err := r.Ledger.MarkMigrated(ctx, legacy.URI); err != nil { return row.State, err @@ -413,3 +565,245 @@ } return row.State, nil } + +// maxRematerializeBlobBytes caps a single blob copy. It is generous — larger than +// any post media the lexicons admit — because the bytes come from our own +// community repo, not an untrusted origin; the cap exists to bound a corrupt or +// runaway response, not to enforce a content policy. +const maxRematerializeBlobBytes = 100 << 20 // 100 MiB + +// postV2Body builds the author-owned postv2 record from the legacy record's +// LOSSLESS raw map: every published field is carried through byte-for-byte, only +// the `author` field is dropped (authorship is the repo now, §3.1) and the $type +// is re-stamped to the postv2 collection. +func postV2Body(legacy LegacyPost) (map[string]any, error) { + if len(legacy.RawRecord) == 0 { + return nil, fmt.Errorf("re-materializing %s: the legacy record carries no RawRecord to convert losslessly", legacy.URI) + } + body := cloneRecord(legacy.RawRecord) + delete(body, "author") + body["$type"] = PostV2Collection + return body, nil +} + +// uploadEmbedBlobs copies every embed blob's BYTES from the community's blob store +// into the author's repo. It runs BEFORE the postv2 record is written, because a +// record's embed may only reference blobs the repo has already received. +// +// The bytes are fetched via com.atproto.sync.getBlob against the instance PDS +// (the host the author repo is bound to, which also hosts the community's repo on +// a Coves instance) and uploaded through the author's own credentialed +// UploadBlob. A blob left uncopied fails the record here rather than after the +// old bytes are gone. +func (r *Rematerializer) uploadEmbedBlobs(ctx context.Context, repo AuthorRepo, legacy LegacyPost) error { + refs := extractBlobRefs(cloneRecord(legacy.RawRecord)) + if len(refs) == 0 { + return nil + } + + host, ok := repoHostURL(repo) + if !ok { + return fmt.Errorf("copying blobs for %s: the author repo exposes no host URL to fetch the community's blobs from", legacy.URI) + } + + for _, ref := range refs { + data, err := fetchBlobBytes(ctx, host, legacy.CommunityDID, ref.cid) + if err != nil { + return fmt.Errorf("fetching embed blob %s from %s: %w", ref.cid, legacy.CommunityDID, err) + } + mimeType := ref.mimeType + if mimeType == "" { + mimeType = "application/octet-stream" + } + if _, err := repo.UploadBlob(ctx, data, mimeType); err != nil { + return fmt.Errorf("uploading embed blob %s into the author repo of %s: %w", ref.cid, repo.DID(), err) + } + } + return nil +} + +// verifyEmbedBlobsPresent confirms every embed blob is served by the author's +// repo — the P4 guarantee that the postv2's media resolves against the author, +// not the community repo the old record is about to be deleted from. It runs +// AFTER the postv2 is written, because the PDS serves a blob only once a record +// pins it; a 200 from getBlob proves the bytes actually landed. +func (r *Rematerializer) verifyEmbedBlobsPresent(ctx context.Context, repo AuthorRepo, legacy LegacyPost) error { + refs := extractBlobRefs(cloneRecord(legacy.RawRecord)) + if len(refs) == 0 { + return nil + } + + host, ok := repoHostURL(repo) + if !ok { + return fmt.Errorf("verifying blobs for %s: the author repo exposes no host URL", legacy.URI) + } + + for _, ref := range refs { + if !blobPresent(ctx, host, repo.DID(), ref.cid) { + return fmt.Errorf("embed blob %s is not present in the author repo of %s after the postv2 write; refusing to proceed toward the delete", ref.cid, repo.DID()) + } + } + return nil +} + +// rematerializeBlobRef is one embed blob to copy: its CID and MIME type, both +// read from the blob reference in the record. +type rematerializeBlobRef struct { + cid string + mimeType string +} + +// extractBlobRefs walks a decoded record tree and collects every blob reference — +// an object whose $type is "blob" carrying a ref CID link. It descends maps and +// arrays so a blob nested anywhere in the embed union (images, video, external +// thumb) is found. +func extractBlobRefs(node any) []rematerializeBlobRef { + var out []rematerializeBlobRef + switch v := node.(type) { + case map[string]any: + if t, _ := v["$type"].(string); t == "blob" { + if cid := blobLinkCID(v); cid != "" { + mimeType, _ := v["mimeType"].(string) + out = append(out, rematerializeBlobRef{cid: cid, mimeType: mimeType}) + } + } + for _, child := range v { + out = append(out, extractBlobRefs(child)...) + } + case []any: + for _, child := range v { + out = append(out, extractBlobRefs(child)...) + } + } + return out +} + +// blobLinkCID reads the CID out of a decoded blob reference, tolerating both the +// canonical {"ref":{"$link":cid}} shape and a bare string ref. +func blobLinkCID(blob map[string]any) string { + switch ref := blob["ref"].(type) { + case map[string]any: + if link, ok := ref["$link"].(string); ok { + return link + } + case string: + return ref + } + return "" +} + +// fetchBlobBytes downloads a blob's bytes from a repo via com.atproto.sync.getBlob. +func fetchBlobBytes(ctx context.Context, host, did, cid string) ([]byte, error) { + blobURL := blobs.HydrateBlobURL(host, did, cid) + if blobURL == "" { + return nil, fmt.Errorf("could not build a getBlob URL for %s / %s", did, cid) + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobURL, nil) + if err != nil { + return nil, err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("getBlob returned status %d", resp.StatusCode) + } + return io.ReadAll(io.LimitReader(resp.Body, maxRematerializeBlobBytes)) +} + +// blobPresent reports whether a repo serves a blob — a 200 from getBlob proves the +// repo actually holds the bytes. +func blobPresent(ctx context.Context, host, did, cid string) bool { + blobURL := blobs.HydrateBlobURL(host, did, cid) + if blobURL == "" { + return false + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobURL, nil) + if err != nil { + return false + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return false + } + defer func() { _ = resp.Body.Close() }() + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxRematerializeBlobBytes)) + return resp.StatusCode == http.StatusOK +} + +// repoHostURL extracts the PDS host the author repo is bound to, when the concrete +// repo exposes one. The production pds.Client does; the state-machine fakes do +// not, and they never need it because their records carry no blobs. +func repoHostURL(repo AuthorRepo) (string, bool) { + if h, ok := repo.(interface{ HostURL() string }); ok { + if host := h.HostURL(); host != "" { + return host, true + } + } + return "", false +} + +// cloneRecord deep-copies a decoded record through a JSON round-trip, which both +// detaches it from the caller's map and NORMALISES any typed value (a struct blob +// ref, a json.Number) into the plain map/slice/string/float64 tree the conversion +// and the blob walk expect. +func cloneRecord(record map[string]any) map[string]any { + raw, err := json.Marshal(record) + if err != nil { + return map[string]any{} + } + var out map[string]any + if err := json.Unmarshal(raw, &out); err != nil { + return map[string]any{} + } + return out +} + +// sameRecordBody reports whether two decoded records are byte-identical once +// canonicalised. json.Marshal sorts map keys, so two equal records — regardless +// of how their values were originally typed — serialise to the same bytes. +func sameRecordBody(a, b map[string]any) bool { + aj, err := json.Marshal(a) + if err != nil { + return false + } + bj, err := json.Marshal(b) + if err != nil { + return false + } + return string(aj) == string(bj) +} + +// legacyFromLedgerRow reconstructs the minimal LegacyPost the reconcile pass needs +// to finish a row past the postv2 write: the delete step keys off the old URI, and +// the community DID is parsed back out of it. It deliberately carries no +// RawRecord — a reconciled row is past the point where the record body is read. +func legacyFromLedgerRow(row RematerializeLedgerRow) (LegacyPost, error) { + communityDID, err := communityDIDFromURI(row.OldURI) + if err != nil { + return LegacyPost{}, err + } + return LegacyPost{ + URI: row.OldURI, + CommunityDID: communityDID, + AuthorDID: row.AuthorDID, + }, nil +} + +// communityDIDFromURI extracts the repo authority (the community DID) from an +// at:// record URI. +func communityDIDFromURI(uri string) (string, error) { + const scheme = "at://" + if len(uri) <= len(scheme) || uri[:len(scheme)] != scheme { + return "", fmt.Errorf("cannot extract a community DID from %q: not an at:// URI", uri) + } + rest := uri[len(scheme):] + for i := 0; i < len(rest); i++ { + if rest[i] == '/' { + return rest[:i], nil + } + } + return "", fmt.Errorf("cannot extract a community DID from %q: no collection path", uri) +} diff --git a/internal/core/posts/service.go b/internal/core/posts/service.go --- a/internal/core/posts/service.go +++ b/internal/core/posts/service.go @@ -334,7 +334,13 @@ // of the community it was accepted into. // // converged reports that no new record was written — the caller is looking at a // post that already existed. -func createAuthorRecord(ctx context.Context, repo AuthorRepo, rkey string, record PostV2Record) (uri, cid string, converged bool, err error) { +// +// record is `any` rather than PostV2Record because the two callers assemble the +// body differently and both are correct: the write path passes a typed +// PostV2Record (postV2From), while the re-materialization tool passes the legacy +// record's lossless map so no published field is dropped in the conversion. Both +// serialise to the same postv2 shape; the guard and read-back are identical. +func createAuthorRecord(ctx context.Context, repo AuthorRepo, rkey string, record any) (uri, cid string, converged bool, err error) { commit, err := repo.PutRecordWithCommit(ctx, PostV2Collection, rkey, record, "") if err == nil { return commit.URI, commit.CID, false, nil -- tangled.sh