From 73c731a86bb1a6f0e2d490bcab166521d85b5f19 Mon Sep 17 00:00:00 2001 From: Bretton <36870434+BrettM86@users.noreply.github.com> Date: Thu, 16 Jul 2026 18:09:04 -0700 Subject: [PATCH] =?UTF-8?q?feat(jetstream):=20rev-gated=20multi-feed=20ing?= =?UTF-8?q?estion=20=E2=80=94=20escape=20bsky.network=20quotas?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The AppView now consumes N Jetstream feeds carrying the same repos: the public bsky.network Jetstream plus our self-hosted relay+Jetstream pair (tidepool stack) that crawls tdpl.io + pds.coves.me with no per-host account quotas. Feeds are internally ordered but skewed by hours, so a lagging feed replays a repo's history AFTER newer events were applied — resurrecting deleted comments/votes, regressing edits, re-subscribing unsubscribed users. The existing time_us recency guards cannot catch this: each feed stamps its own emission time, so a stale copy arrives with a NEWER timestamp than the state it would clobber. The keystone is rev-gating: every commit event carries rev, the repo's monotonic TID. jetstream_record_revs (migration 033, COLLATE "C") stores the last APPLIED rev per record URI; create/update/delete apply only if strictly newer. Equal rev = duplicate replay (no-op, subsumes existing duplicate handling); the gate row survives hard deletes as the tombstone that rejects stale creates. One rule, no heuristics, no CID comparisons. Changes: - rev_gate.go: conditional-upsert gate primitives + RevGate + applyGated (transactional claim held across apply; same-URI handlers serialize on the gate row lock, apply failure rolls back un-advanced for redrive) - posts/comments/votes: gate as first statement of existing transactions; delete paths restructured gate-first so the tombstone commits atomically with the deletion (closes the not-found-delete vs concurrent-create race) - users/communities/aggregators: injected RevGate; deletes tombstone even for never-indexed records; comments re-apply genuinely-newer same-rkey re-creates on active rows; SubscribeWithCount conflict path updates record_uri/cid last-write-wins (cross-rkey redriven-delete safety) - feeds.go: JETSTREAM_FEEDS="bsky=…;self=…" replaces six per-consumer URL env vars (now fatal at boot with migration hint); per-consumer collection filters derived in code via WantedCollections (unknown name = fatal, no more filterless whole-firehose subscriptions); "bsky" feed keeps legacy consumer names so live cursors carry over; @self consumers live-tail - fail-closed boot checks: multi-feed refuses ungated consumers (RevGated()), missing JETSTREAM_FEEDS fatal outside dev, warning when no primary feed key is configured - fixes latent drift: community.block was never in the subscribed collections; users consumer subscribed to the entire firehose in prod - tests: adversarial cross-feed interleavings (zombie create, stale update clobber, phantom vote, delete-before-create tombstones for votes AND posts, subscription zombie, profile stale-replay, gate semantics) + feeds parsing suite; Makefile test target runs packages sequentially (-p 1) so the integration suite's unscoped table wipes can't race other packages on the shared test DB Known limitations (documented in code): identity events carry no rev so handle changes are not cross-feed ordered; posts/votes active-row same- rkey re-creates after a dead-lettered delete keep the old content. Deploy notes: migration 033 auto-runs at boot. Prod compose sets JETSTREAM_FEEDS with self=ws://tidepool-prod-jetstream:8080 (relay + Jetstream deployed 2026-07-17). Expect "rev-gate: skipped stale" log lines for lagging bsky copies — that is the system working. Co-Authored-By: Claude Fable 5 --- .env.dev | 19 +- .env.dev.example | 8 +- .env.prod.example | 33 +- Makefile | 5 +- cmd/server/main.go | 204 ++++--- docker-compose.prod.yml | 22 +- docs/COMMENT_SYSTEM_IMPLEMENTATION.md | 5 +- docs/PRD_ALPHA_GO_LIVE.md | 6 +- .../atproto/jetstream/aggregator_consumer.go | 42 +- .../atproto/jetstream/comment_consumer.go | 252 +++++--- .../atproto/jetstream/community_consumer.go | 61 +- internal/atproto/jetstream/feeds.go | 174 ++++++ internal/atproto/jetstream/feeds_test.go | 110 ++++ internal/atproto/jetstream/post_consumer.go | 106 +++- internal/atproto/jetstream/rev_gate.go | 201 +++++++ internal/atproto/jetstream/rev_gate_test.go | 561 ++++++++++++++++++ internal/atproto/jetstream/user_consumer.go | 76 ++- internal/atproto/jetstream/vote_consumer.go | 320 ++++++---- .../033_create_jetstream_record_revs.sql | 40 ++ .../postgres/community_repo_subscriptions.go | 43 +- scripts/dev-run.sh | 2 +- tests/integration/aggregator_e2e_test.go | 2 +- tests/integration/comment_vote_test.go | 6 +- tests/integration/post_e2e_test.go | 2 +- 24 files changed, 1926 insertions(+), 374 deletions(-) create mode 100644 internal/atproto/jetstream/feeds.go create mode 100644 internal/atproto/jetstream/feeds_test.go create mode 100644 internal/atproto/jetstream/rev_gate.go create mode 100644 internal/atproto/jetstream/rev_gate_test.go create mode 100644 internal/db/migrations/033_create_jetstream_record_revs.sql diff --git a/.env.dev b/.env.dev index 71ae81d..757e0cf 100644 --- a/.env.dev +++ b/.env.dev @@ -86,17 +86,20 @@ POSTGRES_TEST_PASSWORD=test_password POSTGRES_TEST_PORT=5434 # ============================================================================= -# Jetstream Configuration (Read-Forward User Indexing) +# Jetstream Configuration (multi-feed) # ============================================================================= -# Jetstream WebSocket URL for real-time atProto events +# Semicolon-separated = entries. Every consumer runs once per +# feed; per-consumer collection filters are derived in code +# (jetstream.WantedCollections), so base URLs carry no query string (a path +# is optional; a trailing /subscribe is tolerated). +# The "bsky" feed key keeps the legacy consumer names (cursor continuity). # -# Production: Use Bluesky's public Jetstream (indexes entire network) -# JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.actor.profile +# Production example (public Jetstream + self-hosted relay Jetstream): +# JETSTREAM_FEEDS=bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080 # -# Local E2E Testing: Use local Jetstream (indexes only local PDS) -# 1. Start local Jetstream: docker-compose --profile jetstream up pds jetstream -# 2. Use this URL: -JETSTREAM_URL=ws://localhost:6008/subscribe?wantedCollections=social.coves.actor.profile +# Local dev: the dev-stack Jetstream only (indexes only the local PDS). +# Start it with: docker-compose --profile jetstream up pds jetstream +JETSTREAM_FEEDS=self=ws://localhost:6008 # ============================================================================= # Identity Resolution Configuration diff --git a/.env.dev.example b/.env.dev.example index c0b8017..cbba193 100644 --- a/.env.dev.example +++ b/.env.dev.example @@ -58,10 +58,12 @@ PDS_URL=http://localhost:3001 APPVIEW_PUBLIC_URL=http://127.0.0.1:8081 # ============================================================================= -# Jetstream Configuration +# Jetstream Configuration (multi-feed) # ============================================================================= -# User profile indexing - wantedCollections filters to profile events only -JETSTREAM_URL=ws://localhost:6008/subscribe?wantedCollections=social.coves.actor.profile +# Semicolon-separated = entries. Every consumer runs once per +# feed; per-consumer collection filters are derived in code, so base URLs carry +# no query string (path optional). Local dev uses the dev-stack Jetstream only. +JETSTREAM_FEEDS=self=ws://localhost:6008 # ============================================================================= # Identity Resolution diff --git a/.env.prod.example b/.env.prod.example index a61d47d..9e50d7c 100644 --- a/.env.prod.example +++ b/.env.prod.example @@ -135,25 +135,20 @@ CURSOR_SECRET=CHANGE_ME_CURSOR_SECRET # TRUSTED_BRIDGE_PDS_HOSTS=https://tdpl.io # ============================================================================= -# Jetstream Configuration (Real-time Event Indexing) -# ============================================================================= -# User profile indexing -JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.actor.profile - -# Community event indexing (profiles and subscriptions) -# COMMUNITY_JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.community.profile&wantedCollections=social.coves.community.subscription - -# Post indexing -# POST_JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.community.post - -# Vote indexing -# VOTE_JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.feed.vote - -# Comment indexing -# COMMENT_JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.community.comment - -# Aggregator indexing -# AGGREGATOR_JETSTREAM_URL= +# Jetstream Configuration (multi-feed, real-time event indexing) +# ============================================================================= +# Semicolon-separated = entries. Every consumer (users, +# communities, posts, aggregators, votes, comments) runs once per feed; the +# per-consumer collection filters are derived in code +# (jetstream.WantedCollections), so base URLs carry no query string (a path +# is optional; a trailing /subscribe is tolerated). +# bsky = Bluesky's public Jetstream (third-party PDS records + redundancy; +# this feed key keeps the legacy consumer names so cursors carry over) +# self = self-hosted relay+Jetstream crawling our own PDSes (no bsky.network +# account quotas). Container name + port 8080 — the tidepool stack's +# bare `jetstream` alias is a cross-stack collision trap, and 6008 +# is the local dev port. +JETSTREAM_FEEDS=bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080 # ============================================================================= # Cloudflare (for wildcard SSL certificates) diff --git a/Makefile b/Makefile index 8cc5cd0..4f50cd7 100644 --- a/Makefile +++ b/Makefile @@ -128,7 +128,10 @@ test: ## Run fast unit/integration tests (skips slow E2E tests) @echo "$(GREEN)Running migrations on test database...$(RESET)" @goose -dir internal/db/migrations postgres "postgresql://$(POSTGRES_TEST_USER):$(POSTGRES_TEST_PASSWORD)@localhost:$(POSTGRES_TEST_PORT)/$(POSTGRES_TEST_DB)?sslmode=disable" up || true @echo "$(GREEN)Running fast tests (use 'make e2e-test' for E2E tests)...$(RESET)" - @go test ./cmd/... ./internal/... ./tests/... -short -v + @# -p 1 runs packages sequentially: the integration suite's setup wipes + @# shared test-DB tables (unscoped DELETEs), so package-parallel runs race + @# and randomly kill other packages' fixtures (jetstream DB tests above all). + @go test -p 1 ./cmd/... ./internal/... ./tests/... -short -v @echo "$(GREEN)✓ Tests complete$(RESET)" e2e-test: ## Run automated E2E tests (requires: make dev-up + make run in another terminal) diff --git a/cmd/server/main.go b/cmd/server/main.go index 9d93dc0..8d39998 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -407,6 +407,53 @@ func main() { // redriver can replay them once the failure clears. jetstreamStateStore := jetstream.NewPostgresStateStore(db) + // Rev gate: the per-record ordering guard that makes it safe to run every + // consumer against MULTIPLE Jetstream feeds carrying the same repos (see + // rev_gate.go / migration 033). Posts/votes/comments gate inside their own + // transactions; the repo-method consumers get this gate injected. + revGate := jetstream.NewRevGate(db) + + // Feed topology. Each entry is =; every consumer runs once + // per feed with its collection filters appended (see feeds.go). The "bsky" + // feed keeps the legacy consumer names so live cursors carry over. + for _, legacy := range []string{ + "JETSTREAM_URL", "COMMUNITY_JETSTREAM_URL", "POST_JETSTREAM_URL", + "AGGREGATOR_JETSTREAM_URL", "VOTE_JETSTREAM_URL", "COMMENT_JETSTREAM_URL", + } { + if os.Getenv(legacy) != "" { + log.Fatalf("%s is no longer supported: configure feeds via JETSTREAM_FEEDS "+ + "(e.g. \"bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080\") "+ + "and remove the legacy variable", legacy) + } + } + feedsSpec := os.Getenv("JETSTREAM_FEEDS") + if feedsSpec == "" { + if !isDevEnv { + log.Fatalf("JETSTREAM_FEEDS is required in production (the localhost default is dev-only): " + + "set semicolon-separated = entries, e.g. " + + "\"bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080\"") + } + // Dev default: the local dev-stack Jetstream only. Production always + // sets JETSTREAM_FEEDS explicitly (see docker-compose.prod.yml). + feedsSpec = "self=ws://localhost:6008" + } + jetstreamFeeds, err := jetstream.ParseFeeds(feedsSpec) + if err != nil { + log.Fatalf("Invalid JETSTREAM_FEEDS: %v", err) + } + hasPrimaryFeed := false + for _, feed := range jetstreamFeeds { + if feed.Key == jetstream.PrimaryFeedKey { + hasPrimaryFeed = true + break + } + } + if !hasPrimaryFeed { + // Expected in local dev (self-only feed); in production this usually + // means cursor continuity from the single-feed era is being forfeited. + log.Printf("⚠️ No JETSTREAM_FEEDS entry uses the primary key %q: every consumer name will be suffixed \"@\", so cursors persisted under the bare legacy names will NOT be used", jetstream.PrimaryFeedKey) + } + // All consumers run on one cancellable context so SIGTERM drains them: // read loops unblock, an interrupted in-flight event is abandoned without // advancing the cursor (it replays idempotently on next boot), and the @@ -435,10 +482,16 @@ func main() { }() } - // Start Jetstream consumer for read-forward user indexing - jetstreamURL := os.Getenv("JETSTREAM_URL") - if jetstreamURL == "" { - jetstreamURL = "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.actor.profile&wantedCollections=social.coves.actor.block" + // registerFeedConsumer defers wiring until every consumer exists; the + // feeds×consumers loop below then starts one connector per (feed, + // consumer) pair. Registration order is preserved. + type feedConsumer struct { + name string + handler jetstream.EventHandler + } + var feedConsumers []feedConsumer + registerFeedConsumer := func(name string, handler jetstream.EventHandler) { + feedConsumers = append(feedConsumers, feedConsumer{name: name, handler: handler}) } // Create user consumer with session handle updater to sync OAuth sessions on handle changes @@ -458,6 +511,7 @@ func main() { } bridgeTrust := jetstream.NewBridgeTrust(trustedBridgePDSHosts) consumerOpts = append(consumerOpts, jetstream.WithUserBridgeTrust(bridgeTrust)) + consumerOpts = append(consumerOpts, jetstream.WithUserRevGate(revGate)) if sessionUpdater, ok := baseOAuthStore.(jetstream.SessionHandleUpdater); ok { consumerOpts = append(consumerOpts, jetstream.WithSessionHandleUpdater(sessionUpdater)) log.Println("✅ OAuth session handle sync enabled for identity changes") @@ -468,21 +522,15 @@ func main() { consumerOpts = append(consumerOpts, jetstream.WithUserBlockRepo(userBlockRepo)) userConsumer := jetstream.NewUserEventConsumer(userService, identityResolver, consumerOpts...) - startJetstreamConsumer(jetstream.ConsumerUsers, jetstreamURL, userConsumer) - - log.Printf("Started Jetstream user consumer: %s", jetstreamURL) + registerFeedConsumer(jetstream.ConsumerUsers, userConsumer) + log.Println("Registered Jetstream user consumer (actor profiles + blocks)") - // Start Jetstream consumer for community events (profiles and subscriptions) - // This consumer indexes: + // Register Jetstream consumer for community events. This consumer indexes: // 1. Community profiles (social.coves.community.profile) - in community's own repo // 2. User subscriptions (social.coves.community.subscription) - in user's repo - communityJetstreamURL := os.Getenv("COMMUNITY_JETSTREAM_URL") - if communityJetstreamURL == "" { - // Local Jetstream for communities - filter to our instance's collections - // IMPORTANT: We listen to social.coves.community.subscription (not social.coves.community.subscribe) - // because subscriptions are RECORD TYPES in the communities namespace, not XRPC procedures - communityJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.profile&wantedCollections=social.coves.community.subscription" - } + // 3. Community blocks (social.coves.community.block) - in user's repo + // (Record-type collections, not XRPC procedures; filters come from + // jetstream.WantedCollections.) // Initialize community event consumer with did:web verification skipDIDWebVerification := os.Getenv("SKIP_DID_WEB_VERIFICATION") == "true" @@ -492,12 +540,10 @@ func main() { } // Pass identity resolver to consumer for PLC handle resolution (source of truth) - communityEventConsumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, skipDIDWebVerification, identityResolver) - startJetstreamConsumer(jetstream.ConsumerCommunities, communityJetstreamURL, communityEventConsumer) - - log.Printf("Started Jetstream community consumer: %s", communityJetstreamURL) - log.Println(" - Indexing: social.coves.community.profile (community profiles)") - log.Println(" - Indexing: social.coves.community.subscription (user subscriptions)") + communityEventConsumer := jetstream.NewCommunityEventConsumer(communityRepo, instanceDID, skipDIDWebVerification, identityResolver, + jetstream.WithCommunityRevGate(revGate)) + registerFeedConsumer(jetstream.ConsumerCommunities, communityEventConsumer) + log.Println("Registered Jetstream community consumer (profiles, subscriptions, blocks)") // Start OAuth session cleanup background job with cancellable context cleanupCtx, cleanupCancel := context.WithCancel(context.Background()) @@ -779,14 +825,8 @@ func main() { }) log.Printf("Image proxy URL generation config set (enabled: %v)", imageProxyConfig.Enabled) - // Start Jetstream consumer for posts + // Register Jetstream consumer for posts // This consumer indexes posts created in community repositories via the firehose - // Currently handles only CREATE operations - UPDATE/DELETE deferred until those features exist - postJetstreamURL := os.Getenv("POST_JETSTREAM_URL") - if postJetstreamURL == "" { - // Listen to post record creation events - postJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.post" - } // Provenance gate for bridge-asserted vote aggregates (bridgedStats). Only records // whose repo is hosted on a trusted bridge PDS may inflate their displayed vote @@ -803,61 +843,67 @@ func main() { postEventConsumer := jetstream.NewPostEventConsumer(postRepo, communityRepo, userService, db, jetstream.WithPostBridgeTrust(bridgeTrust), jetstream.WithPostIdentityResolver(identityResolver)) - startJetstreamConsumer(jetstream.ConsumerPosts, postJetstreamURL, postEventConsumer) - - log.Printf("Started Jetstream post consumer: %s", postJetstreamURL) - log.Println(" - Indexing: social.coves.community.post CREATE/UPDATE/DELETE operations") - - // Start Jetstream consumer for aggregators - // This consumer indexes aggregator service declarations and authorization records - // Following Bluesky's pattern for feed generators and labelers - // NOTE: Uses the same Jetstream as communities, just filtering different collections - aggregatorJetstreamURL := communityJetstreamURL - // Override if specific URL needed for testing - if envURL := os.Getenv("AGGREGATOR_JETSTREAM_URL"); envURL != "" { - aggregatorJetstreamURL = envURL - } else if aggregatorJetstreamURL == "" { - // Fallback if community URL also not set - aggregatorJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.aggregator.service&wantedCollections=social.coves.aggregator.authorization" - } - - aggregatorEventConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo) - startJetstreamConsumer(jetstream.ConsumerAggregators, aggregatorJetstreamURL, aggregatorEventConsumer) - - log.Printf("Started Jetstream aggregator consumer: %s", aggregatorJetstreamURL) - log.Println(" - Indexing: social.coves.aggregator.service (service declarations)") - log.Println(" - Indexing: social.coves.aggregator.authorization (authorization records)") - - // Start Jetstream consumer for votes - // This consumer indexes votes from user repositories and updates post vote counts - voteJetstreamURL := os.Getenv("VOTE_JETSTREAM_URL") - if voteJetstreamURL == "" { - // Listen to vote record CREATE/DELETE events from user repositories - voteJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.feed.vote" - } - + registerFeedConsumer(jetstream.ConsumerPosts, postEventConsumer) + log.Println("Registered Jetstream post consumer (CREATE/UPDATE/DELETE)") + + // Register Jetstream consumer for aggregators: service declarations and + // authorization records, following Bluesky's feed generator/labeler pattern + aggregatorEventConsumer := jetstream.NewAggregatorEventConsumer(aggregatorRepo, + jetstream.WithAggregatorRevGate(revGate)) + registerFeedConsumer(jetstream.ConsumerAggregators, aggregatorEventConsumer) + log.Println("Registered Jetstream aggregator consumer (services + authorizations)") + + // Register Jetstream consumer for votes: indexes votes from user + // repositories and updates post/comment vote counts atomically voteEventConsumer := jetstream.NewVoteEventConsumer(voteRepo, userService, db) - startJetstreamConsumer(jetstream.ConsumerVotes, voteJetstreamURL, voteEventConsumer) - - log.Printf("Started Jetstream vote consumer: %s", voteJetstreamURL) - log.Println(" - Indexing: social.coves.feed.vote CREATE/DELETE operations") - log.Println(" - Updating: Post vote counts atomically") - - // Start Jetstream consumer for comments - // This consumer indexes comments from user repositories and updates parent counts - commentJetstreamURL := os.Getenv("COMMENT_JETSTREAM_URL") - if commentJetstreamURL == "" { - // Listen to comment record CREATE/UPDATE/DELETE events from user repositories - commentJetstreamURL = "ws://localhost:6008/subscribe?wantedCollections=social.coves.community.comment" - } + registerFeedConsumer(jetstream.ConsumerVotes, voteEventConsumer) + log.Println("Registered Jetstream vote consumer (CREATE/DELETE + count updates)") + // Register Jetstream consumer for comments: indexes comments from user + // repositories and updates parent post/comment counts atomically commentEventConsumer := jetstream.NewCommentEventConsumer(commentRepo, db, jetstream.WithCommentBridgeTrust(bridgeTrust)) - startJetstreamConsumer(jetstream.ConsumerComments, commentJetstreamURL, commentEventConsumer) + registerFeedConsumer(jetstream.ConsumerComments, commentEventConsumer) + log.Println("Registered Jetstream comment consumer (CREATE/UPDATE/DELETE + count updates)") + + // FAIL CLOSED: with more than one feed, every consumer MUST be rev-gated — + // an ungated consumer would apply the lagging feed's stale copies (zombie + // deletes, regressed edits), which is silent data corruption, not a + // degraded mode. A forgotten WithXRevGate option must stop the boot, not + // ship the bug. + if len(jetstreamFeeds) > 1 { + for _, fc := range feedConsumers { + gated, ok := fc.handler.(interface{ RevGated() bool }) + if !ok || !gated.RevGated() { + log.Fatalf("consumer %q is not rev-gated but %d Jetstream feeds are configured; "+ + "multi-feed operation requires every consumer to carry the rev gate (see rev_gate.go)", + fc.name, len(jetstreamFeeds)) + } + } + } - log.Printf("Started Jetstream comment consumer: %s", commentJetstreamURL) - log.Println(" - Indexing: social.coves.community.comment CREATE/UPDATE/DELETE operations") - log.Println(" - Updating: Post comment counts and comment reply counts atomically") + // Start every registered consumer on every configured feed. Consumer names + // on the primary ("bsky") feed stay bare so live cursors carry over from + // the single-feed era; other feeds get "@" names, which + // start cursor-less and live-tail (recovering older records requires the + // source PDSes to re-emit them — see Tidepool's POST /admin/reemit). + // Rev-gating makes the cross-feed overlap safe; expect "rev-gate: skipped + // stale" log lines for the lagging feed's copies — that is the system + // working, not an error. + for _, feed := range jetstreamFeeds { + for _, fc := range feedConsumers { + collections, collectionsErr := jetstream.WantedCollections(fc.name) + if collectionsErr != nil { + log.Fatalf("Failed to resolve wantedCollections for consumer %s: %v", fc.name, collectionsErr) + } + wsURL, urlErr := jetstream.SubscribeURL(feed.BaseURL, collections) + if urlErr != nil { + log.Fatalf("Failed to build Jetstream URL for consumer %s on feed %s: %v", fc.name, feed.Key, urlErr) + } + startJetstreamConsumer(jetstream.FeedConsumerName(fc.name, feed.Key), wsURL, fc.handler) + } + log.Printf("Started %d Jetstream consumers on feed %q (%s)", len(feedConsumers), feed.Key, feed.BaseURL) + } // Start the dead letter redriver: replays events that failed all in-line // retries against the same consumers, so transient failures (e.g. a diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 241c5f1..485aa15 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -67,15 +67,19 @@ services: # PDS connection (separate domain!) PDS_URL: https://coves.me - # Jetstream (Bluesky production firehose) - JETSTREAM_URL: wss://jetstream2.us-east.bsky.network/subscribe - - # Custom lexicon consumers (use production Jetstream with collection filters) - COMMUNITY_JETSTREAM_URL: wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.community.profile&wantedCollections=social.coves.community.subscription - POST_JETSTREAM_URL: wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.community.post - AGGREGATOR_JETSTREAM_URL: wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.aggregator.service&wantedCollections=social.coves.aggregator.authorization - VOTE_JETSTREAM_URL: wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.feed.vote - COMMENT_JETSTREAM_URL: wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.community.comment + # Jetstream feeds: every consumer runs once per feed; per-consumer + # collection filters are derived in code (jetstream.WantedCollections). + # bsky = Bluesky's public Jetstream (third-party PDS records + redundancy; + # keeps the legacy consumer names so live cursors carry over) + # self = self-hosted relay+Jetstream pair (tidepool stack, deployed + # 2026-07-17) on this same network, crawling tdpl.io + + # pds.coves.me with no bsky.network account quotas. Addressed + # by CONTAINER name on port 8080 — the tidepool stack's bare + # `jetstream` service alias is the same cross-stack trap as + # the `postgres` alias, and 6008 is the local dev port. + # Rev-gating (migration 033) makes the cross-feed overlap safe; expect + # "rev-gate: skipped stale" log lines for the lagging bsky copies. + JETSTREAM_FEEDS: "bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080" # Security - MUST be false in production AUTH_SKIP_VERIFY: "false" diff --git a/docs/COMMENT_SYSTEM_IMPLEMENTATION.md b/docs/COMMENT_SYSTEM_IMPLEMENTATION.md index c318ed6..1fe9344 100644 --- a/docs/COMMENT_SYSTEM_IMPLEMENTATION.md +++ b/docs/COMMENT_SYSTEM_IMPLEMENTATION.md @@ -1454,8 +1454,9 @@ go build ./cmd/server ### Environment Variables ```bash -# Jetstream URL (optional, defaults to localhost:6008) -export COMMENT_JETSTREAM_URL="ws://localhost:6008/subscribe?wantedCollections=social.coves.community.comment" +# Jetstream feeds (optional in dev; defaults to the local dev Jetstream). +# NOTE: the legacy COMMENT_JETSTREAM_URL variable is now rejected at boot. +export JETSTREAM_FEEDS="self=ws://localhost:6008" # Database URL export TEST_DATABASE_URL="postgres://test_user:test_password@localhost:5434/coves_test?sslmode=disable" diff --git a/docs/PRD_ALPHA_GO_LIVE.md b/docs/PRD_ALPHA_GO_LIVE.md index 48a7635..13864ff 100644 --- a/docs/PRD_ALPHA_GO_LIVE.md +++ b/docs/PRD_ALPHA_GO_LIVE.md @@ -57,7 +57,8 @@ All 6 critical E2E test suites have been implemented and are passing: - `INSTANCE_DOMAIN=coves.social` - `PDS_URL=https://coves.me` (separate domain) - `SKIP_DID_WEB_VERIFICATION=false` (production) - - `JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe` + - `JETSTREAM_FEEDS=bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080` + (the six legacy `*_JETSTREAM_URL` vars are now rejected at boot) **Verification**: - `curl https://coves.social/.well-known/did.json` (should return DID document) @@ -262,7 +263,8 @@ This document tracks the remaining work required to launch Coves alpha with real - [ ] `PDS_URL=https://coves.me` (separate domain) - [ ] `AUTH_SKIP_VERIFY=false` - [ ] `SKIP_DID_WEB_VERIFICATION=false` - - [ ] `JETSTREAM_URL=wss://jetstream2.us-east.bsky.network/subscribe` + - [ ] `JETSTREAM_FEEDS=bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080` + (legacy `JETSTREAM_URL` / `*_JETSTREAM_URL` vars are rejected at boot — remove them) - [ ] **PDS Environment Variables** - [ ] `PDS_HOSTNAME=coves.me` - [ ] `PDS_PORT=2583` diff --git a/internal/atproto/jetstream/aggregator_consumer.go b/internal/atproto/jetstream/aggregator_consumer.go index a73be4e..4941fc1 100644 --- a/internal/atproto/jetstream/aggregator_consumer.go +++ b/internal/atproto/jetstream/aggregator_consumer.go @@ -12,16 +12,40 @@ import ( // AggregatorEventConsumer consumes aggregator-related events from Jetstream // Following Bluesky's pattern: feed generators (app.bsky.feed.generator) and labelers (app.bsky.labeler.service) type AggregatorEventConsumer struct { - repo aggregators.Repository // Repository for aggregator operations + repo aggregators.Repository // Repository for aggregator operations + revGate *RevGate // Optional: cross-feed ordering guard for commit events (nil = ungated) +} + +// AggregatorConsumerOption configures optional AggregatorEventConsumer behaviour. +type AggregatorConsumerOption func(*AggregatorEventConsumer) + +// WithAggregatorRevGate installs the per-record rev gate (see rev_gate.go) so +// service and authorization commits are applied in repo commit order even when +// the same repo is carried by multiple Jetstream feeds. Both record types are +// hard-deleted, so the gate row is the tombstone that rejects a stale +// cross-feed copy of the create arriving after the delete. +func WithAggregatorRevGate(gate *RevGate) AggregatorConsumerOption { + return func(c *AggregatorEventConsumer) { + c.revGate = gate + } } // NewAggregatorEventConsumer creates a new Jetstream consumer for aggregator events -func NewAggregatorEventConsumer(repo aggregators.Repository) *AggregatorEventConsumer { - return &AggregatorEventConsumer{ +func NewAggregatorEventConsumer(repo aggregators.Repository, opts ...AggregatorConsumerOption) *AggregatorEventConsumer { + c := &AggregatorEventConsumer{ repo: repo, } + for _, opt := range opts { + opt(c) + } + return c } +// RevGated reports whether this consumer applies the per-record rev gate (true when a +// gate was injected via WithAggregatorRevGate). main.go checks this at boot to refuse +// multi-feed operation with an ungated consumer. +func (c *AggregatorEventConsumer) RevGated() bool { return c.revGate != nil } + // HandleEvent processes a Jetstream event for aggregator records // This is called by the main Jetstream consumer when it receives commit events func (c *AggregatorEventConsumer) HandleEvent(ctx context.Context, event *JetstreamEvent) error { @@ -36,11 +60,19 @@ func (c *AggregatorEventConsumer) HandleEvent(ctx context.Context, event *Jetstr // IMPORTANT: Collection names refer to RECORD TYPES in repositories // - social.coves.aggregator.service: Service declaration (in aggregator's own repo, rkey="self") // - social.coves.aggregator.authorization: Authorization (in community's repo, any rkey) + // Both collections run under the rev gate (check→write→advance, see + // rev_gate.go): events apply in repo commit order even when the same repo + // is carried by multiple Jetstream feeds, and the gate row survives the + // hard deletes below as the tombstone rejecting stale create copies. switch commit.Collection { case "social.coves.aggregator.service": - return c.handleServiceDeclaration(ctx, event.Did, commit) + return applyGated(ctx, c.revGate, ConsumerAggregators, event.Did, commit, func() error { + return c.handleServiceDeclaration(ctx, event.Did, commit) + }) case "social.coves.aggregator.authorization": - return c.handleAuthorization(ctx, event.Did, commit) + return applyGated(ctx, c.revGate, ConsumerAggregators, event.Did, commit, func() error { + return c.handleAuthorization(ctx, event.Did, commit) + }) default: // Not an aggregator-related collection return nil diff --git a/internal/atproto/jetstream/comment_consumer.go b/internal/atproto/jetstream/comment_consumer.go index 8de3250..2172332 100644 --- a/internal/atproto/jetstream/comment_consumer.go +++ b/internal/atproto/jetstream/comment_consumer.go @@ -62,6 +62,11 @@ func NewCommentEventConsumer( return c } +// RevGated reports whether this consumer applies the per-record rev gate; always true +// for comments (gating is hardwired via c.db). main.go checks this at boot to refuse +// multi-feed operation with an ungated consumer. +func (c *CommentEventConsumer) RevGated() bool { return true } + // bridgeStatsAllowedForRepo reports whether the given comment repo (a user DID) is a // trusted bridge, i.e. whether its records may assert bridgedStats. It resolves the // repo's PDS host from the already-indexed users row (users.pds_url, populated from @@ -186,8 +191,8 @@ func (c *CommentEventConsumer) createComment(ctx context.Context, repoDID string } } - // Atomically: Index comment + Update parent counts - if err := c.indexCommentAndUpdateCounts(ctx, comment); err != nil { + // Atomically: Rev-gate + Index comment + Update parent counts + if err := c.indexCommentAndUpdateCounts(ctx, comment, commit.Rev); err != nil { return fmt.Errorf("failed to index comment and update counts: %w", err) } @@ -357,7 +362,32 @@ func (c *CommentEventConsumer) updateComment(ctx context.Context, repoDID string WHERE uri = $1 AND deleted_at IS NULL AND ($11::bigint <= 0 OR indexed_at < to_timestamp($11::bigint / 1000000.0)) ` - result, err := c.db.ExecContext(ctx, updateQuery, + // REV GATE + UPDATE in one transaction. The gate (strictly-newer rev wins) + // is the cross-feed ordering guard: the time_us recency guard in the WHERE + // clause below cannot reject a stale copy delivered by ANOTHER feed, because + // each feed stamps its own emission time — a pre-edit update replayed by the + // 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) + } + defer func() { + if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { + log.Printf("Failed to rollback transaction: %v", rollbackErr) + } + }() + + won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + if err != nil { + return err + } + if !won { + logSkippedStaleRev(ConsumerComments, "update", uri, commit.Rev) + return nil + } + + result, err := tx.ExecContext(ctx, updateQuery, uri, commit.CID, commentRecord.Content, facetsJSON, embedJSON, labelsJSON, pq.Array(commentRecord.Langs), incomingUp, incomingDn, incomingAsOf, @@ -371,6 +401,8 @@ func (c *CommentEventConsumer) updateComment(ctx context.Context, repoDID string // (recency guard) — between the load above and this UPDATE; the WHERE guards then // match no rows. Both cases are success: the row's current state supersedes this // event, so skip instead of erroring (an error would re-dead-letter the event). + // The deferred rollback also reverts the gate advance, which is the conservative + // choice: a replay re-evaluates against whatever state superseded this event. rowsAffected, err := result.RowsAffected() if err != nil { return fmt.Errorf("failed to check comment update result: %w", err) @@ -380,6 +412,10 @@ func (c *CommentEventConsumer) updateComment(ctx context.Context, repoDID string return nil } + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit comment update transaction: %w", err) + } + if incomingAsOf != nil { log.Printf("✓ Updated comment: %s (bridgedStats candidate applied if newer-or-equal: up=%d down=%d)", uri, incomingUp, incomingDn) } else { @@ -388,25 +424,79 @@ func (c *CommentEventConsumer) updateComment(ctx context.Context, repoDID string return nil } -// deleteComment soft-deletes a comment and updates parent counts +// deleteComment soft-deletes a comment, blanking content to preserve thread +// structure while respecting user privacy: the row remains and is shown as +// "[deleted]" in thread views, so parent counts are intentionally NOT +// decremented. +// +// The rev-gate claim runs FIRST, inside the same transaction as the soft +// delete (mirrors deletePost). Claiming before touching the comments table +// closes the not-found tombstone race: a concurrent create of the same +// comment (another feed's copy, or a DeadLetterRedriver replay) serializes on +// the gate row lock, so it either commits before our delete (which then finds +// and soft-deletes the row) or blocks until our tombstone commits (its +// equal-or-older rev then loses the gate). The gate row is advanced — and +// committed — even when the comment was never indexed, so the create's late +// copy is rejected too. func (c *CommentEventConsumer) deleteComment(ctx context.Context, repoDID string, commit *CommitEvent) error { // Build AT-URI for the comment being deleted uri := fmt.Sprintf("at://%s/social.coves.community.comment/%s", repoDID, commit.RKey) - // Get existing comment to know its parent (for decrementing the right counter) - existingComment, err := c.commentRepo.GetByURI(ctx, uri) + tx, err := c.db.BeginTx(ctx, nil) if err != nil { - if err == comments.ErrCommentNotFound { - // Idempotent: Comment already deleted or never existed - log.Printf("Comment already deleted or not found: %s", uri) - return nil + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer func() { + if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { + log.Printf("Failed to rollback transaction: %v", rollbackErr) + } + }() + + // 0. REV GATE (see indexCommentAndUpdateCounts): skip duplicate replays and + // stale cross-feed copies; the claimed row doubles as the tombstone that + // rejects the create's later copies. + won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + if err != nil { + return err + } + if !won { + logSkippedStaleRev(ConsumerComments, "delete", uri, commit.Rev) + return nil + } + + // 1. Soft-delete the comment: blank content but preserve structure. + // DELETE event from Jetstream = author deleted their own comment (the repo + // owner IS the commenter), so deleted_by is the repo DID. + // Use the repository's transaction-aware method for DRY. + repoTx, ok := c.commentRepo.(comments.RepositoryTx) + if !ok { + return fmt.Errorf("comment repository does not support transactional operations") + } + + rowsAffected, err := repoTx.SoftDeleteWithReasonTx(ctx, tx, uri, comments.DeletionReasonAuthor, repoDID) + if err != nil { + return fmt.Errorf("failed to delete comment: %w", err) + } + + // Idempotent: zero rows means the comment was already deleted or never + // indexed. Commit anyway — the gate advance is the tombstone that rejects + // a stale cross-feed copy of the CREATE arriving later for a record that + // no longer exists on the PDS. + if rowsAffected == 0 { + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) } - return fmt.Errorf("failed to get existing comment: %w", err) + log.Printf("Comment already deleted or not found: %s", uri) + return nil } - // Atomically: Soft-delete comment + Update parent counts - if err := c.deleteCommentAndUpdateCounts(ctx, existingComment); err != nil { - return fmt.Errorf("failed to delete comment and update counts: %w", err) + // NOTE: We intentionally do NOT decrement parent counts (comment_count/reply_count) + // Deleted comments are shown as "[deleted]" placeholders to preserve thread structure, + // so they should still count toward the displayed total. + + // Commit transaction + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) } log.Printf("✓ Deleted comment: %s", uri) @@ -414,7 +504,7 @@ func (c *CommentEventConsumer) deleteComment(ctx context.Context, repoDID string } // indexCommentAndUpdateCounts atomically indexes a comment and updates parent counts -func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, comment *comments.Comment) error { +func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, comment *comments.Comment, rev string) error { tx, err := c.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) @@ -425,14 +515,31 @@ func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, } }() + // 0. REV GATE: apply this create only if its rev is strictly newer than the + // last applied event for this record. This is what makes the resurrection + // branch below SAFE: a stale cross-feed copy of the original create arriving + // after the comment's delete carries an older rev than the tombstoned delete + // and is rejected here, while a genuine re-creation of the rkey carries a + // fresh, higher rev and passes through to the resurrection path. Runs first, + // inside the transaction, so gate and writes commit or roll back together. + won, err := tryAdvanceRecordRev(ctx, tx, comment.URI, rev) + if err != nil { + return err + } + if !won { + logSkippedStaleRev(ConsumerComments, "create", comment.URI, rev) + return nil + } + // 1. Check if comment exists and handle resurrection case // In atProto, deleted records' rkeys become available - users can recreate with same rkey // We must distinguish: idempotent replay (skip) vs resurrection (update + restore counts) var existingID int64 + var existingCID string var existingDeletedAt *time.Time var existingParentURI, existingRootURI string - checkQuery := `SELECT id, deleted_at, parent_uri, root_uri FROM comments WHERE uri = $1` - checkErr := tx.QueryRowContext(ctx, checkQuery, comment.URI).Scan(&existingID, &existingDeletedAt, &existingParentURI, &existingRootURI) + checkQuery := `SELECT id, cid, deleted_at, parent_uri, root_uri FROM comments WHERE uri = $1` + checkErr := tx.QueryRowContext(ctx, checkQuery, comment.URI).Scan(&existingID, &existingCID, &existingDeletedAt, &existingParentURI, &existingRootURI) var commentID int64 var isResurrectionWithSameParent bool // Track if we should skip parent count increment @@ -440,7 +547,67 @@ func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, if checkErr == nil { // Comment exists if existingDeletedAt == nil { - // Not deleted - this is an idempotent replay, skip gracefully + // Active row. Usually this is an idempotent replay of the same create + // (same CID; equal revs are already rejected by the gate, and empty-rev + // synthetic events land here too). But the gate can also admit a create + // with a STRICTLY NEWER rev for an rkey whose row is still active: + // create A applied → delete dead-lettered (failed, never applied) → + // genuine re-create B of the same rkey. B carries new content and a new + // CID; treating it as a duplicate would drop that content forever (the + // gate has advanced to B's rev, so no replay can fix it). When the gate + // won with a real rev, the incoming CID differs, and the threading refs + // are UNCHANGED, apply B's record content in place — without touching + // deletion metadata, reply counts, or native votes. + if rev != "" && comment.CID != existingCID && + existingParentURI == comment.ParentURI && existingRootURI == comment.RootURI { + log.Printf("Re-create of active comment with newer rev: %s (applying new content, CID %s -> %s)", + comment.URI, existingCID, comment.CID) + recreateQuery := ` + UPDATE comments + SET + cid = $1, + root_cid = $2, + parent_cid = $3, + content = $4, + content_facets = $5, + embed = $6, + content_labels = $7, + langs = $8, + created_at = $9, + indexed_at = $10, + bridged_upvote_count = $11, + bridged_downvote_count = $12, + bridged_stats_as_of = $13, + -- Recompute the inclusive score from the SURVIVING native + -- counts plus the incoming bridged values (see the resurrect + -- path below for the migration 031 invariant). + score = upvote_count + $11 - downvote_count - $12 + WHERE id = $14 + ` + if _, err = tx.ExecContext(ctx, recreateQuery, + comment.CID, comment.RootCID, comment.ParentCID, + comment.Content, comment.ContentFacets, comment.Embed, comment.ContentLabels, + pq.Array(comment.Langs), comment.CreatedAt, comment.IndexedAt, + comment.BridgedUpvoteCount, comment.BridgedDownvoteCount, comment.BridgedStatsAsOf, + existingID, + ); err != nil { + return fmt.Errorf("failed to apply re-created comment content: %w", err) + } + // Parent unchanged and the row was never decounted, so parent counts + // are already correct — commit without the increment sections below. + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil + } + // KNOWN LIMITATION (accepted): the same dead-lettered-delete + + // same-rkey-re-create sequence with a CHANGED parent/root lands here + // and is skipped as a duplicate, keeping the OLD row. Applying it + // would require decrementing the old parent's reply/comment counts + // and incrementing the new ones for a row that was never decounted — + // re-plumbing the count machinery for a case that additionally needs + // a dead-lettered delete AND a cross-thread rkey reuse inside the + // redrive window. Documented rather than fixed. log.Printf("Comment already indexed: %s (idempotent replay)", comment.URI) if commitErr := tx.Commit(); commitErr != nil { return fmt.Errorf("failed to commit transaction: %w", commitErr) @@ -697,55 +864,6 @@ func (c *CommentEventConsumer) indexCommentAndUpdateCounts(ctx context.Context, return nil } -// deleteCommentAndUpdateCounts atomically soft-deletes a comment and updates parent counts -// Blanks content to preserve thread structure while respecting user privacy -// The comment remains in the database but is shown as "[deleted]" in thread views -func (c *CommentEventConsumer) deleteCommentAndUpdateCounts(ctx context.Context, comment *comments.Comment) error { - tx, err := c.db.BeginTx(ctx, nil) - if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) - } - defer func() { - if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { - log.Printf("Failed to rollback transaction: %v", rollbackErr) - } - }() - - // 1. Soft-delete the comment: blank content but preserve structure - // DELETE event from Jetstream = author deleted their own comment - // Content is blanked to respect user privacy while preserving thread structure - // Use the repository's transaction-aware method for DRY - repoTx, ok := c.commentRepo.(comments.RepositoryTx) - if !ok { - return fmt.Errorf("comment repository does not support transactional operations") - } - - rowsAffected, err := repoTx.SoftDeleteWithReasonTx(ctx, tx, comment.URI, comments.DeletionReasonAuthor, comment.CommenterDID) - if err != nil { - return fmt.Errorf("failed to delete comment: %w", err) - } - - // Idempotent: If no rows affected, comment already deleted - return early - if rowsAffected == 0 { - log.Printf("Comment already deleted: %s (idempotent)", comment.URI) - if err := tx.Commit(); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - return nil - } - - // NOTE: We intentionally do NOT decrement parent counts (comment_count/reply_count) - // Deleted comments are shown as "[deleted]" placeholders to preserve thread structure, - // so they should still count toward the displayed total. - - // Commit transaction - if err := tx.Commit(); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - - return nil -} - // validateCommentEvent performs security validation on comment events func (c *CommentEventConsumer) validateCommentEvent(ctx context.Context, repoDID string, comment *CommentRecordFromJetstream) error { // SECURITY: Comments MUST come from user repositories (repo owner = commenter DID) diff --git a/internal/atproto/jetstream/community_consumer.go b/internal/atproto/jetstream/community_consumer.go index 184f7c8..5d10edf 100644 --- a/internal/atproto/jetstream/community_consumer.go +++ b/internal/atproto/jetstream/community_consumer.go @@ -28,6 +28,22 @@ type CommunityEventConsumer struct { wellKnownLimiter *rate.Limiter // Rate limiter for .well-known fetches instanceDID string // DID of this Coves instance skipVerification bool // Skip did:web verification (for dev mode) + revGate *RevGate // Optional: cross-feed ordering guard for commit events (nil = ungated) +} + +// CommunityConsumerOption configures optional CommunityEventConsumer behaviour. +type CommunityConsumerOption func(*CommunityEventConsumer) + +// WithCommunityRevGate installs the per-record rev gate (see rev_gate.go) so +// profile, subscription, and block commits are applied in repo commit order even +// when the same repo is carried by multiple Jetstream feeds. Subscriptions and +// blocks are HARD-deleted, so the gate row is the only tombstone that can reject +// a stale cross-feed copy of the create arriving after the delete (which would +// otherwise silently re-subscribe/re-block the user). +func WithCommunityRevGate(gate *RevGate) CommunityConsumerOption { + return func(c *CommunityEventConsumer) { + c.revGate = gate + } } // cachedDIDDoc represents a cached verification result with expiration @@ -42,7 +58,7 @@ type cachedDIDDoc struct { // identityResolver: Optional resolver for resolving handles from DIDs (can be nil for tests) func NewCommunityEventConsumer(repo communities.Repository, instanceDID string, skipVerification bool, identityResolver interface { Resolve(context.Context, string) (*identity.Identity, error) -}, +}, opts ...CommunityConsumerOption, ) *CommunityEventConsumer { // Create bounded LRU cache for DID document verification results // Max 1000 entries to prevent unbounded memory growth (PR review feedback) @@ -58,7 +74,7 @@ func NewCommunityEventConsumer(repo communities.Repository, instanceDID string, log.Printf("CRITICAL: Failed to create fallback DID cache (size=1): %v", fallbackErr) panic(fmt.Sprintf("cannot create LRU cache: primary error=%v, fallback error=%v", err, fallbackErr)) } - return &CommunityEventConsumer{ + fallback := &CommunityEventConsumer{ repo: repo, identityResolver: identityResolver, instanceDID: instanceDID, @@ -74,9 +90,13 @@ func NewCommunityEventConsumer(repo communities.Repository, instanceDID string, didCache: cache, wellKnownLimiter: rate.NewLimiter(10, 20), } + for _, opt := range opts { + opt(fallback) + } + return fallback } - return &CommunityEventConsumer{ + consumer := &CommunityEventConsumer{ repo: repo, identityResolver: identityResolver, // Optional - can be nil for tests instanceDID: instanceDID, @@ -97,8 +117,17 @@ func NewCommunityEventConsumer(repo communities.Repository, instanceDID string, // Prevents DoS via excessive .well-known fetches wellKnownLimiter: rate.NewLimiter(10, 20), } + for _, opt := range opts { + opt(consumer) + } + return consumer } +// RevGated reports whether this consumer applies the per-record rev gate (true when a +// gate was injected via WithCommunityRevGate). main.go checks this at boot to refuse +// multi-feed operation with an ungated consumer. +func (c *CommunityEventConsumer) RevGated() bool { return c.revGate != nil } + // HandleEvent processes a Jetstream event for community records // This is called by the main Jetstream consumer when it receives commit events func (c *CommunityEventConsumer) HandleEvent(ctx context.Context, event *JetstreamEvent) error { @@ -117,15 +146,27 @@ func (c *CommunityEventConsumer) HandleEvent(ctx context.Context, event *Jetstre // // XRPC procedures (social.coves.community.subscribe/unsubscribe) are just HTTP endpoints // that CREATE or DELETE records in these collections + // All three collections run under the rev gate (check→write→advance, see + // rev_gate.go): events apply in repo commit order even when the same repo + // is carried by multiple Jetstream feeds. Subscriptions and blocks are + // HARD-deleted, so the gate row is the tombstone that rejects a stale + // cross-feed copy of the create arriving after the delete — without it, + // an unsubscribed user would be silently re-subscribed hours later. switch commit.Collection { case "social.coves.community.profile": - return c.handleCommunityProfile(ctx, event.Did, commit) + return applyGated(ctx, c.revGate, ConsumerCommunities, event.Did, commit, func() error { + return c.handleCommunityProfile(ctx, event.Did, commit) + }) case "social.coves.community.subscription": // Handle both create (subscribe) and delete (unsubscribe) operations - return c.handleSubscription(ctx, event.Did, commit) + return applyGated(ctx, c.revGate, ConsumerCommunities, event.Did, commit, func() error { + return c.handleSubscription(ctx, event.Did, commit) + }) case "social.coves.community.block": // Handle both create (block) and delete (unblock) operations - return c.handleBlock(ctx, event.Did, commit) + return applyGated(ctx, c.revGate, ConsumerCommunities, event.Did, commit, func() error { + return c.handleBlock(ctx, event.Did, commit) + }) default: // Not a community-related collection return nil @@ -673,7 +714,13 @@ func (c *CommunityEventConsumer) createSubscription(ctx context.Context, userDID // deleteSubscription removes a subscription from the index // DELETE operations don't include record data, so we need to look up the subscription -// by its URI to find which community the user unsubscribed from +// by its URI to find which community the user unsubscribed from. +// +// Cross-rkey safety: the rev gate is per record URI, so it cannot order a +// redriven unsubscribe of rkey A against a newer subscribe under rkey B. +// SubscribeWithCount therefore pins the row's record_uri to the NEWEST record +// (last-write-wins on conflict); the redriven delete of the old URI then finds +// no row here and is skipped instead of tearing down the valid subscription. func (c *CommunityEventConsumer) deleteSubscription(ctx context.Context, userDID string, commit *CommitEvent) error { // Build AT-URI from the rkey uri := fmt.Sprintf("at://%s/social.coves.community.subscription/%s", userDID, commit.RKey) diff --git a/internal/atproto/jetstream/feeds.go b/internal/atproto/jetstream/feeds.go new file mode 100644 index 0000000..392c9ee --- /dev/null +++ b/internal/atproto/jetstream/feeds.go @@ -0,0 +1,174 @@ +package jetstream + +import ( + "fmt" + "net/url" + "regexp" + "strings" +) + +// This file owns the multi-feed Jetstream configuration. The AppView consumes +// N Jetstream endpoints ("feeds") carrying overlapping repos — typically the +// public bsky.network Jetstream (kept for third-party PDS records and +// redundancy) plus the self-hosted relay+Jetstream pair that crawls our own +// PDSes without bsky.network's per-host quotas. Every consumer runs once per +// feed; rev-gating (rev_gate.go) makes the overlap safe. +// +// Configuration is one env var: +// +// JETSTREAM_FEEDS="bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080" +// +// Each entry is =. The base URL carries no query string; a +// path is optional (a trailing /subscribe is tolerated). The per-consumer +// collection filters live in consumerWantedCollections (exposed via +// WantedCollections) so adding feed N+1 is pure config and the filters exist +// in exactly one place. + +// Feed is one upstream Jetstream endpoint. +type Feed struct { + Key string // short name used in consumer names and logs, e.g. "bsky", "self" + BaseURL string // ws(s)://host[:port], optionally with a path; no query +} + +// PrimaryFeedKey is the feed whose consumers keep the bare legacy names +// ("users", "posts", ...) so their persisted cursors and dead letters carry +// over from the single-feed era. Every other feed's consumers are named +// "@". +const PrimaryFeedKey = "bsky" + +// feedKeyPattern keeps feed keys safe for use inside consumer names (which key +// cursor and dead-letter rows) and log lines. +var feedKeyPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`) + +// consumerWantedCollections maps each canonical consumer name to the record +// collections it indexes. The wiring loop appends these as wantedCollections +// query parameters to every feed's subscribe URL. Unexported on purpose: a +// direct map lookup with a mistyped consumer name would silently yield nil, +// and a filterless subscribe URL means consuming the ENTIRE firehose. Use +// WantedCollections, which fails closed on unknown names. +// +// NOTE: social.coves.community.block is listed for the communities consumer — +// its handler has always supported block records, but the old hand-written +// COMMUNITY_JETSTREAM_URL never subscribed to the collection, so firehose +// block events silently never arrived. Deriving URLs from this table fixes +// that class of drift. +var consumerWantedCollections = map[string][]string{ + ConsumerUsers: { + "social.coves.actor.profile", + "social.coves.actor.block", + }, + ConsumerCommunities: { + "social.coves.community.profile", + "social.coves.community.subscription", + "social.coves.community.block", + }, + ConsumerPosts: { + "social.coves.community.post", + }, + ConsumerAggregators: { + "social.coves.aggregator.service", + "social.coves.aggregator.authorization", + }, + ConsumerVotes: { + "social.coves.feed.vote", + }, + ConsumerComments: { + "social.coves.community.comment", + }, +} + +// WantedCollections returns a copy of the record collections the named +// canonical consumer indexes, for use as wantedCollections filters on a feed's +// subscribe URL. Unknown consumer names return an error rather than an empty +// slice, because an unfiltered subscribe URL would consume the entire firehose. +func WantedCollections(consumer string) ([]string, error) { + collections, ok := consumerWantedCollections[consumer] + if !ok { + return nil, fmt.Errorf("unknown Jetstream consumer %q: no wantedCollections defined (an unfiltered URL would subscribe to the whole firehose)", consumer) + } + return append([]string(nil), collections...), nil +} + +// ParseFeeds parses a JETSTREAM_FEEDS value into an ordered feed list. +// Format: semicolon-separated = entries. Keys must be unique, +// lowercase alphanumeric (plus hyphens); base URLs must be ws:// or wss:// +// and carry no query string (collection filters are code-owned). +func ParseFeeds(spec string) ([]Feed, error) { + var feeds []Feed + seen := make(map[string]bool) + + for _, entry := range strings.Split(spec, ";") { + entry = strings.TrimSpace(entry) + if entry == "" { + continue + } + key, base, found := strings.Cut(entry, "=") + key, base = strings.TrimSpace(key), strings.TrimSpace(base) + if !found || key == "" || base == "" { + return nil, fmt.Errorf("invalid feed entry %q (expected =)", entry) + } + if !feedKeyPattern.MatchString(key) { + return nil, fmt.Errorf("invalid feed key %q (lowercase alphanumeric and hyphens only)", key) + } + if seen[key] { + return nil, fmt.Errorf("duplicate feed key %q", key) + } + seen[key] = true + + parsed, err := url.Parse(base) + if err != nil { + return nil, fmt.Errorf("invalid feed URL %q: %w", base, err) + } + if parsed.Scheme != "ws" && parsed.Scheme != "wss" { + return nil, fmt.Errorf("feed URL %q must use ws:// or wss://", base) + } + if parsed.Host == "" { + return nil, fmt.Errorf("feed URL %q is missing a host", base) + } + if parsed.RawQuery != "" { + return nil, fmt.Errorf("feed URL %q must not carry a query string (collection filters are derived per consumer)", base) + } + + feeds = append(feeds, Feed{Key: key, BaseURL: base}) + } + + if len(feeds) == 0 { + return nil, fmt.Errorf("no feeds configured (expected e.g. %q)", + "bsky=wss://jetstream2.us-east.bsky.network;self=ws://tidepool-prod-jetstream:8080") + } + return feeds, nil +} + +// SubscribeURL builds the full WebSocket subscribe URL for one consumer on one +// feed: the base URL with a /subscribe path (appended unless already present) +// and one wantedCollections parameter per collection. +func SubscribeURL(baseURL string, collections []string) (string, error) { + parsed, err := url.Parse(baseURL) + if err != nil { + return "", fmt.Errorf("invalid feed URL %q: %w", baseURL, err) + } + if !strings.HasSuffix(parsed.Path, "/subscribe") { + parsed.Path = strings.TrimSuffix(parsed.Path, "/") + "/subscribe" + } + query := parsed.Query() + for _, collection := range collections { + query.Add("wantedCollections", collection) + } + parsed.RawQuery = query.Encode() + return parsed.String(), nil +} + +// FeedConsumerName returns the connector name — the key for the persisted +// cursor and dead-letter rows — for a consumer on a feed. The primary feed +// keeps the bare legacy name so live cursors carry over untouched; all other +// feeds get "@". A brand-new name starts with no cursor +// row, i.e. the consumer live-tails from now — a newly added feed starts with +// no history, and recovering records it never delivered requires the source +// PDSes to re-emit them through the relay (Tidepool's POST /admin/reemit +// rewrites records with fresh revs so they pass the rev gate). +func FeedConsumerName(consumer, feedKey string) string { + if feedKey == PrimaryFeedKey { + return consumer + } + return consumer + "@" + feedKey +} diff --git a/internal/atproto/jetstream/feeds_test.go b/internal/atproto/jetstream/feeds_test.go new file mode 100644 index 0000000..9c175e3 --- /dev/null +++ b/internal/atproto/jetstream/feeds_test.go @@ -0,0 +1,110 @@ +package jetstream + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseFeeds_TwoFeeds_OrderPreserved(t *testing.T) { + feeds, err := ParseFeeds("bsky=wss://jetstream2.us-east.bsky.network;self=ws://jetstream:6008") + require.NoError(t, err) + require.Len(t, feeds, 2) + assert.Equal(t, Feed{Key: "bsky", BaseURL: "wss://jetstream2.us-east.bsky.network"}, feeds[0]) + assert.Equal(t, Feed{Key: "self", BaseURL: "ws://jetstream:6008"}, feeds[1]) +} + +func TestParseFeeds_SingleFeedWithWhitespaceAndTrailingSemicolon(t *testing.T) { + feeds, err := ParseFeeds(" self = ws://localhost:6008 ; ") + require.NoError(t, err) + require.Len(t, feeds, 1) + assert.Equal(t, Feed{Key: "self", BaseURL: "ws://localhost:6008"}, feeds[0]) +} + +func TestParseFeeds_Rejections(t *testing.T) { + cases := map[string]string{ + "empty spec": "", + "missing equals": "bsky wss://jetstream2.us-east.bsky.network", + "missing key": "=ws://jetstream:6008", + "missing url": "self=", + "http scheme": "self=http://jetstream:6008", + "missing host": "self=ws://", + "query string in base": "self=ws://jetstream:6008?wantedCollections=social.coves.feed.vote", + "duplicate key": "self=ws://a:1;self=ws://b:2", + "uppercase key": "Self=ws://jetstream:6008", + "key with at sign": "my@feed=ws://jetstream:6008", + "only empty entries": ";;", + "whitespace inside a key": "my feed=ws://jetstream:6008", + } + for name, spec := range cases { + t.Run(name, func(t *testing.T) { + _, err := ParseFeeds(spec) + assert.Error(t, err, "spec %q must be rejected", spec) + }) + } +} + +func TestSubscribeURL_AppendsSubscribeAndCollections(t *testing.T) { + got, err := SubscribeURL("ws://jetstream:6008", []string{ + "social.coves.community.post", + }) + require.NoError(t, err) + assert.Equal(t, "ws://jetstream:6008/subscribe?wantedCollections=social.coves.community.post", got) +} + +func TestSubscribeURL_MultipleCollectionsRepeatParameter(t *testing.T) { + got, err := SubscribeURL("wss://jetstream2.us-east.bsky.network", []string{ + "social.coves.actor.profile", + "social.coves.actor.block", + }) + require.NoError(t, err) + assert.Equal(t, + "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=social.coves.actor.profile&wantedCollections=social.coves.actor.block", + got) +} + +func TestSubscribeURL_ExistingSubscribePathNotDuplicated(t *testing.T) { + got, err := SubscribeURL("ws://jetstream:6008/subscribe", []string{"social.coves.feed.vote"}) + require.NoError(t, err) + assert.Equal(t, "ws://jetstream:6008/subscribe?wantedCollections=social.coves.feed.vote", got) +} + +func TestFeedConsumerName_PrimaryFeedKeepsLegacyName(t *testing.T) { + // The bare names key live production cursors; renaming them would orphan + // the cursors and restart every consumer at the live tail. + assert.Equal(t, "comments", FeedConsumerName(ConsumerComments, "bsky")) + assert.Equal(t, "comments@self", FeedConsumerName(ConsumerComments, "self")) +} + +func TestWantedCollections_CoversEveryCanonicalConsumer(t *testing.T) { + for _, consumer := range []string{ + ConsumerUsers, ConsumerCommunities, ConsumerPosts, + ConsumerAggregators, ConsumerVotes, ConsumerComments, + } { + collections, err := WantedCollections(consumer) + require.NoError(t, err, "consumer %s must have wantedCollections", consumer) + assert.NotEmpty(t, collections, + "consumer %s has no wantedCollections; its per-feed URL would subscribe to the whole firehose", consumer) + } +} + +func TestWantedCollections_UnknownConsumerErrors(t *testing.T) { + // A silent nil here would build a filterless subscribe URL — i.e. the + // consumer would ingest the ENTIRE firehose. Unknown names must fail closed. + collections, err := WantedCollections("no-such-consumer") + assert.Error(t, err) + assert.Nil(t, collections) +} + +func TestWantedCollections_ReturnsACopy(t *testing.T) { + first, err := WantedCollections(ConsumerPosts) + require.NoError(t, err) + require.NotEmpty(t, first) + first[0] = "mutated.collection" + + second, err := WantedCollections(ConsumerPosts) + require.NoError(t, err) + assert.Equal(t, "social.coves.community.post", second[0], + "WantedCollections must return a copy; callers must not be able to mutate the canonical table") +} diff --git a/internal/atproto/jetstream/post_consumer.go b/internal/atproto/jetstream/post_consumer.go index aa201c2..91af9d1 100644 --- a/internal/atproto/jetstream/post_consumer.go +++ b/internal/atproto/jetstream/post_consumer.go @@ -65,6 +65,11 @@ func NewPostEventConsumer( return c } +// RevGated reports whether this consumer applies the per-record rev gate; always true +// for posts (gating is hardwired via c.db). main.go checks this at boot to refuse +// multi-feed operation with an ungated consumer. +func (c *PostEventConsumer) RevGated() bool { return true } + // HandleEvent processes a Jetstream event for post records // Handles CREATE, UPDATE, and DELETE operations func (c *PostEventConsumer) HandleEvent(ctx context.Context, event *JetstreamEvent) error { @@ -226,8 +231,8 @@ func (c *PostEventConsumer) createPost(ctx context.Context, repoDID string, comm post.ContentLabels = &labelsStr } - // Atomically: Index post + Reconcile comment count for out-of-order arrivals - if err := c.indexPostAndReconcileCounts(ctx, post); err != nil { + // Atomically: Rev-gate + Index post + Reconcile comment count for out-of-order arrivals + if err := c.indexPostAndReconcileCounts(ctx, post, commit.Rev); err != nil { return fmt.Errorf("failed to index post and reconcile counts: %w", err) } @@ -243,11 +248,43 @@ func (c *PostEventConsumer) deletePost(ctx context.Context, repoDID string, comm // Format: at://community_did/social.coves.community.post/rkey uri := fmt.Sprintf("at://%s/social.coves.community.post/%s", repoDID, commit.RKey) - // Soft delete the post in AppView - if err := c.postRepo.SoftDelete(ctx, uri); err != nil { + // 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 + // arriving later and resurrecting the post). The gate row is advanced even + // when the post was never indexed, so the late create of an already-deleted + // record is rejected too. + tx, err := c.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer func() { + if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { + log.Printf("Failed to rollback transaction: %v", rollbackErr) + } + }() + + won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + if err != nil { + return err + } + if !won { + logSkippedStaleRev(ConsumerPosts, "delete", uri, commit.Rev) + return nil + } + + // Same statement as postRepo.SoftDelete, inlined for transactionality. + // Idempotent: zero rows (already deleted or never indexed) is success. + if _, err := tx.ExecContext(ctx, + `UPDATE posts SET deleted_at = NOW() WHERE uri = $1 AND deleted_at IS NULL`, uri, + ); err != nil { return fmt.Errorf("failed to soft delete post: %w", err) } + 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) return nil } @@ -441,7 +478,32 @@ func (c *PostEventConsumer) updatePost(ctx context.Context, repoDID string, comm WHERE id = $1 AND deleted_at IS NULL AND ($11::bigint <= 0 OR indexed_at < to_timestamp($11::bigint / 1000000.0)) ` - result, err := c.db.ExecContext(ctx, updateQuery, + // REV GATE + UPDATE in one transaction. The gate (strictly-newer rev wins) + // is the cross-feed ordering guard: the time_us recency guard in the WHERE + // clause below cannot reject a stale copy delivered by ANOTHER feed, because + // each feed stamps its own emission time — a pre-edit update replayed by the + // 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) + } + defer func() { + if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { + log.Printf("Failed to rollback transaction: %v", rollbackErr) + } + }() + + won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + if err != nil { + return err + } + if !won { + logSkippedStaleRev(ConsumerPosts, "update", uri, commit.Rev) + return nil + } + + result, err := tx.ExecContext(ctx, updateQuery, storedID, commit.CID, postRecord.Title, postRecord.Content, facetsJSON, embedJSON, labelsJSON, incomingUp, incomingDown, incomingAsOf, @@ -461,10 +523,16 @@ func (c *PostEventConsumer) updatePost(ctx context.Context, repoDID string, comm return 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 } + if err := tx.Commit(); err != nil { + return 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) } else { @@ -486,7 +554,7 @@ func parseBridgedAsOf(asOf, uri string) (time.Time, error) { // 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) error { +func (c *PostEventConsumer) indexPostAndReconcileCounts(ctx context.Context, post *posts.Post, rev string) error { tx, err := c.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) @@ -497,6 +565,20 @@ func (c *PostEventConsumer) indexPostAndReconcileCounts(ctx context.Context, pos } }() + // 0. REV GATE: apply this create only if its rev is strictly newer than the + // last applied event for this record — rejects duplicate replays (equal rev) + // and stale cross-feed copies of a create arriving after the post's delete + // (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 + } + if !won { + logSkippedStaleRev(ConsumerPosts, "create", post.URI, rev) + return nil + } + // 1. Insert the post (idempotent with RETURNING clause) var facetsJSON, embedJSON, labelsJSON sql.NullString @@ -542,6 +624,18 @@ func (c *PostEventConsumer) indexPostAndReconcileCounts(ctx context.Context, pos // If no rows returned, post already exists (idempotent - OK for Jetstream replays) if insertErr == sql.ErrNoRows { + // KNOWN LIMITATION (accepted): a genuine RE-CREATE of the same rkey while + // the row is still ACTIVE also lands here and is treated as an idempotent + // duplicate, dropping the new content. Reaching that state requires the + // exact sequence: create A applied → delete dead-lettered (failed, never + // applied) → re-create B (same rkey, strictly newer rev) arrives while + // the row is still active. B's content is never applied, the gate + // advances to B's rev, and the redriven delete A is then gate-rejected — + // the row survives with A's content. This needs a dead-lettered delete + // AND an rkey reuse inside the redrive window; rare enough to document + // rather than plumb a full content upsert through the create path + // (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) diff --git a/internal/atproto/jetstream/rev_gate.go b/internal/atproto/jetstream/rev_gate.go new file mode 100644 index 0000000..97c50b4 --- /dev/null +++ b/internal/atproto/jetstream/rev_gate.go @@ -0,0 +1,201 @@ +package jetstream + +import ( + "context" + "database/sql" + "fmt" + "log" +) + +// This file implements rev-gating: the ordering guard that makes it safe to +// consume multiple Jetstream feeds carrying the same repos (see migration +// 033_create_jetstream_record_revs.sql for the full rationale). +// +// Every commit event carries rev, the repo's monotonic TID (lexicographically +// ordered string). The last APPLIED rev per record URI is stored in +// jetstream_record_revs; an incoming create/update/delete applies only when +// its rev is strictly greater. Equal rev = the same event replayed (no-op, +// subsumes duplicate handling); smaller rev = a stale cross-feed copy +// (skipped). The gate row survives hard deletes, acting as a tombstone that +// rejects the stale create that would otherwise resurrect a deleted record. +// +// Two integration patterns, chosen per consumer: +// +// 1. Transactional (posts, comments, votes — the count-mutating consumers): +// tryAdvanceRecordRev runs as the FIRST statement of the consumer's +// existing transaction. The conditional upsert takes a row lock, so +// concurrent handlers of the same record serialize, and a rollback +// reverts the gate together with the writes. Fully atomic. +// +// 2. Transactional claim held across apply (users' blocks, communities, +// aggregators — the repo-method consumers, via applyGated): the gate row +// is claimed as the FIRST statement of a dedicated transaction on the +// gate's own DB handle, apply() runs while that claim (a row lock on the +// record's gate row) is held, and the transaction commits only after +// apply succeeds. Same-URI handlers on other feeds serialize on the gate +// row lock for the full duration of apply, so there is no check→write +// window for a concurrent feed to interleave in; an apply failure rolls +// the claim back un-advanced so retries/redrives replay the event. +// Exception: the user consumer's PROFILE path keeps the older +// check→write→advance protocol (RevGate.IsStale before the write, +// RevGate.Advance after) because its write is an idempotent last-write- +// wins profile update where a brief unguarded window is acceptable. +// +// Events with an empty rev bypass the gate, preserving previous behavior. +// In practice only synthetic test events are rev-less: real Jetstream frames +// always carry rev, and dead letters store the raw frame, so redriven events +// keep theirs. + +// revGateQuerier is the subset of *sql.DB / *sql.Tx the gate needs, so the +// same statements run standalone or inside a consumer's transaction. +type revGateQuerier interface { + ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) + QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row +} + +// tryAdvanceRecordRev atomically claims rev for the record URI. It returns +// true when the event wins (no gate row yet, or a strictly greater rev) and +// the gate row was written; false when the stored rev is greater or equal — +// a duplicate replay or a stale cross-feed copy the caller must skip. +// An empty rev bypasses the gate (returns true without writing). +func tryAdvanceRecordRev(ctx context.Context, q revGateQuerier, uri, rev string) (bool, error) { + if rev == "" { + return true, nil + } + result, err := q.ExecContext(ctx, ` + INSERT INTO jetstream_record_revs (record_uri, rev) + VALUES ($1, $2) + ON CONFLICT (record_uri) DO UPDATE + SET rev = EXCLUDED.rev, updated_at = NOW() + WHERE jetstream_record_revs.rev < EXCLUDED.rev + `, uri, rev) + if err != nil { + return false, fmt.Errorf("failed to advance record rev for %s: %w", uri, err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("failed to check record rev advance for %s: %w", uri, err) + } + return rows > 0, nil +} + +// recordRevIsStale reports whether the stored rev for the record URI is +// greater than or equal to the incoming one (i.e. the event must be skipped). +// No gate row, or an empty incoming rev, means not stale. +func recordRevIsStale(ctx context.Context, q revGateQuerier, uri, rev string) (bool, error) { + if rev == "" { + return false, nil + } + var stale bool + err := q.QueryRowContext(ctx, + `SELECT rev >= $2 FROM jetstream_record_revs WHERE record_uri = $1`, uri, rev, + ).Scan(&stale) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, fmt.Errorf("failed to check record rev for %s: %w", uri, err) + } + return stale, nil +} + +// logSkippedStaleRev is the single, grep-able log line for gate skips. +// Rejected stale events are the system WORKING (e.g. the bsky feed's delayed +// copies of self-feed events); this line makes that observable and +// distinguishable from real trouble. +func logSkippedStaleRev(consumer, operation, uri, rev string) { + log.Printf("rev-gate: %s skipped stale %s for %s (incoming rev %q <= stored)", consumer, operation, uri, rev) +} + +// commitRecordURI builds the AT-URI of the record a commit event addresses. +func commitRecordURI(did string, commit *CommitEvent) string { + return fmt.Sprintf("at://%s/%s/%s", did, commit.Collection, commit.RKey) +} + +// RevGate carries the gate's own DB handle for the consumers that write +// through repository methods and therefore cannot join their writes into one +// transaction with the gate. applyGated uses it to open the claim transaction +// held across apply; IsStale/Advance expose the non-transactional +// check→write→advance flavor still used by the user consumer's profile path. +// A nil *RevGate disables gating entirely (tests, deployments consuming a +// single feed). +type RevGate struct { + db *sql.DB +} + +// NewRevGate creates a rev gate backed by the AppView database. +func NewRevGate(db *sql.DB) *RevGate { + return &RevGate{db: db} +} + +// IsStale reports whether the event's rev is superseded by the stored rev +// for the record URI. Nil-safe: a nil gate never reports stale. +func (g *RevGate) IsStale(ctx context.Context, uri, rev string) (bool, error) { + if g == nil { + return false, nil + } + return recordRevIsStale(ctx, g.db, uri, rev) +} + +// Advance records rev as the last applied rev for the record URI, keeping +// whichever is greater. Called AFTER the consumer's idempotent write +// succeeds, so a failure in between replays the event instead of losing it. +// Nil-safe: a nil gate is a no-op. +func (g *RevGate) Advance(ctx context.Context, uri, rev string) error { + if g == nil { + return nil + } + _, err := tryAdvanceRecordRev(ctx, g.db, uri, rev) + return err +} + +// applyGated runs apply under a TRANSACTIONAL rev-gate claim for a commit +// event. The gate row is claimed (tryAdvanceRecordRev) as the first statement +// of a transaction on the gate's own DB handle, apply() runs while that claim +// is held, and the transaction commits only after apply succeeds. The claim's +// row lock makes two feeds' handlers for the SAME record URI serialize for +// the full duration of apply — the loser blocks on the claim, then observes +// the winner's rev and skips — so there is no check→write window for a stale +// cross-feed copy to sneak through, no matter how long apply takes. +// +// Deadlock note: apply's writes go through repository methods on their own +// connections, which is deliberate and safe — the gate transaction touches +// ONLY jetstream_record_revs, a table no repository write path ever touches, +// so the gate row lock acts as a pure per-record mutex around apply. +// +// An apply error (or panic — the deferred rollback covers both) releases the +// claim WITHOUT advancing, so the connector's retry/redrive replays the event +// instead of losing it behind its own gate entry. Events with an empty rev +// (synthetic test events, legacy dead letters) bypass the gate and run apply +// directly, as does a nil gate. +func applyGated(ctx context.Context, gate *RevGate, consumer, did string, commit *CommitEvent, apply func() error) error { + if gate == nil || commit.Rev == "" { + return apply() + } + uri := commitRecordURI(did, commit) + tx, err := gate.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin rev-gate transaction for %s: %w", uri, err) + } + defer func() { + if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { + log.Printf("Failed to rollback rev-gate transaction for %s: %v", uri, rollbackErr) + } + }() + + won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) + if err != nil { + return err + } + if !won { + logSkippedStaleRev(consumer, commit.Operation, uri, commit.Rev) + return nil + } + if err := apply(); err != nil { + return err // deferred rollback releases the claim un-advanced + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit rev-gate transaction for %s: %w", uri, err) + } + return nil +} diff --git a/internal/atproto/jetstream/rev_gate_test.go b/internal/atproto/jetstream/rev_gate_test.go new file mode 100644 index 0000000..4aaad08 --- /dev/null +++ b/internal/atproto/jetstream/rev_gate_test.go @@ -0,0 +1,561 @@ +package jetstream + +import ( + "context" + "database/sql" + "testing" + "time" + + "Coves/internal/core/users" + "Coves/internal/db/postgres" + + _ "github.com/lib/pq" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests prove the rev gate restores per-repo commit ordering across +// MULTIPLE Jetstream feeds carrying the same repos. The adversarial +// interleavings below cannot be produced by a single feed (which delivers a +// repo's events in commit order) — they model the lagging bsky.network feed +// replaying a repo's history HOURS after the self-hosted feed already +// processed newer events. Crucially, the replayed copies carry NEWER time_us +// values (each feed stamps its own emission time), so the pre-existing +// time-based recency guards cannot reject them; only rev — assigned by the +// repo itself and monotonic per repo — orders events across feeds. + +const ( + revTestPrefix = "did:plc:jsrev" + revTestCommunity = revTestPrefix + "community" + revTestAuthor = revTestPrefix + "author" + revTestCommenter = revTestPrefix + "commenter" + revTestVoter = revTestPrefix + "voter" + + // TIDs are lexicographically ordered strings; these model three + // successive commits of one repo. + revA = "3lrevtestaa2a" + revB = "3lrevtestaa2b" + revC = "3lrevtestaa2c" +) + +func cleanupRevTestData(t *testing.T, db *sql.DB) { + t.Helper() + // Gate rows persist across runs BY DESIGN (they are tombstones); tests + // must clear their own or a re-run would reject its fixture creates. + _, _ = db.Exec("DELETE FROM jetstream_record_revs WHERE record_uri LIKE $1", "at://"+revTestPrefix+"%") + _, _ = db.Exec("DELETE FROM votes WHERE voter_did LIKE $1", revTestPrefix+"%") + _, _ = db.Exec("DELETE FROM comments WHERE commenter_did LIKE $1 OR root_uri LIKE $2", revTestPrefix+"%", "at://"+revTestPrefix+"%") + _, _ = db.Exec("DELETE FROM posts WHERE community_did LIKE $1", revTestPrefix+"%") + _, _ = db.Exec("DELETE FROM community_subscriptions WHERE user_did LIKE $1", revTestPrefix+"%") + _, _ = db.Exec("DELETE FROM communities WHERE did LIKE $1", revTestPrefix+"%") + _, _ = db.Exec("DELETE FROM users WHERE did LIKE $1", revTestPrefix+"%") +} + +// revCommitEvent builds a commit event carrying a rev, the one field the +// duplicate-delivery helpers omit. +func revCommitEvent(did, collection, op, rkey, rev, cid string, timeUS int64, record map[string]interface{}) *JetstreamEvent { + return &JetstreamEvent{ + Kind: "commit", + Did: did, + TimeUS: timeUS, + Commit: &CommitEvent{ + Rev: rev, + Operation: op, + Collection: collection, + RKey: rkey, + CID: cid, + Record: record, + }, + } +} + +// setupRevFixtures indexes the shared user/community fixtures plus one post, +// returning the post's URI/CID and a post consumer wired to the fixtures. +func setupRevFixtures(t *testing.T, db *sql.DB) (pc *PostEventConsumer, postURI, postCID string) { + t.Helper() + insertBridgedUser(t, db, revTestAuthor, "revauthor.test") + insertBridgedUser(t, db, revTestCommenter, "revcommenter.test") + insertBridgedUser(t, db, revTestVoter, "revvoter.test") + insertBridgedCommunity(t, db, revTestCommunity, "revcommunity.test", revTestAuthor) + + us := newMockUserService() + us.users[revTestAuthor] = &users.User{DID: revTestAuthor, Handle: "revauthor.test"} + pc = NewPostEventConsumer(postgres.NewPostRepository(db), postgres.NewCommunityRepository(db), us, db) + + postURI = "at://" + revTestCommunity + "/social.coves.community.post/revpost1" + postCID = "bafrevpost1" + require.NoError(t, pc.HandleEvent(context.Background(), revCommitEvent( + revTestCommunity, "social.coves.community.post", "create", "revpost1", revA, postCID, + time.Now().UnixMicro(), + map[string]interface{}{ + "$type": "social.coves.community.post", + "community": revTestCommunity, + "author": revTestAuthor, + "title": "rev target", + "content": "post v1", + "createdAt": "2026-03-01T00:00:00Z", + }, + ))) + return pc, postURI, postCID +} + +func revCommentRecord(content, rootURI, rootCID, parentURI, parentCID string) map[string]interface{} { + return map[string]interface{}{ + "$type": CommentCollection, + "content": content, + "reply": map[string]interface{}{ + "root": map[string]interface{}{"uri": rootURI, "cid": rootCID}, + "parent": map[string]interface{}{"uri": parentURI, "cid": parentCID}, + }, + "createdAt": "2026-03-01T02:00:00Z", + } +} + +func TestRevGate_AdvanceAndStalenessSemantics(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + ctx := context.Background() + uri := "at://" + revTestPrefix + "gate/social.coves.feed.vote/g1" + + // First writer wins. + won, err := tryAdvanceRecordRev(ctx, db, uri, revB) + require.NoError(t, err) + assert.True(t, won, "first rev must win") + + // Equal rev = the same event replayed = no-op. + won, err = tryAdvanceRecordRev(ctx, db, uri, revB) + require.NoError(t, err) + assert.False(t, won, "equal rev must be rejected (duplicate replay)") + + // Lower rev = stale cross-feed copy. + won, err = tryAdvanceRecordRev(ctx, db, uri, revA) + require.NoError(t, err) + assert.False(t, won, "lower rev must be rejected (stale copy)") + + // Higher rev = genuinely newer commit. + won, err = tryAdvanceRecordRev(ctx, db, uri, revC) + require.NoError(t, err) + assert.True(t, won, "higher rev must win") + + // Read-side check agrees. + gate := NewRevGate(db) + stale, err := gate.IsStale(ctx, uri, revB) + require.NoError(t, err) + assert.True(t, stale) + stale, err = gate.IsStale(ctx, uri, revC) + require.NoError(t, err) + assert.True(t, stale, "equal rev is stale (already applied)") + + // Empty rev bypasses the gate entirely (synthetic/legacy events). + won, err = tryAdvanceRecordRev(ctx, db, uri, "") + require.NoError(t, err) + assert.True(t, won, "rev-less events bypass the gate") + stale, err = gate.IsStale(ctx, uri, "") + require.NoError(t, err) + assert.False(t, stale, "rev-less events are never stale") + + // A nil gate is inert. + var nilGate *RevGate + stale, err = nilGate.IsStale(ctx, uri, revA) + require.NoError(t, err) + assert.False(t, stale) + require.NoError(t, nilGate.Advance(ctx, uri, revC)) +} + +// The zombie-resurrection interleaving: create → delete via the fast feed, +// then the lagging feed replays the original create with an older rev but a +// NEWER time_us. Without the gate, the resurrection branch restores the +// deleted comment permanently. +func TestCommentConsumer_StaleCreateReplayAfterDelete_DoesNotResurrect(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + _, postURI, postCID := setupRevFixtures(t, db) + cc := NewCommentEventConsumer(postgres.NewCommentRepository(db), db) + ctx := context.Background() + base := time.Now().UnixMicro() + + commentURI := "at://" + revTestCommenter + "/" + CommentCollection + "/zomb1" + record := revCommentRecord("hello", postURI, postCID, postURI, postCID) + + // Fast feed: create, then delete. + require.NoError(t, cc.HandleEvent(ctx, revCommitEvent( + revTestCommenter, CommentCollection, "create", "zomb1", revA, "bafzomb1", base, record))) + require.NoError(t, cc.HandleEvent(ctx, revCommitEvent( + revTestCommenter, CommentCollection, "delete", "zomb1", revB, "", base+1_000_000, nil))) + + var deletedAt *time.Time + require.NoError(t, db.QueryRow(`SELECT deleted_at FROM comments WHERE uri=$1`, commentURI).Scan(&deletedAt)) + require.NotNil(t, deletedAt, "fixture: comment soft-deleted") + + // Lagging feed: the original create replayed hours later — older rev, + // NEWER time_us. + require.NoError(t, cc.HandleEvent(ctx, revCommitEvent( + revTestCommenter, CommentCollection, "create", "zomb1", revA, "bafzomb1", base+2_000_000, record))) + + require.NoError(t, db.QueryRow(`SELECT deleted_at FROM comments WHERE uri=$1`, commentURI).Scan(&deletedAt)) + assert.NotNil(t, deletedAt, "stale create replay must NOT resurrect the deleted comment") + + var commentCount int + require.NoError(t, db.QueryRow(`SELECT comment_count FROM posts WHERE uri=$1`, postURI).Scan(&commentCount)) + assert.Equal(t, 1, commentCount, "stale create replay must not re-increment comment_count") +} + +// A genuine re-creation of the same rkey carries a fresh, HIGHER rev and must +// still pass the gate and resurrect the row — proving the gate rejects only +// stale copies, not the legitimate atProto recreate-same-rkey flow. +func TestCommentConsumer_GenuineRecreateSameRKey_StillResurrects(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + _, postURI, postCID := setupRevFixtures(t, db) + cc := NewCommentEventConsumer(postgres.NewCommentRepository(db), db) + ctx := context.Background() + base := time.Now().UnixMicro() + + commentURI := "at://" + revTestCommenter + "/" + CommentCollection + "/resur1" + + require.NoError(t, cc.HandleEvent(ctx, revCommitEvent( + revTestCommenter, CommentCollection, "create", "resur1", revA, "bafresur1", base, + revCommentRecord("first life", postURI, postCID, postURI, postCID)))) + require.NoError(t, cc.HandleEvent(ctx, revCommitEvent( + revTestCommenter, CommentCollection, "delete", "resur1", revB, "", base+1_000_000, nil))) + require.NoError(t, cc.HandleEvent(ctx, revCommitEvent( + revTestCommenter, CommentCollection, "create", "resur1", revC, "bafresur2", base+2_000_000, + revCommentRecord("second life", postURI, postCID, postURI, postCID)))) + + var deletedAt *time.Time + var content string + require.NoError(t, db.QueryRow(`SELECT deleted_at, content FROM comments WHERE uri=$1`, commentURI).Scan(&deletedAt, &content)) + assert.Nil(t, deletedAt, "genuine re-creation (higher rev) must resurrect the comment") + assert.Equal(t, "second life", content) +} + +// The stale-update-clobber interleaving for posts: two successive edits via +// the fast feed, then the lagging feed replays the FIRST edit with a newer +// time_us. The time-based recency guard passes it; only the rev gate rejects +// it. Without the gate the content regresses until the next organic edit. +func TestPostConsumer_StaleUpdateReplay_DoesNotClobberContent(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + pc, postURI, _ := setupRevFixtures(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + update := func(rev, content string, timeUS int64) *JetstreamEvent { + return revCommitEvent(revTestCommunity, "social.coves.community.post", "update", "revpost1", rev, "baf"+rev, + timeUS, map[string]interface{}{ + "$type": "social.coves.community.post", + "community": revTestCommunity, + "author": revTestAuthor, + "title": "rev target", + "content": content, + "createdAt": "2026-03-01T00:00:00Z", + }) + } + + // Fast feed: edit to v2, then to v3. + require.NoError(t, pc.HandleEvent(ctx, update(revB, "post v2", base+1_000_000))) + require.NoError(t, pc.HandleEvent(ctx, update(revC, "post v3", base+2_000_000))) + + // Lagging feed: the v2 edit replayed with a NEWER time_us. + require.NoError(t, pc.HandleEvent(ctx, update(revB, "post v2", base+3_000_000))) + + var content string + require.NoError(t, db.QueryRow(`SELECT content FROM posts WHERE uri=$1`, postURI).Scan(&content)) + assert.Equal(t, "post v3", content, "stale update replay must not regress post content") +} + +// Same interleaving for comments. +func TestCommentConsumer_StaleUpdateReplay_DoesNotClobberContent(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + _, postURI, postCID := setupRevFixtures(t, db) + cc := NewCommentEventConsumer(postgres.NewCommentRepository(db), db) + ctx := context.Background() + base := time.Now().UnixMicro() + + commentURI := "at://" + revTestCommenter + "/" + CommentCollection + "/edit1" + edit := func(rev, content string, timeUS int64) *JetstreamEvent { + op := "update" + if rev == revA { + op = "create" + } + return revCommitEvent(revTestCommenter, CommentCollection, op, "edit1", rev, "baf"+rev, + timeUS, revCommentRecord(content, postURI, postCID, postURI, postCID)) + } + + require.NoError(t, cc.HandleEvent(ctx, edit(revA, "comment v1", base))) + require.NoError(t, cc.HandleEvent(ctx, edit(revB, "comment v2", base+1_000_000))) + require.NoError(t, cc.HandleEvent(ctx, edit(revC, "comment v3", base+2_000_000))) + + // Lagging feed replays the v2 edit with a NEWER time_us. + replay := edit(revB, "comment v2", base+3_000_000) + replay.Commit.Operation = "update" + require.NoError(t, cc.HandleEvent(ctx, replay)) + + var content string + require.NoError(t, db.QueryRow(`SELECT content FROM comments WHERE uri=$1`, commentURI).Scan(&content)) + assert.Equal(t, "comment v3", content, "stale update replay must not regress comment content") +} + +// The phantom-vote interleaving: vote → unvote via the fast feed, then the +// lagging feed replays the vote's create. Without the gate the vote row is +// re-indexed and the count re-incremented, permanently. +func TestVoteConsumer_StaleCreateReplayAfterDelete_NoPhantomVote(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + _, postURI, postCID := setupRevFixtures(t, db) + vc := NewVoteEventConsumer(postgres.NewVoteRepository(db), newMockUserService(), db) + ctx := context.Background() + base := time.Now().UnixMicro() + + voteRecord := map[string]interface{}{ + "subject": map[string]interface{}{"uri": postURI, "cid": postCID}, + "direction": "up", + "createdAt": "2026-03-01T01:00:00Z", + } + + require.NoError(t, vc.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.feed.vote", "create", "rv1", revA, "bafrv1", base, voteRecord))) + require.NoError(t, vc.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.feed.vote", "delete", "rv1", revB, "", base+1_000_000, nil))) + + // Lagging feed: the vote's create replayed after the unvote. + require.NoError(t, vc.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.feed.vote", "create", "rv1", revA, "bafrv1", base+2_000_000, voteRecord))) + + var upvotes, score int + require.NoError(t, db.QueryRow(`SELECT upvote_count, score FROM posts WHERE uri=$1`, postURI).Scan(&upvotes, &score)) + assert.Equal(t, 0, upvotes, "stale vote-create replay must not restore a phantom vote") + assert.Equal(t, 0, score) + + var activeVotes int + require.NoError(t, db.QueryRow( + `SELECT COUNT(*) FROM votes WHERE voter_did=$1 AND deleted_at IS NULL`, revTestVoter).Scan(&activeVotes)) + assert.Equal(t, 0, activeVotes) +} + +// Delete arriving for a never-indexed vote must still tombstone the record's +// rev, so the create's late copy cannot index a vote whose record no longer +// exists on the PDS. +func TestVoteConsumer_DeleteBeforeCreate_TombstoneRejectsLateCreate(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + _, postURI, postCID := setupRevFixtures(t, db) + vc := NewVoteEventConsumer(postgres.NewVoteRepository(db), newMockUserService(), db) + ctx := context.Background() + base := time.Now().UnixMicro() + + // Delete first (create was lost/never delivered on this feed). + require.NoError(t, vc.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.feed.vote", "delete", "rv2", revB, "", base, nil))) + + // The create's copy arrives later from the lagging feed. + require.NoError(t, vc.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.feed.vote", "create", "rv2", revA, "bafrv2", base+1_000_000, + map[string]interface{}{ + "subject": map[string]interface{}{"uri": postURI, "cid": postCID}, + "direction": "up", + "createdAt": "2026-03-01T01:00:00Z", + }))) + + var voteRows int + require.NoError(t, db.QueryRow( + `SELECT COUNT(*) FROM votes WHERE voter_did=$1`, revTestVoter).Scan(&voteRows)) + assert.Equal(t, 0, voteRows, "create arriving after the record's delete must not be indexed") + + var upvotes int + require.NoError(t, db.QueryRow(`SELECT upvote_count FROM posts WHERE uri=$1`, postURI).Scan(&upvotes)) + assert.Equal(t, 0, upvotes) +} + +// deletePost coverage: a delete arriving for a never-indexed post (its create +// was lost or is still in flight on the other feed) must still tombstone the +// record's rev, so the create's late copy cannot index a post whose record no +// longer exists on the PDS. +func TestPostConsumer_DeleteBeforeCreate_TombstoneRejectsLateCreate(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + pc, _, _ := setupRevFixtures(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + postURI := "at://" + revTestCommunity + "/social.coves.community.post/revpost2" + + // Delete first (create never delivered on this feed). + require.NoError(t, pc.HandleEvent(ctx, revCommitEvent( + revTestCommunity, "social.coves.community.post", "delete", "revpost2", revB, "", base, nil))) + + // The create's copy arrives later from the lagging feed — older rev, + // NEWER time_us. + require.NoError(t, pc.HandleEvent(ctx, revCommitEvent( + revTestCommunity, "social.coves.community.post", "create", "revpost2", revA, "bafrevpost2", base+1_000_000, + map[string]interface{}{ + "$type": "social.coves.community.post", + "community": revTestCommunity, + "author": revTestAuthor, + "title": "late create", + "content": "must not be indexed", + "createdAt": "2026-03-01T00:00:00Z", + }))) + + var postRows int + require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM posts WHERE uri=$1`, postURI).Scan(&postRows)) + assert.Equal(t, 0, postRows, "create arriving after the record's delete must not be indexed") +} + +// The zombie-resurrection interleaving for posts: create → delete via the +// fast feed, then the lagging feed replays the original create with an older +// rev but a NEWER time_us. The tombstoned delete rev must keep the post dead. +func TestPostConsumer_StaleCreateReplayAfterDelete_DoesNotResurrect(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + // setupRevFixtures indexes revpost1 with revA / CID bafrevpost1. + pc, postURI, _ := setupRevFixtures(t, db) + ctx := context.Background() + base := time.Now().UnixMicro() + + // Fast feed: delete the post. + require.NoError(t, pc.HandleEvent(ctx, revCommitEvent( + revTestCommunity, "social.coves.community.post", "delete", "revpost1", revB, "", base+1_000_000, nil))) + + var deletedAt *time.Time + require.NoError(t, db.QueryRow(`SELECT deleted_at FROM posts WHERE uri=$1`, postURI).Scan(&deletedAt)) + require.NotNil(t, deletedAt, "fixture: post soft-deleted") + + // Lagging feed: the original create replayed hours later — older rev, + // NEWER time_us. + require.NoError(t, pc.HandleEvent(ctx, revCommitEvent( + revTestCommunity, "social.coves.community.post", "create", "revpost1", revA, "bafrevpost1", base+2_000_000, + map[string]interface{}{ + "$type": "social.coves.community.post", + "community": revTestCommunity, + "author": revTestAuthor, + "title": "rev target", + "content": "post v1", + "createdAt": "2026-03-01T00:00:00Z", + }))) + + require.NoError(t, db.QueryRow(`SELECT deleted_at FROM posts WHERE uri=$1`, postURI).Scan(&deletedAt)) + assert.NotNil(t, deletedAt, "stale create replay must NOT resurrect the deleted post") +} + +// The user consumer's profile gating is hand-rolled (check→write→advance in +// handleProfileCommit, not applyGated). This proves the gate rejects a stale +// cross-feed profile replay that the wall-clock recency guard cannot: the +// replay carries an OLDER rev but a NEWER time_us. +func TestUserConsumer_StaleProfileUpdateReplay_DoesNotRegressProfile(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + // The did shares revTestPrefix so cleanupRevTestData clears the gate row. + const did = revTestPrefix + "profileuser" + profileURI := "at://" + did + "/social.coves.actor.profile/self" + + mockService := newMockUserService() + mockService.users[did] = &users.User{DID: did, Handle: "revprofile.test"} + consumer := NewUserEventConsumer(mockService, &mockIdentityResolverForUser{}, + WithUserRevGate(NewRevGate(db))) + ctx := context.Background() + base := time.Now().UnixMicro() + + profileEvent := func(rev, displayName string, timeUS int64) *JetstreamEvent { + return revCommitEvent(did, CovesProfileCollection, "update", "self", rev, "baf"+rev, timeUS, + map[string]interface{}{"displayName": displayName}) + } + + // Fast feed: the current profile state (v2). + require.NoError(t, consumer.HandleEvent(ctx, profileEvent(revB, "v2", base))) + require.Equal(t, "v2", mockService.users[did].DisplayName, "fixture: v2 applied") + + // Lagging feed: the pre-edit update replayed — older rev, NEWER time_us. + require.NoError(t, consumer.HandleEvent(ctx, profileEvent(revA, "v1", base+2_000_000))) + + assert.Equal(t, "v2", mockService.users[did].DisplayName, + "stale profile replay must not regress the profile") + + var storedRev string + require.NoError(t, db.QueryRow( + `SELECT rev FROM jetstream_record_revs WHERE record_uri=$1`, profileURI).Scan(&storedRev)) + assert.Equal(t, revB, storedRev, "gate row must still hold the newer applied rev") +} + +// The zombie-subscription interleaving. Subscriptions are HARD-deleted, so +// only the gate's surviving row can reject the stale subscribe replay — this +// is the case a per-table rev column could never cover. +func TestCommunityConsumer_StaleSubscribeReplayAfterUnsubscribe_DoesNotResubscribe(t *testing.T) { + db := setupBridgedTestDB(t) + defer func() { _ = db.Close() }() + defer cleanupRevTestData(t, db) + cleanupRevTestData(t, db) + + insertBridgedUser(t, db, revTestVoter, "revsubscriber.test") + insertBridgedUser(t, db, revTestAuthor, "revauthor.test") + insertBridgedCommunity(t, db, revTestCommunity, "revcommunity.test", revTestAuthor) + + cec := NewCommunityEventConsumer(postgres.NewCommunityRepository(db), "did:web:test.local", true, nil, + WithCommunityRevGate(NewRevGate(db))) + ctx := context.Background() + base := time.Now().UnixMicro() + + subRecord := map[string]interface{}{ + "$type": "social.coves.community.subscription", + "subject": revTestCommunity, + "createdAt": "2026-03-01T03:00:00Z", + } + + // Fast feed: subscribe, then unsubscribe (hard delete). + require.NoError(t, cec.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.community.subscription", "create", "rsub1", revA, "bafrsub1", base, subRecord))) + + var subscriptions int + require.NoError(t, db.QueryRow( + `SELECT COUNT(*) FROM community_subscriptions WHERE user_did=$1 AND community_did=$2`, + revTestVoter, revTestCommunity).Scan(&subscriptions)) + require.Equal(t, 1, subscriptions, "fixture: subscription indexed") + + require.NoError(t, cec.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.community.subscription", "delete", "rsub1", revB, "", base+1_000_000, nil))) + + // Lagging feed: the subscribe replayed after the unsubscribe. + require.NoError(t, cec.HandleEvent(ctx, revCommitEvent( + revTestVoter, "social.coves.community.subscription", "create", "rsub1", revA, "bafrsub1", base+2_000_000, subRecord))) + + require.NoError(t, db.QueryRow( + `SELECT COUNT(*) FROM community_subscriptions WHERE user_did=$1 AND community_did=$2`, + revTestVoter, revTestCommunity).Scan(&subscriptions)) + assert.Equal(t, 0, subscriptions, "stale subscribe replay must not re-subscribe the user") + + var subscriberCount int + require.NoError(t, db.QueryRow( + `SELECT subscriber_count FROM communities WHERE did=$1`, revTestCommunity).Scan(&subscriberCount)) + assert.Equal(t, 0, subscriberCount, "subscriber_count must not be re-incremented by the stale replay") +} diff --git a/internal/atproto/jetstream/user_consumer.go b/internal/atproto/jetstream/user_consumer.go index 5475816..acff5a0 100644 --- a/internal/atproto/jetstream/user_consumer.go +++ b/internal/atproto/jetstream/user_consumer.go @@ -74,6 +74,7 @@ type UserEventConsumer struct { sessionHandleUpdater SessionHandleUpdater // Optional: updates OAuth sessions on handle change userBlockRepo userblocks.Repository // Optional: indexes user-to-user blocks bridgeTrust *BridgeTrust // Optional: admits new identities hosted by trusted bridge PDSes + revGate *RevGate // Optional: cross-feed ordering guard for commit events (nil = ungated) } // ConsumerOption is a functional option for configuring UserEventConsumer @@ -104,6 +105,15 @@ func WithUserBridgeTrust(bt *BridgeTrust) ConsumerOption { } } +// WithUserRevGate installs the per-record rev gate (see rev_gate.go) so profile +// and block commits are applied in repo commit order even when the same repo is +// carried by multiple Jetstream feeds. Without it, commit events are ungated. +func WithUserRevGate(gate *RevGate) ConsumerOption { + return func(c *UserEventConsumer) { + c.revGate = gate + } +} + // NewUserEventConsumer creates a new Jetstream consumer for user events func NewUserEventConsumer(userService users.UserService, identityResolver identity.Resolver, opts ...ConsumerOption) *UserEventConsumer { c := &UserEventConsumer{ @@ -116,6 +126,11 @@ func NewUserEventConsumer(userService users.UserService, identityResolver identi return c } +// RevGated reports whether this consumer applies the per-record rev gate (true when a +// gate was injected via WithUserRevGate). main.go checks this at boot to refuse +// multi-feed operation with an ungated consumer. +func (c *UserEventConsumer) RevGated() bool { return c.revGate != nil } + // HandleEvent implements EventHandler; it is invoked by the Connector for live // events and by the DeadLetterRedriver for replays, so it must stay idempotent. // The two callers may invoke the same consumer instance concurrently, so it must @@ -143,6 +158,13 @@ func (c *UserEventConsumer) HandleIdentityEventPublic(ctx context.Context, event // NOTE: This only UPDATES existing users - it does NOT create new users. // Users are created during OAuth login or signup, not from Jetstream events. // This prevents indexing millions of Bluesky users who never interact with Coves. +// +// KNOWN LIMITATION (accepted): identity events carry NO rev (they are not repo +// commits), so cross-feed ordering of handle changes is NOT rev-gated. A lagging +// feed's stale identity event can transiently revert a handle until the next +// identity event for that DID arrives. Re-resolving the DID against PLC (the +// source of truth) on every identity event would fix this and is deliberately +// not implemented yet. func (c *UserEventConsumer) handleIdentityEvent(ctx context.Context, event *JetstreamEvent) error { if event.Identity == nil { // PERMANENT: structurally invalid event — replays fail identically. @@ -295,6 +317,21 @@ func (c *UserEventConsumer) handleProfileCommit(ctx context.Context, event *Jets } } + // REV GATE: the exact cross-feed ordering guard. rev is the repo's own + // monotonic commit TID, so unlike the wall-clock guard below it orders the + // same repo's events correctly across feeds with hours of skew. Checked + // before the write and advanced after it (check→write→advance; the write is + // idempotent, so a crash in between replays safely). + uri := commitRecordURI(event.Did, event.Commit) + stale, err := c.revGate.IsStale(ctx, uri, event.Commit.Rev) + if err != nil { + return err + } + if stale { + logSkippedStaleRev(ConsumerUsers, event.Commit.Operation, uri, event.Commit.Rev) + return nil + } + // RECENCY GUARD: a redriven (DeadLetterRedriver) or rewound profile event can // arrive AFTER a newer profile write was already applied; applying it would // silently revert the newer profile. users.updated_at is bumped by every @@ -306,7 +343,8 @@ func (c *UserEventConsumer) handleProfileCommit(ctx context.Context, event *Jets // domains. That is safe for redrives (replayed minutes after the newer write) // and for human-paced profile edits; only two writes for the same user landing // within the ingest lag could be spuriously skipped, and the next profile edit - // self-heals. Making this exact needs a dedicated event-time watermark column. + // self-heals. The rev gate above is exact for events that carry a rev; this + // guard remains for rev-less events (old dead letters, synthetic tests). if evTime, ok := eventTime(event.TimeUS); ok && existingUser != nil && !existingUser.UpdatedAt.IsZero() && !existingUser.UpdatedAt.Before(evTime) { log.Printf("INFO: skipping stale profile event for %s (event time %s <= last user write %s; newer state already applied)", @@ -314,14 +352,19 @@ func (c *UserEventConsumer) handleProfileCommit(ctx context.Context, event *Jets return nil } + var opErr error switch event.Commit.Operation { case "create", "update": - return c.handleProfileUpdate(ctx, event.Did, event.Commit) + opErr = c.handleProfileUpdate(ctx, event.Did, event.Commit) case "delete": - return c.handleProfileDelete(ctx, event.Did) + opErr = c.handleProfileDelete(ctx, event.Did) default: return nil } + if opErr != nil { + return opErr + } + return c.revGate.Advance(ctx, uri, event.Commit.Rev) } // handleProfileUpdate processes profile create/update operations @@ -398,17 +441,22 @@ func (c *UserEventConsumer) handleUserBlock(ctx context.Context, userDID string, return nil } - switch commit.Operation { - case "create": - return c.createUserBlock(ctx, userDID, commit) - case "delete": - return c.deleteUserBlock(ctx, userDID, commit) - default: - // Update operations shouldn't happen on blocks, but ignore gracefully - log.Printf("Ignoring unexpected operation on user block: %s (userDID=%s, rkey=%s)", - commit.Operation, userDID, commit.RKey) - return nil - } + // REV GATE (check→write→advance, see rev_gate.go). The gate row survives + // the hard delete of the block row, so a stale cross-feed copy of the + // block's CREATE arriving after the unblock cannot re-index a phantom block. + return applyGated(ctx, c.revGate, ConsumerUsers, userDID, commit, func() error { + switch commit.Operation { + case "create": + return c.createUserBlock(ctx, userDID, commit) + case "delete": + return c.deleteUserBlock(ctx, userDID, commit) + default: + // Update operations shouldn't happen on blocks, but ignore gracefully + log.Printf("Ignoring unexpected operation on user block: %s (userDID=%s, rkey=%s)", + commit.Operation, userDID, commit.RKey) + return nil + } + }) } // createUserBlock indexes a new user-to-user block from the firehose. diff --git a/internal/atproto/jetstream/vote_consumer.go b/internal/atproto/jetstream/vote_consumer.go index 616cc23..eaa666b 100644 --- a/internal/atproto/jetstream/vote_consumer.go +++ b/internal/atproto/jetstream/vote_consumer.go @@ -33,6 +33,11 @@ func NewVoteEventConsumer( } } +// RevGated reports whether this consumer applies the per-record rev gate; always true +// for votes (gating is hardwired via c.db). main.go checks this at boot to refuse +// multi-feed operation with an ungated consumer. +func (c *VoteEventConsumer) RevGated() bool { return true } + // HandleEvent processes a Jetstream event for vote records func (c *VoteEventConsumer) HandleEvent(ctx context.Context, event *JetstreamEvent) error { // We only care about commit events for vote records @@ -98,8 +103,8 @@ func (c *VoteEventConsumer) createVote(ctx context.Context, repoDID string, comm IndexedAt: time.Now(), } - // Atomically: Index vote + Update post counts - wasNew, err := c.indexVoteAndUpdateCounts(ctx, vote) + // Atomically: Rev-gate + Index vote + Update post counts + wasNew, err := c.indexVoteAndUpdateCounts(ctx, vote, commit.Rev) if err != nil { return fmt.Errorf("failed to index vote and update counts: %w", err) } @@ -110,34 +115,178 @@ func (c *VoteEventConsumer) createVote(ctx context.Context, repoDID string, comm return nil } -// deleteVote soft-deletes a vote and updates post counts +// deleteVote soft-deletes a vote and updates post counts. +// +// The rev-gate claim runs FIRST, inside the same transaction as the load and +// the soft delete (mirrors deletePost). Claiming before reading closes the +// not-found tombstone race: a concurrent create of the same vote (another +// feed's copy, or a DeadLetterRedriver replay) serializes on the gate row +// lock, so it either commits before our read (we see the row and delete it) +// or blocks until our tombstone commits (its equal-or-older rev then loses +// the gate). The gate row is advanced — and committed — even when the vote +// was never indexed, so the create's late copy is rejected too. func (c *VoteEventConsumer) deleteVote(ctx context.Context, repoDID string, commit *CommitEvent) error { // Build AT-URI for the vote being deleted uri := fmt.Sprintf("at://%s/social.coves.feed.vote/%s", repoDID, commit.RKey) - // Get existing vote to know its direction (for decrementing the right counter) - existingVote, err := c.voteRepo.GetByURI(ctx, uri) + tx, err := c.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer func() { + if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { + log.Printf("Failed to rollback transaction: %v", rollbackErr) + } + }() + + // 0. REV GATE (see indexVoteAndUpdateCounts): skip duplicate replays and + // stale cross-feed copies; the claimed row doubles as the tombstone that + // rejects the create's later copies. + won, err := tryAdvanceRecordRev(ctx, tx, uri, commit.Rev) if err != nil { - if err == votes.ErrVoteNotFound { - // Idempotent: Vote already deleted or never existed - log.Printf("Vote already deleted or not found: %s", uri) - return nil + return err + } + if !won { + logSkippedStaleRev(ConsumerVotes, "delete", uri, commit.Rev) + return nil + } + + // 1. Load the vote INSIDE the gate transaction: direction and subject + // drive the count decrement below, and reading under the gate claim means + // no concurrent create for this URI can commit between this read and our + // tombstone commit. + var direction, subjectURI string + var deletedAt *time.Time + err = tx.QueryRowContext(ctx, + `SELECT direction, subject_uri, deleted_at FROM votes WHERE uri = $1`, uri, + ).Scan(&direction, &subjectURI, &deletedAt) + if err == sql.ErrNoRows { + // Idempotent: vote never indexed. Commit the gate advance anyway — it + // is the tombstone that rejects a stale cross-feed copy of the CREATE + // arriving later for a record that no longer exists on the PDS. + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) } + log.Printf("Vote already deleted or not found: %s", uri) + return nil + } + if err != nil { return fmt.Errorf("failed to get existing vote: %w", err) } + if deletedAt != nil { + // Idempotent: already soft-deleted. Still commit the tombstone advance. + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + log.Printf("Vote already deleted: %s (idempotent)", uri) + return nil + } - // Atomically: Soft-delete vote + Update post counts - if err := c.deleteVoteAndUpdateCounts(ctx, existingVote); err != nil { - return fmt.Errorf("failed to delete vote and update counts: %w", err) + // 2. Soft-delete the vote + deleteQuery := ` + UPDATE votes + SET deleted_at = NOW() + WHERE uri = $1 AND deleted_at IS NULL + ` + result, err := tx.ExecContext(ctx, deleteQuery, uri) + if err != nil { + return fmt.Errorf("failed to delete vote: %w", err) + } + rowsAffected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to check delete result: %w", err) + } + // Defensive: createVote's stale-vote cleanup (different URI, so a different + // gate row) can soft-delete this row concurrently. Zero rows then means the + // vote is already deleted and its count already decremented — commit the + // tombstone and skip our decrement. + if rowsAffected == 0 { + log.Printf("Vote already deleted: %s (idempotent)", uri) + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil + } + + // 3. Decrement vote counts on the subject (post or comment) + // Parse collection from subject URI to determine target table + collection := utils.ExtractCollectionFromURI(subjectURI) + + var updateQuery string + switch collection { + case "social.coves.community.post": + // Vote on post - update posts table + if direction == "up" { + updateQuery = ` + UPDATE posts + SET upvote_count = GREATEST(0, upvote_count - 1), + score = GREATEST(0, upvote_count - 1) - downvote_count + bridged_upvote_count - bridged_downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } else { // "down" + updateQuery = ` + UPDATE posts + SET downvote_count = GREATEST(0, downvote_count - 1), + score = upvote_count - GREATEST(0, downvote_count - 1) + bridged_upvote_count - bridged_downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } + + case "social.coves.community.comment": + // Vote on comment - update comments table + if direction == "up" { + updateQuery = ` + UPDATE comments + SET upvote_count = GREATEST(0, upvote_count - 1), + score = GREATEST(0, upvote_count - 1) - downvote_count + bridged_upvote_count - bridged_downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } else { // "down" + updateQuery = ` + UPDATE comments + SET downvote_count = GREATEST(0, downvote_count - 1), + score = upvote_count - GREATEST(0, downvote_count - 1) + bridged_upvote_count - bridged_downvote_count + WHERE uri = $1 AND deleted_at IS NULL + ` + } + + default: + // Unknown or unsupported collection + // Vote is still deleted, we just don't update denormalized counts + log.Printf("Vote subject has unsupported collection: %s (vote deleted, counts not updated)", collection) + if commitErr := tx.Commit(); commitErr != nil { + return fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return nil + } + + result, err = tx.ExecContext(ctx, updateQuery, subjectURI) + if err != nil { + return fmt.Errorf("failed to update vote counts: %w", err) + } + + rowsAffected, err = result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to check update result: %w", err) + } + + // If subject doesn't exist or is deleted, that's OK (vote still deleted) + if rowsAffected == 0 { + log.Printf("Warning: Vote subject not found or deleted: %s (vote deleted anyway)", subjectURI) + } + + // Commit transaction + if err := tx.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) } - log.Printf("✓ Deleted vote: %s (%s on %s)", uri, existingVote.Direction, existingVote.SubjectURI) + log.Printf("✓ Deleted vote: %s (%s on %s)", uri, direction, subjectURI) return nil } // indexVoteAndUpdateCounts atomically indexes a vote and updates post vote counts // Returns (true, nil) if vote was newly inserted, (false, nil) if already existed (idempotent) -func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote *votes.Vote) (bool, error) { +func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote *votes.Vote, rev string) (bool, error) { tx, err := c.db.BeginTx(ctx, nil) if err != nil { return false, fmt.Errorf("failed to begin transaction: %w", err) @@ -148,6 +297,21 @@ func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote * } }() + // 0. REV GATE: apply this create only if its rev is strictly newer than the + // last applied event for this record. Rejects duplicate replays (equal rev) + // and — critically — a stale cross-feed copy of a CREATE arriving after this + // vote's DELETE was already applied, which would otherwise restore a phantom + // vote and re-increment counts. Runs first, inside the transaction, so the + // gate and the writes commit or roll back together. + won, err := tryAdvanceRecordRev(ctx, tx, vote.URI, rev) + if err != nil { + return false, err + } + if !won { + logSkippedStaleRev(ConsumerVotes, "create", vote.URI, rev) + return false, nil + } + // 1. Check for existing active vote with different URI (stale record) // This handles cases where: // - User voted on another client and we missed the delete event @@ -229,7 +393,20 @@ func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote * // If no rows returned, vote already exists (idempotent - OK for Jetstream replays) if err == sql.ErrNoRows { - // Silently handle idempotent case - no log needed for replayed events + // KNOWN LIMITATION (accepted): a genuine RE-CREATE of the same rkey while + // the row is still ACTIVE also lands here and is treated as an idempotent + // duplicate. Reaching that state requires the exact sequence: create A + // applied → delete dead-lettered (failed, never applied) → re-create B + // (same rkey, strictly newer rev, possibly flipped direction) arrives + // while the row is still active. B's direction is never applied, the gate + // advances to B's rev, and the redriven delete A is then gate-rejected — + // the row survives with A's direction. This needs a dead-lettered delete + // AND an rkey reuse inside the redrive window; rare enough to document + // rather than plumb a direction-flipping upsert (with paired count + // adjustments) through the create path. Comments handle their analogous + // case in place (see indexCommentAndUpdateCounts). + // + // Silently handle the common idempotent case - no log needed for replays. if commitErr := tx.Commit(); commitErr != nil { return false, fmt.Errorf("failed to commit transaction: %w", commitErr) } @@ -315,119 +492,6 @@ func (c *VoteEventConsumer) indexVoteAndUpdateCounts(ctx context.Context, vote * return true, nil // Vote was newly indexed } -// deleteVoteAndUpdateCounts atomically soft-deletes a vote and updates post vote counts -func (c *VoteEventConsumer) deleteVoteAndUpdateCounts(ctx context.Context, vote *votes.Vote) error { - tx, err := c.db.BeginTx(ctx, nil) - if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) - } - defer func() { - if rollbackErr := tx.Rollback(); rollbackErr != nil && rollbackErr != sql.ErrTxDone { - log.Printf("Failed to rollback transaction: %v", rollbackErr) - } - }() - - // 1. Soft-delete the vote (idempotent) - deleteQuery := ` - UPDATE votes - SET deleted_at = NOW() - WHERE uri = $1 AND deleted_at IS NULL - ` - - result, err := tx.ExecContext(ctx, deleteQuery, vote.URI) - if err != nil { - return fmt.Errorf("failed to delete vote: %w", err) - } - - rowsAffected, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("failed to check delete result: %w", err) - } - - // Idempotent: If no rows affected, vote already deleted - if rowsAffected == 0 { - log.Printf("Vote already deleted: %s (idempotent)", vote.URI) - if commitErr := tx.Commit(); commitErr != nil { - return fmt.Errorf("failed to commit transaction: %w", commitErr) - } - return nil - } - - // 2. Decrement vote counts on the subject (post or comment) - // Parse collection from subject URI to determine target table - collection := utils.ExtractCollectionFromURI(vote.SubjectURI) - - var updateQuery string - switch collection { - case "social.coves.community.post": - // Vote on post - update posts table - if vote.Direction == "up" { - updateQuery = ` - UPDATE posts - SET upvote_count = GREATEST(0, upvote_count - 1), - score = GREATEST(0, upvote_count - 1) - downvote_count + bridged_upvote_count - bridged_downvote_count - WHERE uri = $1 AND deleted_at IS NULL - ` - } else { // "down" - updateQuery = ` - UPDATE posts - SET downvote_count = GREATEST(0, downvote_count - 1), - score = upvote_count - GREATEST(0, downvote_count - 1) + bridged_upvote_count - bridged_downvote_count - WHERE uri = $1 AND deleted_at IS NULL - ` - } - - case "social.coves.community.comment": - // Vote on comment - update comments table - if vote.Direction == "up" { - updateQuery = ` - UPDATE comments - SET upvote_count = GREATEST(0, upvote_count - 1), - score = GREATEST(0, upvote_count - 1) - downvote_count + bridged_upvote_count - bridged_downvote_count - WHERE uri = $1 AND deleted_at IS NULL - ` - } else { // "down" - updateQuery = ` - UPDATE comments - SET downvote_count = GREATEST(0, downvote_count - 1), - score = upvote_count - GREATEST(0, downvote_count - 1) + bridged_upvote_count - bridged_downvote_count - WHERE uri = $1 AND deleted_at IS NULL - ` - } - - default: - // Unknown or unsupported collection - // Vote is still deleted, we just don't update denormalized counts - log.Printf("Vote subject has unsupported collection: %s (vote deleted, counts not updated)", collection) - if commitErr := tx.Commit(); commitErr != nil { - return fmt.Errorf("failed to commit transaction: %w", commitErr) - } - return nil - } - - result, err = tx.ExecContext(ctx, updateQuery, vote.SubjectURI) - if err != nil { - return fmt.Errorf("failed to update vote counts: %w", err) - } - - rowsAffected, err = result.RowsAffected() - if err != nil { - return fmt.Errorf("failed to check update result: %w", err) - } - - // If subject doesn't exist or is deleted, that's OK (vote still deleted) - if rowsAffected == 0 { - log.Printf("Warning: Vote subject not found or deleted: %s (vote deleted anyway)", vote.SubjectURI) - } - - // Commit transaction - if err := tx.Commit(); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - - return nil -} - // validateVoteEvent performs security validation on vote events func (c *VoteEventConsumer) validateVoteEvent(ctx context.Context, repoDID string, vote *VoteRecordFromJetstream) error { // SECURITY: Votes MUST come from user repositories (repo owner = voter DID) diff --git a/internal/db/migrations/033_create_jetstream_record_revs.sql b/internal/db/migrations/033_create_jetstream_record_revs.sql new file mode 100644 index 0000000..a90ba88 --- /dev/null +++ b/internal/db/migrations/033_create_jetstream_record_revs.sql @@ -0,0 +1,40 @@ +-- +goose Up +-- Per-record rev gate for Jetstream consumers. +-- +-- WHY THIS EXISTS: the AppView is about to consume MULTIPLE Jetstream feeds +-- carrying the SAME repos (the public bsky.network feed plus our self-hosted +-- relay feed). Each feed is internally ordered, but the copies are skewed by +-- hours, so a later event for a repo can be processed before an earlier copy +-- of the same repo's history arriving on the other feed. Without a gate that +-- stale copy resurrects deleted records (create replayed after delete) or +-- regresses edited content (pre-edit update replayed after the edit). +-- +-- THE MECHANISM: every commit event carries `rev`, the repo's monotonic TID — +-- a fixed-length base32-sortable string, so plain lexicographic comparison +-- IS commit order within one repo. Consumers record the rev of the last +-- APPLIED event per record URI here and apply an incoming create/update/ +-- delete only when its rev is strictly greater than the stored one. Equal +-- rev means the same event replayed (reconnect rewind, redrive, duplicate +-- feed) and is a no-op; smaller means a stale cross-feed copy and is +-- skipped. One rule restores per-repo commit ordering regardless of how +-- feeds interleave — no heuristics, no CID comparisons. +-- +-- WHY A SEPARATE TABLE instead of a rev column on each indexed table: the +-- row must SURVIVE the record it describes. Hard-deleted record types +-- (subscriptions, blocks, communities, aggregator rows) leave no row to +-- carry the delete's rev, so a per-table column cannot reject the stale +-- create that follows — this table doubles as the tombstone. Rows are tiny +-- (uri + 13-char rev) and bounded by the number of records ever indexed. +-- +-- Events with no rev (synthetic test events, pre-existing dead letters) +-- bypass the gate entirely, preserving previous behavior. +CREATE TABLE jetstream_record_revs ( + record_uri TEXT PRIMARY KEY, + -- COLLATE "C" pins the gate's rev comparison to bytewise order (for TIDs, + -- bytewise IS commit order) regardless of the database's default collation. + rev TEXT COLLATE "C" NOT NULL, + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +-- +goose Down +DROP TABLE IF EXISTS jetstream_record_revs; diff --git a/internal/db/postgres/community_repo_subscriptions.go b/internal/db/postgres/community_repo_subscriptions.go index 61b8eb5..fb03697 100644 --- a/internal/db/postgres/community_repo_subscriptions.go +++ b/internal/db/postgres/community_repo_subscriptions.go @@ -50,13 +50,24 @@ func (r *postgresCommunityRepo) SubscribeWithCount(ctx context.Context, subscrip } }() - // Insert subscription with ON CONFLICT DO NOTHING for idempotency + // Insert subscription; idempotent for Jetstream replays. On conflict (the + // user already has an active subscription row for this community) the + // stored record_uri/record_cid are updated to the INCOMING record + // (last-write-wins). This matters when the user re-subscribed under a NEW + // rkey while the old rkey's unsubscribe was dead-lettered: pinning the row + // to the newest record means the redriven delete of the OLD record URI no + // longer matches the row (the consumer looks it up by record_uri) and is + // skipped, instead of tearing down a valid newer subscription. + // (xmax = 0) distinguishes a fresh insert (count must be incremented) from + // a conflict-update (row already existed; count unchanged). query := ` INSERT INTO community_subscriptions (user_did, community_did, subscribed_at, record_uri, record_cid, content_visibility) VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (user_did, community_did) DO NOTHING - RETURNING id, subscribed_at, content_visibility` + ON CONFLICT (user_did, community_did) DO UPDATE + SET record_uri = EXCLUDED.record_uri, record_cid = EXCLUDED.record_cid + RETURNING id, subscribed_at, content_visibility, (xmax = 0) AS inserted` + var inserted bool err = tx.QueryRowContext(ctx, query, subscription.UserDID, subscription.CommunityDID, @@ -64,22 +75,7 @@ func (r *postgresCommunityRepo) SubscribeWithCount(ctx context.Context, subscrip nullString(subscription.RecordURI), nullString(subscription.RecordCID), subscription.ContentVisibility, - ).Scan(&subscription.ID, &subscription.SubscribedAt, &subscription.ContentVisibility) - - // If no rows returned, subscription already existed (idempotent behavior) - if err == sql.ErrNoRows { - // Get existing subscription - query = `SELECT id, subscribed_at, content_visibility FROM community_subscriptions WHERE user_did = $1 AND community_did = $2` - err = tx.QueryRowContext(ctx, query, subscription.UserDID, subscription.CommunityDID).Scan(&subscription.ID, &subscription.SubscribedAt, &subscription.ContentVisibility) - if err != nil { - return nil, fmt.Errorf("failed to get existing subscription: %w", err) - } - // Don't increment count - subscription already existed - if commitErr := tx.Commit(); commitErr != nil { - return nil, fmt.Errorf("failed to commit transaction: %w", commitErr) - } - return subscription, nil - } + ).Scan(&subscription.ID, &subscription.SubscribedAt, &subscription.ContentVisibility, &inserted) if err != nil { if strings.Contains(err.Error(), "foreign key") { @@ -88,6 +84,15 @@ func (r *postgresCommunityRepo) SubscribeWithCount(ctx context.Context, subscrip return nil, fmt.Errorf("failed to create subscription: %w", err) } + // Subscription already existed (idempotent replay or re-subscribe under a + // new rkey): record pointer refreshed above, count stays as-is. + if !inserted { + if commitErr := tx.Commit(); commitErr != nil { + return nil, fmt.Errorf("failed to commit transaction: %w", commitErr) + } + return subscription, nil + } + // Increment subscriber count only if insert succeeded incrementQuery := ` UPDATE communities diff --git a/scripts/dev-run.sh b/scripts/dev-run.sh index 8766646..65a440b 100755 --- a/scripts/dev-run.sh +++ b/scripts/dev-run.sh @@ -9,7 +9,7 @@ set +a echo "🚀 Starting Coves server in DEV mode..." echo " IS_DEV_ENV: $IS_DEV_ENV" echo " PLC_DIRECTORY_URL: $PLC_DIRECTORY_URL" -echo " JETSTREAM_URL: $JETSTREAM_URL" +echo " JETSTREAM_FEEDS: $JETSTREAM_FEEDS" echo " APPVIEW_PUBLIC_URL: $APPVIEW_PUBLIC_URL" echo " PDS_URL: $PDS_URL" echo " Build tags: dev" diff --git a/tests/integration/aggregator_e2e_test.go b/tests/integration/aggregator_e2e_test.go index 588aa98..3506d32 100644 --- a/tests/integration/aggregator_e2e_test.go +++ b/tests/integration/aggregator_e2e_test.go @@ -847,7 +847,7 @@ func TestAggregator_E2E_WithJetstream(t *testing.T) { // TestAggregator_E2E_LivePDS tests the COMPLETE end-to-end flow with a live PDS // This would require: // - Live PDS running at PDS_URL -// - Live Jetstream running at JETSTREAM_URL +// - Live Jetstream running at the local dev Jetstream (ws://localhost:6008; configured via JETSTREAM_FEEDS) // - Ability to provision aggregator accounts on PDS // - Real WebSocket connection to Jetstream firehose // diff --git a/tests/integration/comment_vote_test.go b/tests/integration/comment_vote_test.go index 04ec265..2f01fce 100644 --- a/tests/integration/comment_vote_test.go +++ b/tests/integration/comment_vote_test.go @@ -289,12 +289,14 @@ func TestCommentVote_CreateAndUpdate(t *testing.T) { t.Fatalf("Expected upvote_count = 1 before delete, got %d", commentAfterVote.UpvoteCount) } - // Delete vote + // Delete vote. The rev must be LATER than the create's: revs are the + // repo's monotonic commit IDs, and the rev gate treats an equal rev as + // the same event replayed (a no-op by design). deleteVoteEvent := &jetstream.JetstreamEvent{ Did: testUser.DID, Kind: "commit", Commit: &jetstream.CommitEvent{ - Rev: "test-rev", + Rev: "test-rev-2", Operation: "delete", Collection: "social.coves.feed.vote", RKey: voteRKey, diff --git a/tests/integration/post_e2e_test.go b/tests/integration/post_e2e_test.go index 8261937..3e7a5a4 100644 --- a/tests/integration/post_e2e_test.go +++ b/tests/integration/post_e2e_test.go @@ -320,7 +320,7 @@ func TestPostCreation_E2E_WithJetstream(t *testing.T) { // // This is a TRUE E2E test that requires: // - Live PDS running at PDS_URL (default: http://localhost:3001) -// - Live Jetstream running at JETSTREAM_URL (default: ws://localhost:6008/subscribe) +// - Live Jetstream running at the local dev Jetstream (JETSTREAM_FEEDS default: self=ws://localhost:6008) // - Test database running func TestPostCreation_E2E_LivePDS(t *testing.T) { if testing.Short() { -- 2.51.2