diff --git a/cmd/server/consumers.go b/cmd/server/consumers.go --- a/cmd/server/consumers.go +++ b/cmd/server/consumers.go @@ -2,6 +2,7 @@ package main import ( "Coves/internal/atproto/jetstream" + postgresRepo "Coves/internal/db/postgres" "context" "errors" "fmt" @@ -160,12 +161,30 @@ a.communityRepo, a.cfg.Instance.DID, a.cfg.Instance.SkipDIDWebVerification, a.identityResolver, jetstream.WithCommunityRevGate(a.revGate)), }) - // Posts created in community repositories. + // Posts, in both shapes, plus the community records that decide about + // them: the deprecated community-repo post, the author-repo postv2, and + // the acceptance/removal pair. One consumer, because they write the same + // admission row and an acceptance is meaningless without the post it pins. + // + // The direct fetcher is what makes acceptance-before-post converge without + // full relay coverage (PRD §5.4). It dials a PDS named by a DID document + // anyone can publish, so its SSRF guard stays on in production and the + // stood-down constructor is reachable only under IS_DEV_ENV — where the + // hermetic stack's PDS is a private address the guard would otherwise + // refuse. + postFetcher := jetstream.NewDirectPostFetcher(a.identityResolver) + if a.cfg.IsDevEnv { + slog.Warn("direct post fetch has SSRF protection DISABLED (IS_DEV_ENV); this must never be set in production") + postFetcher = jetstream.NewDevDirectPostFetcher(a.identityResolver) + } consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerPosts, handler: jetstream.NewPostEventConsumer(a.postRepo, a.communityRepo, a.userService, a.db, jetstream.WithPostBridgeTrust(a.bridgeTrust), - jetstream.WithPostIdentityResolver(a.identityResolver)), + jetstream.WithPostIdentityResolver(a.identityResolver), + jetstream.WithAdmissions(a.admissionRepo), + jetstream.WithDeletedAccounts(postgresRepo.NewDeletedAccountRepository(a.db)), + jetstream.WithPostRecordFetcher(postFetcher)), }) // Aggregators: service declarations and authorization records, following diff --git a/cmd/server/routes.go b/cmd/server/routes.go --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -73,7 +73,8 @@ // Posts accept dual auth so aggregator bots can publish with a service // JWT or API key rather than a user's OAuth session. routes.RegisterPostRoutes(r, app.postService, app.voteService, app.blueskyService, - app.dualAuth, app.authMiddleware) + app.dualAuth, app.authMiddleware, + routes.WithPostStatusService(app.postStatusService)) routes.RegisterVoteRoutes(r, app.voteService, app.authMiddleware) routes.RegisterUserBlockRoutes(r, app.userBlockService, app.authMiddleware) diff --git a/cmd/server/wiring.go b/cmd/server/wiring.go --- a/cmd/server/wiring.go +++ b/cmd/server/wiring.go @@ -97,11 +97,19 @@ voteRepo votes.Repository commentRepo comments.Repository userBlockRepo userblocks.Repository aggregatorRepo aggregators.Repository + // admissionRepo is shared by the ingestion consumer, which WRITES the + // per-(community, post) decisions, and the status query, which reads them. + admissionRepo posts.AdmissionRepository // Domain services - userService users.UserService - communityService communities.Service - postService posts.Service + userService users.UserService + communityService communities.Service + postService posts.Service + // postStatusService answers post.getStatus. Separate from postService + // because a status query needs the admissions store and nothing else, and + // widening the write-path interface to reach it would make every test + // double of posts.Service carry a method it has no opinion about. + postStatusService posts.StatusService voteService votes.Service commentService comments.Service userBlockService userblocks.Service @@ -251,6 +259,7 @@ a.voteRepo = postgresRepo.NewVoteRepository(a.db) a.commentRepo = postgresRepo.NewCommentRepository(a.db) a.userBlockRepo = postgresRepo.NewUserBlockRepository(a.db) a.aggregatorRepo = postgresRepo.NewAggregatorRepository(a.db) + a.admissionRepo = postgresRepo.NewAdmissionRepository(a.db) } func (a *application) buildServices(ctx context.Context) error { @@ -347,6 +356,12 @@ }, Now: time.Now, }), ) + + // getStatus is how an author on another server learns what happened to + // their post. It is the ONLY way a rejection is reachable — a submission + // refused before it was ever accepted writes no community record, so there + // is nothing on the firehose to read. + a.postStatusService = posts.NewStatusService(a.admissionRepo) // Subject existence is deliberately not validated: the vote is written to // the user's own PDS regardless, and the Jetstream consumer only updates diff --git a/internal/api/handlers/post/getstatus.go b/internal/api/handlers/post/getstatus.go --- a/internal/api/handlers/post/getstatus.go +++ b/internal/api/handlers/post/getstatus.go @@ -1,14 +1,13 @@ package post import ( + "encoding/json" + "log" "net/http" + "time" "Coves/internal/core/posts" ) - -// RED STUB (task 5, cycle 1). Signature only — HandleGetStatus writes nothing, -// so every assertion in getstatus_integration_test.go fails on the response -// rather than on a missing symbol. The implementation is GREEN's. // GetStatusHandler serves social.coves.community.post.getStatus: one // community's decision about one post (docs/PRD_AUTHOR_OWNED_POSTS.md §3.4). @@ -33,4 +32,72 @@ // HandleGetStatus handles // GET /xrpc/social.coves.community.post.getStatus?post=at://...&community=did:... func (h *GetStatusHandler) HandleGetStatus(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + postURI := r.URL.Query().Get("post") + communityDID := r.URL.Query().Get("community") + + // Both halves are refused rather than defaulted. A post carries independent + // decisions from several communities (§2), so an incomplete subject is not + // an under-specified question with an obvious answer — it is a different + // question, and answering about whichever row turned up first would report + // one community's verdict as another's. + if postURI == "" { + writeError(w, http.StatusBadRequest, "InvalidRequest", "post parameter is required") + return + } + if communityDID == "" { + writeError(w, http.StatusBadRequest, "InvalidRequest", "community parameter is required") + return + } + if len(postURI) > maxURILength { + writeError(w, http.StatusBadRequest, "InvalidRequest", "post URI exceeds maximum length") + return + } + if len(communityDID) > maxURILength { + writeError(w, http.StatusBadRequest, "InvalidRequest", "community DID exceeds maximum length") + return + } + + status, err := h.service.GetStatus(r.Context(), posts.GetStatusRequest{ + PostURI: postURI, + CommunityDID: communityDID, + }) + if err != nil { + handleServiceError(w, err) + return + } + + // Built field by field so an absent optional field is ABSENT from the JSON + // rather than present and null. The distinction is the client's: a caller + // polling for the accepted transition reads `"decisionCode": null` as a + // decision that was made, when in fact none exists. + body := map[string]interface{}{"status": string(status.Status)} + if status.DecisionCode != nil { + body["decisionCode"] = *status.DecisionCode + } + if status.DecisionAt != nil { + body["decisionAt"] = status.DecisionAt.UTC().Format(time.RFC3339) + } + if status.AcceptanceURI != nil { + body["acceptanceUri"] = *status.AcceptanceURI + } + + // Pre-encoded so an encoding failure still yields a proper error response + // rather than a 200 with a truncated body (mirrors post.get). + responseBytes, err := json.Marshal(body) + if err != nil { + log.Printf("ERROR: Failed to encode getStatus response: %v", err) + writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to encode response") + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + if _, err := w.Write(responseBytes); err != nil { + log.Printf("ERROR: Failed to write getStatus response: %v", err) + } } diff --git a/internal/api/routes/post.go b/internal/api/routes/post.go --- a/internal/api/routes/post.go +++ b/internal/api/routes/post.go @@ -10,6 +10,28 @@ "github.com/go-chi/chi/v5" ) +// PostRouteOption supplies a collaborator that only some of the post routes +// need. +// +// It is variadic rather than another positional parameter because the status +// query is served by its OWN service (posts.StatusService, which needs the +// admissions store and nothing else) and every existing caller — including the +// minimal wirings in tests — would otherwise have to name a dependency it has +// no opinion about. +type PostRouteOption func(*postRouteConfig) + +type postRouteConfig struct { + statusService posts.StatusService +} + +// WithPostStatusService supplies the service behind +// social.coves.community.post.getStatus. The route is registered either way, so +// that the HTTP surface does not silently change shape with the wiring; without +// this option the handler has no service to call. +func WithPostStatusService(service posts.StatusService) PostRouteOption { + return func(c *postRouteConfig) { c.statusService = service } +} + // RegisterPostRoutes registers post-related XRPC endpoints on the router // Implements social.coves.community.post.* lexicon endpoints // authMiddleware can be either OAuthAuthMiddleware or DualAuthMiddleware (used for @@ -22,7 +44,12 @@ voteService votes.Service, blueskyService blueskypost.Service, authMiddleware middleware.AuthMiddleware, oauthMiddleware *middleware.OAuthAuthMiddleware, + opts ...PostRouteOption, ) { + var cfg postRouteConfig + for _, opt := range opts { + opt(&cfg) + } // oauthMiddleware.OptionalAuth gates the public get endpoint below. A nil value is a // wiring bug (minimal/test setups) that would otherwise panic on the first request to // post.get; fail fast at registration with a clear message instead. @@ -49,6 +76,19 @@ // social.coves.community.post.get - batch fetch post views by AT-URI. // Public endpoint with optional auth so authenticated viewers receive their vote state. // Used for feed-skeleton hydration and permalink / cold-load rendering. r.With(oauthMiddleware.OptionalAuth).Get("/xrpc/social.coves.community.post.get", getHandler.HandleGet) + + // social.coves.community.post.getStatus - one community's admission + // decision about one post. + // + // NO auth middleware at all, unlike post.get beside it. The caller this is + // built for is an author on ANOTHER server asking this host whether it + // accepted their post (PRD §7): they have no account here, so there is no + // session to require and no viewer state to personalise. OptionalAuth would + // be harmless but pointless — the answer does not vary by viewer — while + // RequireAuth would make the cross-server case unanswerable, which is the + // asymmetry internal/api/routes/registration_test.go declares. + statusHandler := post.NewGetStatusHandler(cfg.statusService) + r.Get("/xrpc/social.coves.community.post.getStatus", statusHandler.HandleGetStatus) // Future endpoints (Beta): // r.With(authMiddleware.RequireAuth).Post("/xrpc/social.coves.community.post.update", updateHandler.HandleUpdate) diff --git a/internal/atproto/jetstream/authorpost.go b/internal/atproto/jetstream/authorpost.go --- a/internal/atproto/jetstream/authorpost.go +++ b/internal/atproto/jetstream/authorpost.go @@ -2,17 +2,40 @@ package jetstream import ( "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "log" "net/http" + "net/url" + "strconv" + "strings" + "time" "Coves/internal/atproto/identity" + "Coves/internal/atproto/oauth" + "Coves/internal/core/communities" "Coves/internal/core/posts" + "Coves/internal/core/users" ) -// RED STUB (task 5, cycle 1). Declarations only — every function body here -// returns zero values, so the tests describing author-owned post ingestion -// compile and fail on their assertions rather than on missing symbols. The -// implementations, the HandleEvent dispatch for the three new collections, and -// the consumerWantedCollections entries are GREEN's. +// Ingesting author-owned posts and the community records that decide about +// them (docs/PRD_AUTHOR_OWNED_POSTS.md §5.3-§5.6). +// +// Three collections land here, and they invert each other: +// +// - social.coves.community.postv2 arrives from the AUTHOR's repo, so +// event.Did IS the author and the community is a claim the record makes. +// - social.coves.community.acceptance and .removal arrive from the +// COMMUNITY's repo, so event.Did IS the community and the post is a +// subject the record names. +// +// The old community.post path (still in post_consumer.go) checks that the repo +// DID EQUALS the record's community. Here that check splits in two opposite +// directions, which is why these handlers are their own file rather than more +// branches in the old ones. // PostV2Collection is the author-repo post record of // docs/PRD_AUTHOR_OWNED_POSTS.md §3.1 — the §3.0 successor to the deprecated @@ -60,6 +83,10 @@ func WithPostRecordFetcher(fetcher PostRecordFetcher) PostEventConsumerOption { return func(c *PostEventConsumer) { c.postFetcher = fetcher } } +// --------------------------------------------------------------------------- +// §5.4 direct fetch +// --------------------------------------------------------------------------- + // FetchedPost is one author-repo record read directly from its PDS. // // It carries the CID separately from the record because the CID is what the @@ -105,6 +132,19 @@ // network, driven by any stranger who writes an acceptance record. allowPrivateHosts bool } +// maxFetchedRecordBytes bounds how much of a PDS getRecord response is read. +// +// A post record has a lexicon-bounded size, so a PDS streaming megabytes is +// either broken or hostile — and the host is chosen by a stranger's record, so +// an unbounded read here is a memory-exhaustion primitive handed to the public. +// The cap mirrors users.maxProfileResponseBytes, which bounds the same call for +// the same reason. +const maxFetchedRecordBytes = 1 << 20 // 1 MiB + +// maxFetchErrorDetailBytes caps how much of a failing PDS response is echoed +// into the logs, so a hostile host cannot flood them. +const maxFetchErrorDetailBytes = 256 + // NewDirectPostFetcher wires the §5.4 fetch. SSRF protection is ON and there is // no parameter to turn it off: a constructor that accepted a boolean is a // constructor someone eventually passes true to from production wiring. @@ -112,12 +152,912 @@ func NewDirectPostFetcher(resolver identity.Resolver) *DirectPostFetcher { return &DirectPostFetcher{resolver: resolver} } +// NewDevDirectPostFetcher builds a fetcher with the SSRF guard STOOD DOWN, for +// development and hermetic test stacks whose PDS is reachable only on a private +// address. +// +// It is a separate constructor rather than a flag on the safe one so that the +// dangerous choice has to be named at the call site, where a reviewer sees it, +// and so that no production wiring can reach it by passing a variable that +// happens to be true. Its one caller is gated on IS_DEV_ENV. +func NewDevDirectPostFetcher(resolver identity.Resolver) *DirectPostFetcher { + return &DirectPostFetcher{resolver: resolver, allowPrivateHosts: true} +} + // httpClient builds the guarded client for one fetch. Declared here so the // guard is derived from allowPrivateHosts at call time rather than baked into a // client at construction, where a test seam could not reach it. -func (f *DirectPostFetcher) httpClient() *http.Client { return nil } +func (f *DirectPostFetcher) httpClient() *http.Client { + return oauth.NewSSRFSafeHTTPClient(f.allowPrivateHosts) +} // FetchPost implements PostRecordFetcher. func (f *DirectPostFetcher) FetchPost(ctx context.Context, postURI string) (*FetchedPost, error) { - return nil, nil + repoDID, collection, rkey, ok := parseRecordURI(postURI) + if !ok { + return nil, fmt.Errorf("cannot fetch %q: not an at:// record URI", postURI) + } + if f.resolver == nil { + return nil, fmt.Errorf("cannot fetch %s: no identity resolver is wired", postURI) + } + + // The PDS is resolved from the DID document rather than taken from anything + // the acceptance record said. The record names a subject; where that + // subject's repo lives is a fact about the DID, and letting a record assert + // it would let the record choose the host this request goes to. + resolved, err := f.resolver.Resolve(ctx, repoDID) + if err != nil { + return nil, fmt.Errorf("resolving the repo of %s: %w", postURI, err) + } + if resolved == nil || resolved.PDSURL == "" { + return nil, fmt.Errorf("resolving the repo of %s: no PDS endpoint in the DID document", postURI) + } + + endpoint := strings.TrimSuffix(resolved.PDSURL, "/") + "/xrpc/com.atproto.repo.getRecord?repo=" + + url.QueryEscape(repoDID) + "&collection=" + url.QueryEscape(collection) + + "&rkey=" + url.QueryEscape(rkey) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, fmt.Errorf("building the getRecord request for %s: %w", postURI, err) + } + + resp, err := f.httpClient().Do(req) + if err != nil { + return nil, fmt.Errorf("fetching %s from its PDS: %w", postURI, err) + } + defer func() { _ = resp.Body.Close() }() + + // One byte past the cap, so an over-cap body is DETECTED rather than + // silently truncated and then parsed as if it were whole. A truncated + // record that happened to parse would be indexed as the author's content. + body, err := io.ReadAll(io.LimitReader(resp.Body, maxFetchedRecordBytes+1)) + if err != nil { + return nil, fmt.Errorf("reading the getRecord response for %s: %w", postURI, err) + } + if len(body) > maxFetchedRecordBytes { + return nil, fmt.Errorf("the PDS serving %s returned more than %d bytes", postURI, maxFetchedRecordBytes) + } + + if resp.StatusCode != http.StatusOK { + detail := string(body) + if len(detail) > maxFetchErrorDetailBytes { + detail = detail[:maxFetchErrorDetailBytes] + } + // Quoted so control characters and ANSI escapes from a hostile PDS + // cannot corrupt log output. + return nil, fmt.Errorf("the PDS serving %s answered getRecord with status %d: %s", + postURI, resp.StatusCode, strconv.Quote(detail)) + } + + var parsed struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value map[string]interface{} `json:"value"` + } + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, fmt.Errorf("parsing the getRecord response for %s: %w", postURI, err) + } + if parsed.Value == nil { + return nil, fmt.Errorf("the getRecord response for %s carried no record value", postURI) + } + if parsed.CID == "" { + // Without a CID there is nothing to verify the pinned reference + // against, and an unverified record is exactly what the fetch must + // never index. + return nil, fmt.Errorf("the getRecord response for %s carried no CID", postURI) + } + + return &FetchedPost{URI: postURI, CID: parsed.CID, Record: parsed.Value}, nil +} + +// --------------------------------------------------------------------------- +// The author-repo post record +// --------------------------------------------------------------------------- + +// AuthorPostRecord is a social.coves.community.postv2 record as it arrives from +// Jetstream. +// +// It has NO author field, and that absence is enforced by the type rather than +// by discipline: authorship comes from the repo the record lives in, so a +// struct that could hold an author is a struct someone eventually reads one +// from — which is precisely the impersonation the flip removed. The lexicon has +// no such property either, so a record carrying one is a forger's field and is +// ignored here by construction. +type AuthorPostRecord struct { + Title *string `json:"title,omitempty"` + Content *string `json:"content,omitempty"` + Embed map[string]interface{} `json:"embed,omitempty"` + Labels *posts.SelfLabels `json:"labels,omitempty"` + BridgedStats *BridgedStatsFromJetstream `json:"bridgedStats,omitempty"` + Type string `json:"$type"` + Community string `json:"community"` + CreatedAt string `json:"createdAt"` + Facets []interface{} `json:"facets,omitempty"` +} + +// parseAuthorPostRecord converts a raw Jetstream record map into an +// AuthorPostRecord, refusing the shapes that can never become valid. +func parseAuthorPostRecord(record map[string]interface{}) (*AuthorPostRecord, error) { + recordJSON, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal postv2 record: %w", err) + } + + var parsed AuthorPostRecord + if err := json.Unmarshal(recordJSON, &parsed); err != nil { + // PERMANENT: the record's shape doesn't match the lexicon (wrong field + // types); replaying the identical bytes can never parse differently. + return nil, fmt.Errorf("%w: failed to unmarshal postv2 record: %v", ErrPermanentEvent, err) + } + + // PERMANENT for the same reason: a record missing a required field is + // structurally invalid forever. `community` is the submission target, so a + // record without one names no admission subject at all. + if parsed.Community == "" { + return nil, fmt.Errorf("%w: postv2 record missing community field", ErrPermanentEvent) + } + if parsed.CreatedAt == "" { + return nil, fmt.Errorf("%w: postv2 record missing createdAt field", ErrPermanentEvent) + } + + return &parsed, nil +} + +// --------------------------------------------------------------------------- +// postv2: the author's own post record +// --------------------------------------------------------------------------- + +// handleAuthorPostEvent routes one social.coves.community.postv2 commit. +// +// event.Did is the AUTHOR, unconditionally and for every operation below. +func (c *PostEventConsumer) handleAuthorPostEvent(ctx context.Context, event *JetstreamEvent, commit *CommitEvent) error { + authorDID := event.Did + + // THE ERASURE GATE, and it runs before anything else touches the database — + // before parsing, before hydration, before the rev gate. An event from an + // account this AppView was asked to forget has nothing to do, and the + // cheapest way to guarantee that is to leave before any code path that + // could write. Ordering it after a parse would also mean a malformed record + // from an erased account dead-letters, which is operational noise about + // content nobody may keep. + erased, err := c.authorWasErased(ctx, authorDID) + if err != nil { + return err + } + if erased { + // Nil, not an error. The connector dead-letters whatever a handler + // returns, so refusing here would fill the queue with rows that redrive, + // fail identically and retire — every erased account becoming a + // permanent stream of noise. This is not a failure; it is an event with + // nothing to do. + log.Printf("INFO: dropping %s %s for erased account %s (migration 036 marker)", + PostV2Collection, commit.Operation, authorDID) + return nil + } + + switch commit.Operation { + case "create", "update": + return c.upsertAuthorPost(ctx, authorDID, commit, event.TimeUS) + case "delete": + return c.tombstoneRecord(ctx, recordURI(authorDID, PostV2Collection, commit.RKey), commit.Rev) + } + return nil +} + +// canRecordAdmissions reports whether this consumer has somewhere to put a +// decision. +// +// Every one of the three author-owned collections exists to write an admission +// row, so a consumer built without the store cannot handle any of them — it is +// running in its pre-034 shape. Ignoring the events is the honest answer: +// indexing a postv2 with no admission row would publish a post no community +// ever decided about, which is worse than not indexing it at all. It is logged +// because in production this is always a wiring bug. +func (c *PostEventConsumer) canRecordAdmissions(collection string) bool { + if c.admissions != nil { + return true + } + log.Printf("WARNING: ignoring %s event - this consumer has no admissions store wired", collection) + return false +} + +// authorWasErased reports whether this DID carries a migration-036 erasure +// marker. A lookup failure is an ERROR, never a false: failing open would +// re-index the content a deletion erased, which is the one outcome the marker +// exists to prevent. With no lookup wired the gate is absent and everything +// indexes, which is the pre-036 behaviour. +func (c *PostEventConsumer) authorWasErased(ctx context.Context, did string) (bool, error) { + if c.deletedAccounts == nil { + return false, nil + } + erased, err := c.deletedAccounts.IsAccountDeleted(ctx, did) + if err != nil { + return false, fmt.Errorf("checking the erasure marker for %s: %w", did, err) + } + return erased, nil +} + +// upsertAuthorPost indexes an author-repo post and opens (or refreshes) the +// pending admission the community will decide against. +// +// A postv2 event never DECIDES anything. It records content plus the fact that +// the author submitted it; whether the community shows the post lives in +// community_post_admissions and is written only by community events. +func (c *PostEventConsumer) upsertAuthorPost(ctx context.Context, authorDID string, commit *CommitEvent, timeUS int64) error { + if commit.Record == nil { + return fmt.Errorf("%w: postv2 %s event missing record data", ErrPermanentEvent, commit.Operation) + } + + record, err := parseAuthorPostRecord(commit.Record) + if err != nil { + return err + } + + uri := recordURI(authorDID, PostV2Collection, commit.RKey) + + // The community must be one this AppView has indexed, or there is no + // subject to open an admission against. + // + // Deliberately NOT permanent: BigSky preserves order within a repo, not + // across repos, so a post can genuinely arrive before the community's own + // profile event. Marking this permanent would discard every post that + // merely arrived early, with the redrive that would have fixed it already + // spent. + if _, err := c.communityRepo.GetByDID(ctx, record.Community); err != nil { + if communities.IsNotFound(err) { + log.Printf("Error: cannot index %s before its community %s is indexed", uri, record.Community) + return fmt.Errorf("community not found: %s - cannot index post before community", record.Community) + } + return fmt.Errorf("%w: failed to verify community %s exists: %v", errValidationInfra, record.Community, err) + } + + stored, found, err := c.loadStoredPost(ctx, uri) + if err != nil { + return err + } + + // IMMUTABILITY (§3.1): an update that changes `community` invalidates the + // WHOLE event. Not merely the community field — applying the content while + // keeping the old community would leave the first community's admission + // holding a CID it never evaluated, publishing content nobody judged under + // a standing acceptance. Retargeting a post means writing a new record. + // + // A skip, not an error: an invalid record from a stranger's repo is not an + // infrastructure failure, and dead-lettering it would retry a record that + // can never become valid. + if found && stored.communityDID != record.Community { + log.Printf("🚨 SECURITY: ignoring the whole %s update for %s - community is immutable (stored %s, incoming %s)", + PostV2Collection, uri, stored.communityDID, record.Community) + return nil + } + + // Provenance for bridgedStats is keyed on the AUTHOR's PDS now, because the + // record lives in the author's repo. The community's host has no say over + // what an author asserts about their own record any more, so checking the + // community's PDS (as the community-repo path does) would trust the wrong + // party entirely. An author this AppView holds no row for — the ordinary + // unhydrated federated author — has no provenance to prove, so the gate + // default-denies. + up, down, asOf := c.trustedBridgedStats(ctx, authorDID, record.BridgedStats, uri) + + facetsJSON, embedJSON, labelsJSON, err := serializePostContent( + sanitizeFacets(record.Facets, record.Content, uri), record.Embed, record.Labels) + if err != nil { + return err + } + + var applied bool + if found { + applied, err = c.applyPostContentUpdate(ctx, postContentUpdate{ + uri: uri, storedID: stored.id, rev: commit.Rev, cid: commit.CID, + title: record.Title, content: record.Content, + facets: facetsJSON, embed: embedJSON, labels: labelsJSON, + bridgedUpvotes: up, bridgedDownvotes: down, bridgedAsOf: asOf, + storedAsOf: stored.bridgedAsOf, storedDeletedAt: stored.deletedAt, + storedIndexedAt: stored.indexedAt, timeUS: timeUS, + }) + if err != nil { + return err + } + } else { + applied, err = c.insertAuthorPost(ctx, authorPostInsert{ + uri: uri, authorDID: authorDID, record: record, commit: commit, timeUS: timeUS, + facets: facetsJSON, embed: embedJSON, labels: labelsJSON, + bridgedUpvotes: up, bridgedDownvotes: down, bridgedAsOf: asOf, + }) + if err != nil { + return err + } + } + + if !applied { + // The rev gate or the recency guard refused this event: a newer state is + // already indexed. Opening or refreshing an admission from it would move + // evaluated_cid BACKWARDS onto content the row no longer holds, which is + // how an accepted post gets flipped to pending_reacceptance by a + // duplicate delivery. + return nil + } + + // The author is looked up only after the content is safely indexed, and a + // failure here is never fatal. Under §5.3 an author this AppView has never + // seen is a normal state that must index anyway, so hydration is an + // enrichment: getting a profile row for them is nice, and not getting one + // must not cost the post. + c.hydrateAuthorOpportunistically(ctx, authorDID) + + // The admission row: PENDING, always. The post claims the community; the + // community has said nothing. A row opened as anything else would publish + // speech the community never agreed to carry. + // + // It follows the content write rather than sharing its transaction, which + // leaves one bounded window: a failure between the two indexes the post + // without opening its admission, and the redrive is then rev-gated away, so + // the repair needs the record re-emitted. The alternative — opening the + // admission first — trades that for a phantom moderation-queue entry for a + // post that was never indexed, which is the worse of the two because it is + // invisible to the operator rather than visible as a dead letter. + if _, err := c.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: record.Community, + PostURI: uri, + EvaluatedCID: commit.CID, + }); err != nil { + return fmt.Errorf("recording the pending admission for %s in %s: %w", uri, record.Community, err) + } + + log.Printf("✓ Indexed author post: %s (author: %s, community: %s)", uri, authorDID, record.Community) + return nil +} + +// hydrateAuthorOpportunistically indexes a minimal profile for an author this +// AppView has not seen, so their posts are not permanently authorless. +// +// Bounded and non-fatal, both deliberately. Bounded because it is an outbound +// resolution on the hot firehose path driven by an identifier a stranger chose; +// non-fatal because §5.3 makes indexing the post the obligation and the profile +// merely an enrichment — refusing the event over a slow PLC lookup would +// reinstate exactly the refusal the flip removed. +func (c *PostEventConsumer) hydrateAuthorOpportunistically(ctx context.Context, authorDID string) { + if c.identityResolver == nil { + return + } + + if _, err := c.userService.GetUserByDID(ctx, authorDID); err == nil { + return // already indexed + } else if !errors.Is(err, users.ErrUserNotFound) { + log.Printf("debug: skipping author hydration for %s (lookup failed: %v)", authorDID, err) + return + } + + hydrateCtx, cancel := context.WithTimeout(ctx, authorHydrationTimeout) + defer cancel() + + resolved, err := c.identityResolver.Resolve(hydrateCtx, authorDID) + if err != nil || resolved == nil { + log.Printf("debug: could not resolve post author %s for hydration: %v", authorDID, err) + return + } + // The resolver returns a bidirectionally verified handle, or the reserved + // "handle.invalid" when verification failed. Indexing the latter would + // write a placeholder into a column with a uniqueness constraint, so the + // second unverifiable author would collide with the first. + if resolved.DID != authorDID || resolved.Handle == "" || resolved.Handle == invalidHandle { + log.Printf("debug: not hydrating post author %s (unverified identity)", authorDID) + return + } + + if err := c.userService.IndexUser(hydrateCtx, resolved.DID, resolved.Handle, resolved.PDSURL); err != nil { + log.Printf("debug: could not hydrate post author %s: %v", authorDID, err) + } +} + +// invalidHandle is the reserved handle atProto identity resolution reports when +// a DID's handle cannot be bidirectionally verified. +const invalidHandle = "handle.invalid" + +// authorHydrationTimeout bounds the opportunistic identity resolution above. +// Short on purpose: it is an enrichment on the firehose path, so it must never +// become the reason events back up. +const authorHydrationTimeout = 5 * time.Second + +// trustedBridgedStats returns the bridged aggregate to apply for an author-repo +// post, or a nil asOf meaning "leave the stored bridged columns alone". +// +// Default-deny at every step: no aggregate, no users row for the author, a PDS +// outside the trusted bridge set, or an aggregate failing input hygiene all +// return nothing to apply. +func (c *PostEventConsumer) trustedBridgedStats(ctx context.Context, authorDID string, stats *BridgedStatsFromJetstream, uri string) (int, int, *time.Time) { + if stats == nil { + return 0, 0, nil + } + + author, err := c.userService.GetUserByDID(ctx, authorDID) + if err != nil || author == nil { + log.Printf("debug: ignoring bridgedStats on %s (no indexed author %s to prove provenance)", uri, authorDID) + return 0, 0, nil + } + if !c.bridgeTrust.TrustsPDS(author.PDSURL) { + log.Printf("debug: ignoring bridgedStats on %s from untrusted author repo %s", uri, authorDID) + return 0, 0, nil + } + up, down, asOf, ok := validatedBridgedStats(stats, uri) + if !ok { + return 0, 0, nil + } + return up, down, &asOf +} + +// authorPostInsert is everything the first indexing of an author-repo post +// needs, gathered so the insert reads as one decision rather than a dozen +// positional arguments. +type authorPostInsert struct { + uri string + authorDID string + record *AuthorPostRecord + commit *CommitEvent + timeUS int64 + facets sql.NullString + embed sql.NullString + labels sql.NullString + bridgedUpvotes int + bridgedDownvotes int + bridgedAsOf *time.Time +} + +// insertAuthorPost indexes a post the AppView has never held. It reports +// whether the write applied — false means the rev gate refused the event. +func (c *PostEventConsumer) insertAuthorPost(ctx context.Context, in authorPostInsert) (bool, error) { + createdAt := parseRecordCreatedAt(in.record.CreatedAt, in.uri) + + post := &posts.Post{ + URI: in.uri, + CID: in.commit.CID, + RKey: in.commit.RKey, + // THE AUTHOR IS THE REPO. Not a field, not a lookup — deriving it from + // anywhere else is what would let any repo claim any author. + AuthorDID: in.authorDID, + // The community is a CLAIM the record makes: the author's submission + // target, which the community has not yet agreed to. + CommunityDID: in.record.Community, + Title: in.record.Title, + Content: in.record.Content, + ContentFacets: nullableString(in.facets), + Embed: nullableString(in.embed), + ContentLabels: nullableString(in.labels), + CreatedAt: createdAt, + IndexedAt: indexedAtForEvent(in.timeUS), + } + if in.bridgedAsOf != nil { + post.BridgedUpvoteCount = in.bridgedUpvotes + post.BridgedDownvoteCount = in.bridgedDownvotes + post.BridgedStatsAsOf = in.bridgedAsOf + post.Score = in.bridgedUpvotes - in.bridgedDownvotes + } + + // The rev gate decides inside this transaction, so a refusal and the writes + // it refuses can never half-apply. A gate skip surfaces as applied=false. + applied, err := c.indexPostIfRevWins(ctx, post, in.commit.Rev) + if err != nil { + return false, fmt.Errorf("failed to index author post %s: %w", in.uri, err) + } + return applied, nil +} + +// --------------------------------------------------------------------------- +// acceptance and removal: the community's decision records +// --------------------------------------------------------------------------- + +// communityDecisionRecord is an acceptance or a removal as it arrives from +// Jetstream. Both name their subject by strongRef; only a removal carries a +// code. +type communityDecisionRecord struct { + Type string `json:"$type"` + Subject struct { + URI string `json:"uri"` + CID string `json:"cid"` + } `json:"subject"` + Code string `json:"code,omitempty"` + Reason string `json:"reason,omitempty"` + CreatedAt string `json:"createdAt"` +} + +// parseCommunityDecision converts a raw acceptance/removal record, refusing the +// shapes that can never become valid. +func parseCommunityDecision(record map[string]interface{}, collection string) (*communityDecisionRecord, error) { + recordJSON, err := json.Marshal(record) + if err != nil { + return nil, fmt.Errorf("failed to marshal %s record: %w", collection, err) + } + + var parsed communityDecisionRecord + if err := json.Unmarshal(recordJSON, &parsed); err != nil { + return nil, fmt.Errorf("%w: failed to unmarshal %s record: %v", ErrPermanentEvent, collection, err) + } + if parsed.Subject.URI == "" { + return nil, fmt.Errorf("%w: %s record names no subject", ErrPermanentEvent, collection) + } + // The pinned CID is half the decision, not decoration: agreeing to a URI is + // not agreeing to whatever that URI holds later. A record without one gives + // the consumer nothing to compare against the indexed content. + if parsed.Subject.CID == "" { + return nil, fmt.Errorf("%w: %s record for %s pins no CID", ErrPermanentEvent, collection, parsed.Subject.URI) + } + if collection == posts.RemovalCollection && parsed.Code == "" { + return nil, fmt.Errorf("%w: removal record for %s carries no code", ErrPermanentEvent, parsed.Subject.URI) + } + return &parsed, nil +} + +// handleCommunityDecisionEvent routes one acceptance or removal commit. +// +// event.Did is the COMMUNITY, and that is the only thing in the event that says +// which community decided — the record names a post, not a decider. So the repo +// has to BE an indexed community: taking an arbitrary repo at its word would let +// anyone with a PDS publish into any feed by writing a record about someone +// else's post. +func (c *PostEventConsumer) handleCommunityDecisionEvent(ctx context.Context, event *JetstreamEvent, commit *CommitEvent) error { + communityDID := event.Did + + if _, err := c.communityRepo.GetByDID(ctx, communityDID); err != nil { + if communities.IsNotFound(err) { + // Transient, and the reason is delivery order rather than leniency: + // a community's first acceptance can genuinely outrun its own + // profile event, and marking this permanent would spend the redrive + // budget that resolves the race and discard a real decision. + log.Printf("🚨 SECURITY: refusing %s from %s - not an indexed community repo", + commit.Collection, communityDID) + return fmt.Errorf("community not found: %s - cannot apply a %s from a repo that is not an indexed community", + communityDID, commit.Collection) + } + return fmt.Errorf("%w: failed to verify community %s exists: %v", errValidationInfra, communityDID, err) + } + + if commit.Operation == "delete" { + return c.applyCommunityDecisionDelete(ctx, communityDID, commit) + } + + if commit.Record == nil { + return fmt.Errorf("%w: %s %s event missing record data", ErrPermanentEvent, commit.Collection, commit.Operation) + } + decision, err := parseCommunityDecision(commit.Record, commit.Collection) + if err != nil { + return err + } + + // The subject's author, and therefore the erasure gate. An admission row + // for an erased account's post is exactly the row migration 036 exists to + // stop being recreated, and an acceptance replayed months later is one of + // the two ways it comes back. + subjectAuthor, subjectCollection, _, ok := parseRecordURI(decision.Subject.URI) + if !ok { + return fmt.Errorf("%w: %s names subject %q, which is not an at:// record URI", + ErrPermanentEvent, commit.Collection, decision.Subject.URI) + } + erased, err := c.authorWasErased(ctx, subjectAuthor) + if err != nil { + return err + } + if erased { + log.Printf("INFO: dropping %s %s about %s - its author was erased", + commit.Collection, commit.Operation, decision.Subject.URI) + return nil + } + + watermark := posts.CommunityWatermark{Rev: commit.Rev} + + switch commit.Collection { + case posts.AcceptanceCollection: + return c.applyAcceptance(ctx, communityDID, commit, decision, subjectCollection, watermark) + case posts.RemovalCollection: + return c.applyRemoval(ctx, communityDID, decision, watermark) + } + return nil +} + +// applyAcceptance records a community's agreement to exactly one version of one +// post, converging on the subject first when the AppView has never seen it. +func (c *PostEventConsumer) applyAcceptance( + ctx context.Context, + communityDID string, + commit *CommitEvent, + decision *communityDecisionRecord, + subjectCollection string, + watermark posts.CommunityWatermark, +) error { + indexedCommunity, indexed, err := c.indexedPostCommunity(ctx, decision.Subject.URI) + if err != nil { + return err + } + switch { + case !indexed: + if err := c.convergeOnAcceptedSubject(ctx, communityDID, decision, subjectCollection); err != nil { + return err + } + case indexedCommunity != communityDID: + // The same refusal the fetch path makes, on the path where the post is + // already indexed. Both are the §10.2 rule: a community accepting a + // post that names a DIFFERENT community is the fork/import flow, which + // the data model supports and nothing is built for — so today it is a + // community pulling another community's content into its feed on its + // own say-so. Enforcing it in only one of the two places would leave + // the whole check bypassable by getting the post indexed first, which + // an attacker controls: they simply post before they accept. + // + // PERMANENT: a post's community is immutable across updates (§3.1), so + // no retry makes this valid. + return fmt.Errorf("%w: %s was submitted to community %s, but the acceptance came from %s (the fork/import flow is not built)", + ErrPermanentEvent, decision.Subject.URI, indexedCommunity, communityDID) + } + + result, err := c.admissions.ApplyAcceptance(ctx, posts.ApplyAcceptanceCommand{ + CommunityDID: communityDID, + PostURI: decision.Subject.URI, + AcceptanceURI: recordURI(communityDID, posts.AcceptanceCollection, commit.RKey), + AcceptanceRkey: commit.RKey, + PinnedCID: decision.Subject.CID, + Watermark: watermark, + }) + if err != nil { + return fmt.Errorf("applying the acceptance of %s in %s: %w", decision.Subject.URI, communityDID, err) + } + logAdmissionOutcome(posts.AcceptanceCollection, communityDID, decision.Subject.URI, result.Outcome) + return nil +} + +// convergeOnAcceptedSubject reads an accepted post straight from its author's +// PDS and indexes it (§5.4). +// +// Redrive alone cannot solve acceptance-before-post: bounded retries cannot +// manufacture an event that a relay-coverage gap will never deliver. This is +// the mechanism that makes convergence a guarantee rather than a bet on full +// relay coverage — and because it is an outbound request whose destination is +// chosen by a stranger's record, most of what follows is refusals. +func (c *PostEventConsumer) convergeOnAcceptedSubject( + ctx context.Context, + communityDID string, + decision *communityDecisionRecord, + subjectCollection string, +) error { + if subjectCollection != PostV2Collection { + // An acceptance is about an author-repo post. A subject in any other + // collection is not a thing this community can accept, and no retry + // changes which collection a URI names. + return fmt.Errorf("%w: acceptance names subject %s, which is not a %s record", + ErrPermanentEvent, decision.Subject.URI, PostV2Collection) + } + if c.postFetcher == nil { + // Without the fetch the only convergence mechanism left is redrive, so + // the event must stay retryable rather than be dropped. + return fmt.Errorf("acceptance for unindexed post %s: no direct fetcher is wired, so only redrive can converge", + decision.Subject.URI) + } + + fetched, err := c.postFetcher.FetchPost(ctx, decision.Subject.URI) + if err != nil { + // Transient: a PDS that is down, slow, or briefly unreachable is the + // ordinary case, and the redrive is what it is for. + return fmt.Errorf("fetching the accepted post %s directly: %w", decision.Subject.URI, err) + } + + // THE CID CHECK IS WHAT MAKES THE FETCH TRUSTWORTHY AT ALL. Without it the + // AppView indexes whatever the author's PDS chooses to serve under that + // rkey — the author (or whoever holds their keys) picks the content, and + // the community's signed acceptance is made to cover it retroactively. + // + // PERMANENT: the pinned version is gone from the repo and no retry brings + // it back, so re-fetching the same mismatch ten times is pure noise. + if fetched.CID != decision.Subject.CID { + return fmt.Errorf("%w: the PDS serving %s returned CID %s, but the acceptance pinned %s", + ErrPermanentEvent, decision.Subject.URI, fetched.CID, decision.Subject.CID) + } + + record, err := parseAuthorPostRecord(fetched.Record) + if err != nil { + return err + } + + // Cross-community acceptance is the privileged fork/import flow, and §10.2 + // is explicit that it is deliberately NOT built. Until it exists, a + // community accepting a post that names someone else is a community pulling + // another community's content into its feed on its own say-so. + // + // PERMANENT: the record's community field is immutable across updates + // (§3.1), so this can never become valid. + if record.Community != communityDID { + return fmt.Errorf("%w: %s names community %s, but the acceptance came from %s (the fork/import flow is not built)", + ErrPermanentEvent, decision.Subject.URI, record.Community, communityDID) + } + + authorDID, _, rkey, _ := parseRecordURI(decision.Subject.URI) + + facetsJSON, embedJSON, labelsJSON, err := serializePostContent( + sanitizeFacets(record.Facets, record.Content, decision.Subject.URI), record.Embed, record.Labels) + if err != nil { + return err + } + up, down, asOf := c.trustedBridgedStats(ctx, authorDID, record.BridgedStats, decision.Subject.URI) + + // The fetch is not a firehose event, so it carries no rev to gate on and no + // event time to stamp: an empty rev bypasses the gate (which is correct — a + // later real event for this record still wins on its own rev) and the + // watermark falls back to wall clock. + if _, err := c.insertAuthorPost(ctx, authorPostInsert{ + uri: decision.Subject.URI, + authorDID: authorDID, + record: record, + commit: &CommitEvent{ + Operation: "create", Collection: PostV2Collection, + RKey: rkey, CID: fetched.CID, + }, + facets: facetsJSON, embed: embedJSON, labels: labelsJSON, + bridgedUpvotes: up, bridgedDownvotes: down, bridgedAsOf: asOf, + }); err != nil { + return err + } + + // The pending admission comes with it. ApplyAcceptance would create a row + // on its own, but one with no evaluated content: recording what was indexed + // is what lets the next author edit be recognised as an edit. + if _, err := c.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: communityDID, + PostURI: decision.Subject.URI, + EvaluatedCID: fetched.CID, + }); err != nil { + return fmt.Errorf("recording the pending admission for fetched post %s: %w", decision.Subject.URI, err) + } + + c.hydrateAuthorOpportunistically(ctx, authorDID) + log.Printf("✓ Converged on accepted post %s by direct fetch (author: %s, community: %s)", + decision.Subject.URI, authorDID, communityDID) + return nil +} + +// applyRemoval records a community's moderation decision about a post. +// +// A removal with no prior acceptance is VALID: a community that has decided in +// advance about a post — an author it is about to ban, content it has already +// seen elsewhere — must be able to say so, and requiring an acceptance first +// would drop exactly the decisions a community most wants to make early. No +// direct fetch either: a removed post is not rendered, so there is nothing to +// converge on, and fetching content in order to hide it would hand a moderation +// record the power to make the AppView dial an arbitrary host. +func (c *PostEventConsumer) applyRemoval( + ctx context.Context, + communityDID string, + decision *communityDecisionRecord, + watermark posts.CommunityWatermark, +) error { + result, err := c.admissions.ApplyRemoval(ctx, posts.ApplyRemovalCommand{ + CommunityDID: communityDID, + PostURI: decision.Subject.URI, + DecisionCode: decision.Code, + Watermark: watermark, + }) + if err != nil { + return fmt.Errorf("applying the removal of %s in %s: %w", decision.Subject.URI, communityDID, err) + } + logAdmissionOutcome(posts.RemovalCollection, communityDID, decision.Subject.URI, result.Outcome) + return nil +} + +// applyCommunityDecisionDelete applies the withdrawal of an acceptance or a +// removal. +// +// A delete event carries NO record, so the subject cannot be read from it — and +// the rkey is a SHA-256 digest of the subject URI (§3.2), which is one-way. The +// subject is therefore recoverable only from state the AppView already holds: +// acceptance_rkey, which is stored exactly while an acceptance stands, i.e. +// exactly when an acceptance deletion has something to withdraw. +// +// A removal deletion has no such column and needs none. Every moderation commit +// is a PAIR (§3.3) — the removal commit is {acceptance-delete, removal-put} and +// the restore commit is {removal-delete, acceptance-put} — and the put half +// carries the subject in-record and outranks its paired delete under the §5.2 +// tuple. So the put alone converges the row whichever half arrives first, and +// an unresolvable delete is a no-op rather than a lost transition. The lone +// acceptance deletion, which the host writes when an author deletes their post +// (§5.3), is the one delete that arrives unpaired — and it is the one this +// lookup resolves. +func (c *PostEventConsumer) applyCommunityDecisionDelete(ctx context.Context, communityDID string, commit *CommitEvent) error { + if commit.Collection != posts.AcceptanceCollection { + log.Printf("INFO: %s deletion in %s carries no subject and is superseded by its paired write; skipping", + commit.Collection, communityDID) + return nil + } + + var postURI string + err := c.db.QueryRowContext(ctx, + `SELECT post_uri FROM community_post_admissions + WHERE community_did = $1 AND acceptance_rkey = $2`, + communityDID, commit.RKey, + ).Scan(&postURI) + if errors.Is(err, sql.ErrNoRows) { + // No acceptance of that rkey stands here — the removal half of the same + // commit already cleared it, or this AppView never saw the acceptance. + // Either way there is nothing to withdraw. + log.Printf("INFO: acceptance deletion %s/%s matches no standing acceptance; nothing to withdraw", + communityDID, commit.RKey) + return nil + } + if err != nil { + return fmt.Errorf("resolving the subject of acceptance deletion %s/%s: %w", communityDID, commit.RKey, err) + } + + result, err := c.admissions.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + CommunityDID: communityDID, + PostURI: postURI, + Watermark: posts.CommunityWatermark{Rev: commit.Rev}, + }) + if err != nil { + return fmt.Errorf("withdrawing the acceptance of %s in %s: %w", postURI, communityDID, err) + } + logAdmissionOutcome(posts.AcceptanceCollection+"#delete", communityDID, postURI, result.Outcome) + return nil +} + +// logAdmissionOutcome records what a community event DID, including the skips. +// +// A skip is the ordering gate working — a multi-feed duplicate, a dead-letter +// redrive, an event superseded by its own commit's other half — so it is logged +// rather than returned as an error, which would bury healthy skips in the +// dead-letter queue among genuine failures. +func logAdmissionOutcome(collection, communityDID, postURI string, outcome posts.AdmissionOutcome) { + if outcome == posts.AdmissionApplied { + log.Printf("✓ Applied %s for %s in %s", collection, postURI, communityDID) + return + } + log.Printf("admission: %s for %s in %s was %s (an outcome, not a failure)", + collection, postURI, communityDID, outcome) +} + +// --------------------------------------------------------------------------- +// small shared helpers +// --------------------------------------------------------------------------- + +// recordURI builds the AT-URI of a record from its repo, collection and rkey. +func recordURI(repoDID, collection, rkey string) string { + return fmt.Sprintf("at://%s/%s/%s", repoDID, collection, rkey) +} + +// parseRecordURI splits at:////. It reports ok=false +// for anything else, including URIs with extra path segments — a subject the +// AppView cannot address is a subject it must refuse rather than guess at. +func parseRecordURI(uri string) (repoDID, collection, rkey string, ok bool) { + rest, found := strings.CutPrefix(uri, "at://") + if !found { + return "", "", "", false + } + parts := strings.Split(rest, "/") + if len(parts) != 3 || parts[0] == "" || parts[1] == "" || parts[2] == "" { + return "", "", "", false + } + return parts[0], parts[1], parts[2], true +} + +// indexedPostCommunity returns the community an indexed post was submitted to. +// +// Soft-deleted rows COUNT as indexed: a tombstoned post has been seen, and +// treating it as absent would send the direct fetch to resurrect content its +// author deleted. +func (c *PostEventConsumer) indexedPostCommunity(ctx context.Context, uri string) (string, bool, error) { + var communityDID string + err := c.db.QueryRowContext(ctx, `SELECT community_did FROM posts WHERE uri = $1`, uri).Scan(&communityDID) + if errors.Is(err, sql.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("reading the indexed community of %s: %w", uri, err) + } + return communityDID, true, nil +} + +// nullableString converts a serialized JSON column back to the pointer shape +// posts.Post uses, where nil means "the record carried none". +func nullableString(v sql.NullString) *string { + if !v.Valid { + return nil + } + s := v.String + return &s } diff --git a/internal/atproto/jetstream/feeds.go b/internal/atproto/jetstream/feeds.go --- a/internal/atproto/jetstream/feeds.go +++ b/internal/atproto/jetstream/feeds.go @@ -63,8 +63,21 @@ "social.coves.community.profile", "social.coves.community.subscription", "social.coves.community.block", }, + // One consumer for all four post-related collections, because they decide + // about each other: an acceptance is meaningless without the postv2 it + // pins, and both write the same admission row. Splitting them across + // connectors would give the two halves independent cursors and independent + // dead letters for one conversation. + // + // social.coves.community.post stays subscribed even though it is DEPRECATED + // (§3.0): the records already written to community repos keep arriving, and + // dropping the filter would silently stop indexing edits and deletes of + // every post that exists today. ConsumerPosts: { "social.coves.community.post", + "social.coves.community.postv2", + "social.coves.community.acceptance", + "social.coves.community.removal", }, ConsumerAggregators: { "social.coves.aggregator.service", diff --git a/internal/atproto/jetstream/post_consumer.go b/internal/atproto/jetstream/post_consumer.go --- a/internal/atproto/jetstream/post_consumer.go +++ b/internal/atproto/jetstream/post_consumer.go @@ -94,8 +94,10 @@ } commit := event.Commit - // Handle post record operations - if commit.Collection == "social.coves.community.post" { + switch commit.Collection { + // The DEPRECATED community-repo post (§3.0). Here the repo DID must EQUAL + // the record's community; the three collections below invert that. + case "social.coves.community.post": switch commit.Operation { case "create": return c.createPost(ctx, event.Did, commit, event.TimeUS) @@ -104,6 +106,22 @@ return c.updatePost(ctx, event.Did, commit, event.TimeUS) case "delete": return c.deletePost(ctx, event.Did, commit) } + + // The author-repo post: event.Did IS the author, and the community is a + // claim the record makes (authorpost.go). + case PostV2Collection: + if !c.canRecordAdmissions(commit.Collection) { + return nil + } + return c.handleAuthorPostEvent(ctx, event, commit) + + // The community's decision records: event.Did IS the community, and the + // post is a subject the record names (authorpost.go). + case posts.AcceptanceCollection, posts.RemovalCollection: + if !c.canRecordAdmissions(commit.Collection) { + return nil + } + return c.handleCommunityDecisionEvent(ctx, event, commit) } // Silently ignore other operations and other collections @@ -161,22 +179,7 @@ // Build AT-URI for this post // Format: at://community_did/social.coves.community.post/rkey uri := fmt.Sprintf("at://%s/social.coves.community.post/%s", repoDID, commit.RKey) - // Parse timestamp from record - createdAt, err := time.Parse(time.RFC3339, postRecord.CreatedAt) - if err != nil { - // Fallback to current time if parsing fails - log.Printf("Warning: Failed to parse createdAt timestamp, using current time: %v", err) - createdAt = time.Now() - } - - // SECURITY: Clamp future timestamps to now. created_at drives the "new" sort - // and the hot-rank age, so a record asserting a future date (hostile or - // clock-skewed federated repo) could otherwise pin itself to the top of - // feeds until wall-clock catches up. - if now := time.Now(); createdAt.After(now) { - log.Printf("Warning: post %s has future createdAt %s, clamping to now", uri, postRecord.CreatedAt) - createdAt = now - } + createdAt := parseRecordCreatedAt(postRecord.CreatedAt, uri) // Build post entity post := &posts.Post{ @@ -218,36 +221,17 @@ } // Serialize JSON fields (facets, embed, labels) // Return error if any non-empty field fails to serialize (prevents silent data loss) - postRecord.Facets = sanitizedPostFacets(postRecord, uri) - if postRecord.Facets != nil { - facetsJSON, marshalErr := json.Marshal(postRecord.Facets) - if marshalErr != nil { - return fmt.Errorf("failed to serialize facets: %w", marshalErr) - } - facetsStr := string(facetsJSON) - post.ContentFacets = &facetsStr - } - - if postRecord.Embed != nil { - embedJSON, marshalErr := json.Marshal(postRecord.Embed) - if marshalErr != nil { - return fmt.Errorf("failed to serialize embed: %w", marshalErr) - } - embedStr := string(embedJSON) - post.Embed = &embedStr + facetsJSON, embedJSON, labelsJSON, err := serializePostContent( + sanitizedPostFacets(postRecord, uri), postRecord.Embed, postRecord.Labels) + if err != nil { + return err } - - if postRecord.Labels != nil { - labelsJSON, marshalErr := json.Marshal(postRecord.Labels) - if marshalErr != nil { - return fmt.Errorf("failed to serialize labels: %w", marshalErr) - } - labelsStr := string(labelsJSON) - post.ContentLabels = &labelsStr - } + post.ContentFacets = nullableString(facetsJSON) + post.Embed = nullableString(embedJSON) + post.ContentLabels = nullableString(labelsJSON) // Atomically: Rev-gate + Index post + Reconcile comment count for out-of-order arrivals - if err := c.indexPostAndReconcileCounts(ctx, post, commit.Rev); err != nil { + if _, err := c.indexPostIfRevWins(ctx, post, commit.Rev); err != nil { return fmt.Errorf("failed to index post and reconcile counts: %w", err) } @@ -259,10 +243,18 @@ // deletePost handles post deletion events from Jetstream // Soft-deletes the post in AppView database by setting deleted_at timestamp func (c *PostEventConsumer) deletePost(ctx context.Context, repoDID string, commit *CommitEvent) error { - // Build AT-URI for this post // Format: at://community_did/social.coves.community.post/rkey - uri := fmt.Sprintf("at://%s/social.coves.community.post/%s", repoDID, commit.RKey) + return c.tombstoneRecord(ctx, fmt.Sprintf("at://%s/social.coves.community.post/%s", repoDID, commit.RKey), commit.Rev) +} +// tombstoneRecord soft-deletes the post at uri under the rev gate. +// +// SOFT, never hard, whichever repo the record lived in: the row is the rev +// gate's tombstone, the comment thread's parent, and what moderation still +// reads. It is shared by the community-repo and author-repo delete paths +// because a deletion is the one operation where the two are identical — the +// URI already says whose repo it was. +func (c *PostEventConsumer) tombstoneRecord(ctx context.Context, uri, rev string) error { // REV GATE + soft delete in one transaction (the repo's SoftDelete is not // transaction-aware, and the delete's rev must be recorded atomically with // the tombstone: it is what rejects a stale cross-feed copy of the CREATE @@ -279,12 +271,12 @@ log.Printf("Failed to rollback transaction: %v", rollbackErr) } }() - won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + won, err := tryAdvanceRecordRev(ctx, tx, uri, rev) if err != nil { return err } if !won { - logSkippedStaleRev(ConsumerPosts, "delete", uri, commit.Rev) + logSkippedStaleRev(ConsumerPosts, "delete", uri, rev) return nil } @@ -300,7 +292,7 @@ if err := tx.Commit(); err != nil { return fmt.Errorf("failed to commit post delete transaction: %w", err) } - log.Printf("✓ Deleted post: %s (community: %s, rkey: %s)", uri, repoDID, commit.RKey) + log.Printf("✓ Deleted post: %s", uri) return nil } @@ -340,78 +332,28 @@ uri := fmt.Sprintf("at://%s/social.coves.community.post/%s", repoDID, commit.RKey) // Fetch the stored row so we can enforce immutability and run the asOf regression guard. - var ( - storedID int64 - storedCommunityDID string - storedAuthorDID string - storedDeletedAt *time.Time - storedAsOf *time.Time - storedIndexedAt time.Time - ) - err = c.db.QueryRowContext(ctx, - `SELECT id, community_did, author_did, deleted_at, bridged_stats_as_of, indexed_at FROM posts WHERE uri = $1`, - uri, - ).Scan(&storedID, &storedCommunityDID, &storedAuthorDID, &storedDeletedAt, &storedAsOf, &storedIndexedAt) - if errors.Is(err, sql.ErrNoRows) { + stored, found, err := c.loadStoredPost(ctx, uri) + if err != nil { + return err + } + if !found { // Not indexed yet (out-of-order delivery). Jetstream will replay CREATE; skip. log.Printf("Update event for non-indexed post: %s (will be indexed on CREATE)", uri) return nil - } - if err != nil { - return fmt.Errorf("failed to load stored post for update: %w", err) - } - - // Skip soft-deleted rows: a deleted post should not be resurrected by an edit. - if storedDeletedAt != nil { - log.Printf("Update event for soft-deleted post: %s (skipping)", uri) - return nil - } - - // RECENCY GUARD: a redriven (DeadLetterRedriver) or rewound update can arrive - // AFTER a newer update was already indexed. indexed_at is the watermark of the - // last applied event for this row (event time, see indexedAtForEvent); an event - // whose time_us is not strictly newer must be skipped, or a stale replay would - // silently revert newer content. Skipping is SUCCESS (the newer state wins) — - // returning an error would re-dead-letter an event that must never be applied. - // This Go pre-check exists for clean logging; the UPDATE below repeats the - // comparison atomically so a concurrent newer write between this read and the - // write still cannot be clobbered. - if evTime, ok := eventTime(timeUS); ok && !storedIndexedAt.Before(evTime) { - log.Printf("INFO: skipping stale post update for %s (event time %s <= last indexed %s; newer state already applied)", - uri, evTime.Format(time.RFC3339Nano), storedIndexedAt.Format(time.RFC3339Nano)) - return nil } // SECURITY: community and author are immutable. Reassignment is rejected (skipped). - if storedCommunityDID != postRecord.Community || storedAuthorDID != postRecord.Author { + if stored.communityDID != postRecord.Community || stored.authorDID != postRecord.Author { log.Printf("🚨 SECURITY: Rejecting post update - community/author reassignment is not allowed: %s (stored community=%s author=%s; incoming community=%s author=%s)", - uri, storedCommunityDID, storedAuthorDID, postRecord.Community, postRecord.Author) + uri, stored.communityDID, stored.authorDID, postRecord.Community, postRecord.Author) return nil } // Serialize optional JSON content fields (return on failure to avoid silent data loss). - var facetsJSON, embedJSON, labelsJSON sql.NullString - postRecord.Facets = sanitizedPostFacets(postRecord, uri) - if postRecord.Facets != nil { - b, marshalErr := json.Marshal(postRecord.Facets) - if marshalErr != nil { - return fmt.Errorf("failed to serialize facets: %w", marshalErr) - } - facetsJSON.String, facetsJSON.Valid = string(b), true - } - if postRecord.Embed != nil { - b, marshalErr := json.Marshal(postRecord.Embed) - if marshalErr != nil { - return fmt.Errorf("failed to serialize embed: %w", marshalErr) - } - embedJSON.String, embedJSON.Valid = string(b), true - } - if postRecord.Labels != nil { - b, marshalErr := json.Marshal(postRecord.Labels) - if marshalErr != nil { - return fmt.Errorf("failed to serialize labels: %w", marshalErr) - } - labelsJSON.String, labelsJSON.Valid = string(b), true + facetsJSON, embedJSON, labelsJSON, err := serializePostContent( + sanitizedPostFacets(postRecord, uri), postRecord.Embed, postRecord.Labels) + if err != nil { + return err } // Decide the candidate bridged aggregate to hand to the atomic UPDATE. It is applied @@ -431,20 +373,124 @@ if postRecord.BridgedStats != nil { if c.bridgeTrust.TrustsPDS(community.PDSURL) { if up, down, asOf, ok := validatedBridgedStats(postRecord.BridgedStats, uri); ok { incomingUp, incomingDown, incomingAsOf = up, down, &asOf - // Best-effort log only (the write is authoritative and atomic): a - // strictly-older asOf is dropped by the SQL guard. Kept at debug because - // the bridge re-sends the same asOf on every content edit, so this is - // noise, not an anomaly. - if storedAsOf != nil && asOf.Before(*storedAsOf) { - log.Printf("debug: ignoring strictly-older bridgedStats for %s (incoming asOf %s < stored %s)", - uri, asOf.Format(time.RFC3339), storedAsOf.Format(time.RFC3339)) - } } } else { log.Printf("debug: ignoring bridgedStats on post %s from untrusted repo %s (not a trusted bridge PDS)", uri, repoDID) } } + if _, err := c.applyPostContentUpdate(ctx, postContentUpdate{ + uri: uri, storedID: stored.id, rev: commit.Rev, cid: commit.CID, + title: postRecord.Title, content: postRecord.Content, + facets: facetsJSON, embed: embedJSON, labels: labelsJSON, + bridgedUpvotes: incomingUp, bridgedDownvotes: incomingDown, bridgedAsOf: incomingAsOf, + storedAsOf: stored.bridgedAsOf, storedDeletedAt: stored.deletedAt, + storedIndexedAt: stored.indexedAt, timeUS: timeUS, + }); err != nil { + return err + } + return nil +} + +// storedPost is the slice of an indexed post row the write paths need: the +// identity to update, the columns immutability is checked against, and the two +// watermarks (bridged asOf, indexed_at) the guards compare. +type storedPost struct { + id int64 + communityDID string + authorDID string + deletedAt *time.Time + bridgedAsOf *time.Time + indexedAt time.Time +} + +// loadStoredPost reads the row for uri. found=false means the post has never +// been indexed, which is an ordinary out-of-order arrival rather than an error. +func (c *PostEventConsumer) loadStoredPost(ctx context.Context, uri string) (storedPost, bool, error) { + var stored storedPost + err := c.db.QueryRowContext(ctx, + `SELECT id, community_did, author_did, deleted_at, bridged_stats_as_of, indexed_at FROM posts WHERE uri = $1`, + uri, + ).Scan(&stored.id, &stored.communityDID, &stored.authorDID, + &stored.deletedAt, &stored.bridgedAsOf, &stored.indexedAt) + if errors.Is(err, sql.ErrNoRows) { + return storedPost{}, false, nil + } + if err != nil { + return storedPost{}, false, fmt.Errorf("failed to load stored post %s: %w", uri, err) + } + return stored, true, nil +} + +// postContentUpdate is one already-validated edit of an indexed post. +// +// It exists so the community-repo and author-repo paths share ONE content +// write. What differs between them is who may claim what — the repo/community +// check inverts, and bridgedStats provenance keys on a different repo — and all +// of that is settled by the caller before it gets here. What does not differ is +// how an edit is applied: the same rev gate, the same recency guard, the same +// atomic bridged-stats regression rule. Two copies of that would drift. +type postContentUpdate struct { + uri string + storedID int64 + rev string + cid string + + title *string + content *string + facets sql.NullString + embed sql.NullString + labels sql.NullString + + bridgedUpvotes int + bridgedDownvotes int + // bridgedAsOf nil means "leave the stored bridged columns alone". + bridgedAsOf *time.Time + + storedAsOf *time.Time + storedDeletedAt *time.Time + storedIndexedAt time.Time + timeUS int64 +} + +// applyPostContentUpdate runs the rev gate and the atomic content UPDATE. +// +// It reports whether the write APPLIED. A false with no error is a skip — the +// stored row already holds a newer state — and every skip here is the system +// working: multi-feed duplicates, dead-letter redrives, and edits of posts +// deleted between the load and the write all land in it. Returning any of them +// as an error would dead-letter healthy events. +func (c *PostEventConsumer) applyPostContentUpdate(ctx context.Context, in postContentUpdate) (bool, error) { + // Skip soft-deleted rows: a deleted post should not be resurrected by an edit. + if in.storedDeletedAt != nil { + log.Printf("Update event for soft-deleted post: %s (skipping)", in.uri) + return false, nil + } + + // RECENCY GUARD: a redriven (DeadLetterRedriver) or rewound update can arrive + // AFTER a newer update was already indexed. indexed_at is the watermark of the + // last applied event for this row (event time, see indexedAtForEvent); an event + // whose time_us is not strictly newer must be skipped, or a stale replay would + // silently revert newer content. Skipping is SUCCESS (the newer state wins) — + // returning an error would re-dead-letter an event that must never be applied. + // This Go pre-check exists for clean logging; the UPDATE below repeats the + // comparison atomically so a concurrent newer write between this read and the + // write still cannot be clobbered. + if evTime, ok := eventTime(in.timeUS); ok && !in.storedIndexedAt.Before(evTime) { + log.Printf("INFO: skipping stale post update for %s (event time %s <= last indexed %s; newer state already applied)", + in.uri, evTime.Format(time.RFC3339Nano), in.storedIndexedAt.Format(time.RFC3339Nano)) + return false, nil + } + + // Best-effort log only (the write is authoritative and atomic): a + // strictly-older asOf is dropped by the SQL guard. Kept at debug because + // the bridge re-sends the same asOf on every content edit, so this is + // noise, not an anomaly. + if in.bridgedAsOf != nil && in.storedAsOf != nil && in.bridgedAsOf.Before(*in.storedAsOf) { + log.Printf("debug: ignoring strictly-older bridgedStats for %s (incoming asOf %s < stored %s)", + in.uri, in.bridgedAsOf.Format(time.RFC3339), in.storedAsOf.Format(time.RFC3339)) + } + // Single atomic UPDATE. edited_at is bumped only when content actually changed (so a // debounced stats-only refresh does not mark the post edited). The bridged columns // and the inclusive score move together via a shared applies-guard: apply the @@ -502,7 +548,7 @@ // lagging bsky feed carries a NEWER time_us than the edit it would regress. // Only rev, assigned by the repo itself, orders events across feeds. tx, err := c.db.BeginTx(ctx, nil) if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) + return false, fmt.Errorf("failed to begin transaction: %w", err) } defer func() { if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { @@ -510,23 +556,23 @@ log.Printf("Failed to rollback transaction: %v", rollbackErr) } }() - won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + won, err := tryAdvanceRecordRev(ctx, tx, in.uri, in.rev) if err != nil { - return err + return false, err } if !won { - logSkippedStaleRev(ConsumerPosts, "update", uri, commit.Rev) - return nil + logSkippedStaleRev(ConsumerPosts, "update", in.uri, in.rev) + return false, nil } result, err := tx.ExecContext(ctx, updateQuery, - storedID, commit.CID, postRecord.Title, postRecord.Content, - facetsJSON, embedJSON, labelsJSON, - incomingUp, incomingDown, incomingAsOf, - timeUS, + in.storedID, in.cid, in.title, in.content, + in.facets, in.embed, in.labels, + in.bridgedUpvotes, in.bridgedDownvotes, in.bridgedAsOf, + in.timeUS, ) if err != nil { - return fmt.Errorf("failed to update post: %w", err) + return false, fmt.Errorf("failed to update post: %w", err) } // A post can be soft-deleted — or overtaken by a concurrent NEWER update (recency @@ -536,25 +582,26 @@ // (mirrors vote_consumer's RowsAffected check). Both cases are success: the row's // current state supersedes this event. rowsAffected, err := result.RowsAffected() if err != nil { - return fmt.Errorf("failed to check post update result: %w", err) + return false, fmt.Errorf("failed to check post update result: %w", err) } if rowsAffected == 0 { // The deferred rollback also reverts the gate advance — conservative: a // replay re-evaluates against whatever state superseded this event. - log.Printf("Update event for post that was deleted or superseded by a newer update between load and write: %s (skipping)", uri) - return nil + log.Printf("Update event for post that was deleted or superseded by a newer update between load and write: %s (skipping)", in.uri) + return false, nil } if err := tx.Commit(); err != nil { - return fmt.Errorf("failed to commit post update transaction: %w", err) + return false, fmt.Errorf("failed to commit post update transaction: %w", err) } - if incomingAsOf != nil { - log.Printf("✓ Updated post: %s (bridgedStats candidate applied if newer-or-equal: up=%d down=%d)", uri, incomingUp, incomingDown) + if in.bridgedAsOf != nil { + log.Printf("✓ Updated post: %s (bridgedStats candidate applied if newer-or-equal: up=%d down=%d)", + in.uri, in.bridgedUpvotes, in.bridgedDownvotes) } else { - log.Printf("✓ Updated post: %s", uri) + log.Printf("✓ Updated post: %s", in.uri) } - return nil + return true, nil } // parseBridgedAsOf parses a bridgedStats.asOf timestamp, logging (and returning the @@ -568,12 +615,17 @@ } return t, nil } -// indexPostAndReconcileCounts atomically indexes a post and reconciles comment counts -// This fixes the race condition where comments arrive before their parent post -func (c *PostEventConsumer) indexPostAndReconcileCounts(ctx context.Context, post *posts.Post, rev string) error { +// indexPostIfRevWins atomically indexes a post and reconciles comment counts. +// This fixes the race condition where comments arrive before their parent post. +// +// It reports whether the insert APPLIED: false means the rev gate refused the +// event, or the row already existed. Callers that must not act on content they +// did not write — the author-repo path, which opens an admission from the CID +// it just indexed — read that flag rather than assuming the write happened. +func (c *PostEventConsumer) indexPostIfRevWins(ctx context.Context, post *posts.Post, rev string) (bool, error) { tx, err := c.db.BeginTx(ctx, nil) if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) + return false, fmt.Errorf("failed to begin transaction: %w", err) } defer func() { if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { @@ -588,11 +640,11 @@ // (which would resurrect it). Runs first, inside the transaction, so gate // and writes commit or roll back together. won, err := tryAdvanceRecordRev(ctx, tx, post.URI, rev) if err != nil { - return err + return false, err } if !won { logSkippedStaleRev(ConsumerPosts, "create", post.URI, rev) - return nil + return false, nil } // 1. Insert the post (idempotent with RETURNING clause) @@ -654,13 +706,15 @@ // (comments implement the in-place re-create because their resurrection // machinery already exists; see comment_consumer.go). log.Printf("Post already indexed: %s (idempotent)", post.URI) if commitErr := tx.Commit(); commitErr != nil { - return fmt.Errorf("failed to commit transaction: %w", commitErr) + return false, fmt.Errorf("failed to commit transaction: %w", commitErr) } - return nil + // Reported as NOT applied: no content was written, so a caller that + // would record what it just indexed has nothing new to record. + return false, nil } if insertErr != nil { - return fmt.Errorf("failed to insert post: %w", insertErr) + return false, fmt.Errorf("failed to insert post: %w", insertErr) } // 2. Reconcile comment_count for this newly inserted post @@ -689,15 +743,15 @@ if reconcileErr != nil { // Reconciliation failure is a critical error - it means comment_count will be incorrect // This could cause data inconsistency where the displayed count doesn't match reality // Roll back the transaction to maintain consistency - return fmt.Errorf("failed to reconcile comment_count for %s: %w", post.URI, reconcileErr) + return false, fmt.Errorf("failed to reconcile comment_count for %s: %w", post.URI, reconcileErr) } // Commit transaction if err := tx.Commit(); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) + return false, fmt.Errorf("failed to commit transaction: %w", err) } - return nil + return true, nil } // errValidationInfra marks a post-validation failure caused by an infrastructure fault @@ -823,25 +877,83 @@ Downvotes int `json:"downvotes"` AsOf string `json:"asOf"` } -// sanitizedPostFacets drops facets whose byte ranges fall outside the post's +// sanitizedPostFacets sanitizes the facets on a community-repo post record. +// +// A record-shaped wrapper over sanitizeFacets, kept because the author-repo +// record type deliberately has no author field and so cannot be the same type: +// the shared work is the range checking, not the unwrapping. +func sanitizedPostFacets(postRecord *PostRecordFromJetstream, uri string) []interface{} { + return sanitizeFacets(postRecord.Facets, postRecord.Content, uri) +} + +// sanitizeFacets drops facets whose byte ranges fall outside the post's // content (or are otherwise structurally invalid) before indexing. Firehose // records from federated repos cannot be rejected back to their author, and // clients must never receive ranges that slice outside the content, so invalid // facets are dropped rather than failing the event. Returns nil when no // facets survive, preserving the callers' nil-means-absent serialization. -func sanitizedPostFacets(postRecord *PostRecordFromJetstream, uri string) []interface{} { - if postRecord.Facets == nil { +func sanitizeFacets(facets []interface{}, content *string, uri string) []interface{} { + if facets == nil { return nil } contentByteLen := 0 - if postRecord.Content != nil { - contentByteLen = len(*postRecord.Content) + if content != nil { + contentByteLen = len(*content) } - kept, dropped := richtext.SanitizeFacets(postRecord.Facets, contentByteLen) + kept, dropped := richtext.SanitizeFacets(facets, contentByteLen) if dropped > 0 { log.Printf("Warning: dropped %d invalid facet(s) on post %s during indexing", dropped, uri) } return kept +} + +// serializePostContent marshals the three optional JSON columns a post record +// carries. A marshal failure is returned rather than swallowed: silently +// dropping facets, an embed, or labels would index a post that reads as though +// its author never sent them. +func serializePostContent(facets []interface{}, embed map[string]interface{}, labels *posts.SelfLabels) (facetsJSON, embedJSON, labelsJSON sql.NullString, err error) { + if facets != nil { + b, marshalErr := json.Marshal(facets) + if marshalErr != nil { + return facetsJSON, embedJSON, labelsJSON, fmt.Errorf("failed to serialize facets: %w", marshalErr) + } + facetsJSON.String, facetsJSON.Valid = string(b), true + } + if embed != nil { + b, marshalErr := json.Marshal(embed) + if marshalErr != nil { + return facetsJSON, embedJSON, labelsJSON, fmt.Errorf("failed to serialize embed: %w", marshalErr) + } + embedJSON.String, embedJSON.Valid = string(b), true + } + if labels != nil { + b, marshalErr := json.Marshal(labels) + if marshalErr != nil { + return facetsJSON, embedJSON, labelsJSON, fmt.Errorf("failed to serialize labels: %w", marshalErr) + } + labelsJSON.String, labelsJSON.Valid = string(b), true + } + return facetsJSON, embedJSON, labelsJSON, nil +} + +// parseRecordCreatedAt reads a record's author-supplied createdAt, falling back +// to now when it does not parse. +// +// SECURITY: future timestamps are clamped to now. created_at drives the "new" +// sort and the hot-rank age, so a record asserting a future date (hostile or +// clock-skewed federated repo) could otherwise pin itself to the top of feeds +// until wall-clock catches up. +func parseRecordCreatedAt(raw, uri string) time.Time { + createdAt, err := time.Parse(time.RFC3339, raw) + if err != nil { + log.Printf("Warning: Failed to parse createdAt timestamp for %s, using current time: %v", uri, err) + return time.Now() + } + if now := time.Now(); createdAt.After(now) { + log.Printf("Warning: post %s has future createdAt %s, clamping to now", uri, raw) + return now + } + return createdAt } // parsePostRecord converts a raw Jetstream record map to a PostRecordFromJetstream diff --git a/internal/atproto/lexicon/social/coves/community/post/getStatus.json b/internal/atproto/lexicon/social/coves/community/post/getStatus.json new file mode 100644 --- /dev/null +++ b/internal/atproto/lexicon/social/coves/community/post/getStatus.json @@ -0,0 +1,58 @@ +{ + "lexicon": 1, + "id": "social.coves.community.post.getStatus", + "defs": { + "main": { + "type": "query", + "description": "Get one community's admission decision about one post. Intentionally UNAUTHENTICATED: the caller with the strongest need is an author on another server whose post is pending on this host, and they have no account here to authenticate with. It is also the only way a rejection is reachable at all - a submission refused before it was ever accepted writes no community record, so there is no repository record and no firehose event carrying it, and without this endpoint an author whose post vanished could never learn that it was refused or why. The accepted cost is that anyone who can name a post AT-URI learns its status in a community. Both parameters are required: a post carries independent decisions from several communities, so there is no single status of a post.", + "parameters": { + "type": "params", + "required": ["post", "community"], + "properties": { + "post": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the post, in the author's repository" + }, + "community": { + "type": "string", + "format": "did", + "description": "DID of the community whose decision is being asked about" + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["status"], + "properties": { + "status": { + "type": "string", + "knownValues": ["pending", "accepted", "pending_reacceptance", "rejected", "removed"], + "description": "The community's decision state. Reported verbatim rather than collapsed: an author who edited an accepted post needs pending_reacceptance to be distinguishable from a post that was never accepted, because the two have completely different next steps." + }, + "decisionCode": { + "type": "string", + "description": "Why the post was refused. Present only for rejected and removed. The vocabulary is open and spans both the codes a community publishes in a removal record and the admission-time codes that never reach a repository." + }, + "decisionAt": { + "type": "string", + "format": "datetime", + "description": "When the refusal above was decided" + }, + "acceptanceUri": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the live community acceptance record, so the caller can read the signed attestation rather than trusting this AppView's summary of it. Present only while an acceptance stands." + } + } + } + }, + "errors": [ + {"name": "InvalidRequest", "description": "A missing or malformed post or community parameter"}, + {"name": "NotFound", "description": "This community has no decision about this post"} + ] + } + } +} diff --git a/internal/core/posts/status.go b/internal/core/posts/status.go --- a/internal/core/posts/status.go +++ b/internal/core/posts/status.go @@ -2,12 +2,9 @@ package posts import ( "context" + "strings" "time" ) - -// RED STUB (task 5, cycle 1). Signatures only — every method returns zero -// values so the tests that describe this surface compile and fail on their -// assertions rather than on a missing symbol. The implementation is GREEN's. // The read side of an admission decision: social.coves.community.post.getStatus // (docs/PRD_AUTHOR_OWNED_POSTS.md §3.4). @@ -78,6 +75,45 @@ func NewStatusService(admissions AdmissionRepository) StatusService { return &statusService{admissions: admissions} } +// GetStatus reads one community's decision about one post. +// +// Both halves of the subject are required rather than defaulted, because a +// post genuinely carries independent decisions from several communities (§2) +// and answering about whichever row was found first would report one +// community's verdict as though it were another's. func (s *statusService) GetStatus(ctx context.Context, req GetStatusRequest) (*PostStatus, error) { - return nil, nil + if strings.TrimSpace(req.PostURI) == "" { + return nil, NewValidationError("post", "post URI is required") + } + if strings.TrimSpace(req.CommunityDID) == "" { + return nil, NewValidationError("community", "community DID is required") + } + + admission, err := s.admissions.Get(ctx, req.CommunityDID, req.PostURI) + if err != nil { + // ErrNotFound travels out unchanged: a subject the community has never + // been offered is a genuine 404, not a status to invent. Reporting it + // as `pending` would promise the author that somebody is going to + // decide. + return nil, err + } + + status := &PostStatus{ + Status: admission.Status, + // The live acceptance record, and only while one stands. The repository + // clears these columns on removal and never sets them on a rejection, + // so this is the acceptance a caller can actually go and read. + AcceptanceURI: admission.AcceptanceURI, + } + + // The decision fields are gated on the status rather than copied blind. + // They describe a REFUSAL, and the two statuses above are the only ones a + // refusal produces; surfacing a code beside `pending` would tell an author + // their post was refused while it is still waiting. + if admission.Status == AdmissionStatusRejected || admission.Status == AdmissionStatusRemoved { + status.DecisionCode = admission.DecisionCode + status.DecisionAt = admission.DecisionAt + } + + return status, nil } diff --git a/internal/db/migrations/036_create_deleted_accounts.sql b/internal/db/migrations/036_create_deleted_accounts.sql new file mode 100644 --- /dev/null +++ b/internal/db/migrations/036_create_deleted_accounts.sql @@ -0,0 +1,60 @@ +-- +goose Up +-- The erasure marker: proof that a DID was deleted ON PURPOSE +-- (docs/PRD_AUTHOR_OWNED_POSTS.md §5.3, rev 2.7). +-- +-- WHY THIS EXISTS. Account deletion used to leave no trace. userRepo.Delete +-- removes the users row, the posts, and (since migration 034) the admission +-- rows — and then the firehose redelivers a post event for that same author, +-- or a dead letter for one is redriven, and every swept row comes straight +-- back. Nothing in the schema could tell the consumer not to re-index it. +-- +-- The absence of a users row cannot carry that meaning, because under +-- author-owned posts it already means something else and something normal: a +-- post record now lives in the AUTHOR's repo, so its author may be someone +-- this AppView has never indexed, and §5.3 REQUIRES that event to index +-- anyway. "No users row" is therefore the ordinary state of a federated +-- author, and reading it as "erased" would refuse the open federated posting +-- the whole design exists to enable. +-- +-- A row here means "this DID was erased on purpose"; no row means "never +-- seen". That is the entire distinction, and it is why the table holds a DID +-- and almost nothing else. +-- +-- WHY NO FOREIGN KEY. The marker outlives the users row by construction — it +-- is written in the same transaction that deletes it — so a reference to +-- users(did) could never be satisfied. It is deliberately not scoped to +-- accounts this AppView hosts either: an erasure request may name a DID whose +-- repo lives elsewhere. +-- +-- HOW IT IS CLEARED. Re-registration. A DID that comes back — the same person +-- signing up again, or an account restored after a mistaken deletion — must +-- index normally, so the repository's user INSERT removes the marker in the +-- same transaction. A marker left standing would make the AppView accept the +-- account's profile and then silently drop every post it writes, forever, +-- with nothing anywhere explaining why. +CREATE TABLE deleted_accounts ( + -- The DID is the whole key: one marker per account, so a re-delete + -- updates in place rather than accumulating rows the ingestion gate would + -- have to deduplicate on every event it reads. + did TEXT PRIMARY KEY, + + -- NOT NULL because the marker's only job is to be READ by a consumer + -- deciding whether to index an event, and a marker with no time cannot + -- participate in any retention or audit answer later. + deleted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + + -- Nullable, and expected to stay NULL for AppView-initiated deletions: + -- nothing knows the account's repo revision at deletion time, because the + -- deletion is a local administrative act rather than a commit. A column + -- that had to be filled would be filled with a fabricated watermark, at + -- the one place a real comparison happens. It exists for the future case + -- where an erasure IS observed as a repo event carrying a rev. + deleted_rev TEXT +); + +COMMENT ON TABLE deleted_accounts IS 'Erasure markers: DIDs deleted on purpose, so ingestion can tell an erased account from a federated author it has never indexed (PRD_AUTHOR_OWNED_POSTS 5.3)'; +COMMENT ON COLUMN deleted_accounts.deleted_at IS 'When the deletion happened; read by retention and audit, never by the ingestion gate itself'; +COMMENT ON COLUMN deleted_accounts.deleted_rev IS 'Repo revision the erasure was observed at, when one is known; NULL for AppView-initiated deletions'; + +-- +goose Down +DROP TABLE IF EXISTS deleted_accounts; diff --git a/internal/db/postgres/deleted_account_repo.go b/internal/db/postgres/deleted_account_repo.go --- a/internal/db/postgres/deleted_account_repo.go +++ b/internal/db/postgres/deleted_account_repo.go @@ -3,9 +3,8 @@ import ( "context" "database/sql" + "fmt" ) - -// RED STUB (task 5, cycle 1). Signatures only; the query is GREEN's. // DeletedAccountRepository reads the migration-036 erasure markers. // @@ -31,5 +30,11 @@ // is indistinguishable from a healthy answer — a database blip would silently // re-index the content a deletion erased, which is the exact outcome the marker // table exists to prevent. func (r *DeletedAccountRepository) IsAccountDeleted(ctx context.Context, did string) (bool, error) { - return false, nil + var deleted bool + if err := r.db.QueryRowContext(ctx, + `SELECT EXISTS (SELECT 1 FROM deleted_accounts WHERE did = $1)`, did, + ).Scan(&deleted); err != nil { + return false, fmt.Errorf("checking whether %s was erased: %w", did, err) + } + return deleted, nil } diff --git a/internal/db/postgres/user_repo.go b/internal/db/postgres/user_repo.go --- a/internal/db/postgres/user_repo.go +++ b/internal/db/postgres/user_repo.go @@ -21,14 +21,44 @@ func NewUserRepository(db *sql.DB) users.UserRepository { return &postgresUserRepo{db: db} } -// Create inserts a new user into the users table +// Create inserts a new user into the users table. +// +// It also clears any migration-036 erasure marker for the DID, in the same +// transaction, because registering IS the marker's exit. A DID that comes back +// — the same person signing up again, or an account restored after a mistaken +// deletion — must index normally, and a marker left standing would have the +// ingestion gate silently drop every post the returning account writes. Both +// service paths funnel through here (IndexUser via CreateUser, and +// RegisterAccount), which is why the clear lives at the repository statement +// rather than in either of them. func (r *postgresUserRepo) Create(ctx context.Context, user *users.User) (*users.User, error) { + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("failed to start transaction creating user did=%s: %w", user.DID, err) + } + defer func() { + if err := tx.Rollback(); err != nil && err != sql.ErrTxDone { + slog.Error("failed to rollback user create transaction", + slog.String("did", user.DID), + slog.String("error", err.Error()), + ) + } + }() + + // Ordered before the insert so that a failing insert — a duplicate DID or a + // taken handle — rolls the clear back with it. Clearing a marker for an + // account that did not actually re-register would silently re-open + // ingestion for content the AppView was asked to forget. + if _, err := tx.ExecContext(ctx, `DELETE FROM deleted_accounts WHERE did = $1`, user.DID); err != nil { + return nil, fmt.Errorf("failed to clear deletion marker for did=%s: %w", user.DID, err) + } + query := ` INSERT INTO users (did, handle, pds_url) VALUES ($1, $2, $3) RETURNING did, handle, pds_url, created_at, updated_at` - err := r.db.QueryRowContext(ctx, query, user.DID, user.Handle, user.PDSURL). + err = tx.QueryRowContext(ctx, query, user.DID, user.Handle, user.PDSURL). Scan(&user.DID, &user.Handle, &user.PDSURL, &user.CreatedAt, &user.UpdatedAt) if err != nil { // Check for unique constraint violations @@ -41,6 +71,10 @@ return nil, users.ErrHandleAlreadyTaken } } return nil, fmt.Errorf("failed to create user: %w", err) + } + + if err := tx.Commit(); err != nil { + return nil, fmt.Errorf("failed to commit user create transaction for did=%s: %w", user.DID, err) } return user, nil @@ -250,6 +284,30 @@ slog.String("error", err.Error()), ) } }() + + // 0. Record the erasure marker (migration 036). + // + // It goes FIRST and inside this transaction, both deliberately. Inside, + // because a marker that survived a rolled-back deletion would name an + // account that still exists — and the ingestion gate reads this table, so + // that account's future posts would be dropped forever with no row + // anywhere explaining it. First, because every statement below erases + // content, and the marker is what stops the firehose putting it back: a + // redriven post event or a replayed acceptance for this DID arrives long + // after the sweep, and without a marker the consumer cannot tell an erased + // account from a federated author it has simply never indexed (§5.3). + // + // deleted_rev is left NULL: an AppView-initiated deletion is a local + // administrative act, not a repo commit, so there is no revision to record + // and inventing one would put a fabricated watermark where real + // comparisons happen. A re-delete refreshes the timestamp rather than + // erroring, so the sweep stays idempotent. + if _, err := tx.ExecContext(ctx, ` + INSERT INTO deleted_accounts (did, deleted_at) VALUES ($1, NOW()) + ON CONFLICT (did) DO UPDATE SET deleted_at = NOW() + `, did); err != nil { + return fmt.Errorf("failed to record deletion marker for did=%s: %w", did, err) + } // Delete in correct order to avoid foreign key violations // Tables without FK constraints on user_did are deleted first