Something went wrong. Try again.
A community based topic aggregation platform built on atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313package main
import ( "context" "errors" "fmt" "log/slog" "sync"
"Coves/internal/atproto/jetstream" postgresRepo "Coves/internal/db/postgres")
// consumerSet is the collection of Jetstream consumers started for this// process, kept so /health/consumers can report on them and so shutdown can// drain them.type consumerSet struct { connectors []*jetstream.Connector}
// feedConsumer pairs a consumer's stable name with its handler. The name keys// the consumer's persisted cursor and dead letter rows, so it must not change// between releases.type feedConsumer struct { name string handler jetstream.EventHandler}
// startConsumers wires every Jetstream consumer, validates the feed topology,// and starts one connector per (feed, consumer) pair plus the dead letter// redriver.//// All consumers share ctx so a single cancellation drains them: read loops// unblock, an interrupted in-flight event is abandoned without advancing its// cursor (it replays idempotently on the next boot), and the final cursor is// flushed.func startConsumers(ctx context.Context, wg *sync.WaitGroup, app *application) (*consumerSet, error) { feeds, err := jetstream.ParseFeeds(app.cfg.Jetstream.FeedsSpec) if err != nil { return nil, fmt.Errorf("invalid JETSTREAM_FEEDS: %w", err) } warnIfNoPrimaryFeed(feeds)
consumers, err := app.registerFeedConsumers() if err != nil { return nil, err }
// 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(feeds) > 1 { for _, consumer := range consumers { gated, ok := consumer.handler.(interface{ RevGated() bool }) if !ok || !gated.RevGated() { return nil, fmt.Errorf("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)", consumer.name, len(feeds)) } } }
set := &consumerSet{} handlers := make(map[string]jetstream.EventHandler, len(consumers)*len(feeds))
// Consumer names on the primary ("bsky") feed stay bare so live cursors // carry over from the single-feed era; other feeds get "<consumer>@<feed>" // names, which start cursor-less and live-tail. Rev-gating makes the // cross-feed overlap safe — expect "rev-gate: skipped stale" lines for the // lagging feed's copies. That is the system working, not an error. for _, feed := range feeds { for _, consumer := range consumers { collections, err := jetstream.WantedCollections(consumer.name) if err != nil { return nil, fmt.Errorf("resolving wantedCollections for consumer %s: %w", consumer.name, err) } wsURL, err := jetstream.SubscribeURL(feed.BaseURL, collections) if err != nil { return nil, fmt.Errorf("building Jetstream URL for consumer %s on feed %s: %w", consumer.name, feed.Key, err) }
name := jetstream.FeedConsumerName(consumer.name, feed.Key) handlers[name] = consumer.handler set.start(ctx, wg, app, name, wsURL, consumer.handler) } slog.Info("started Jetstream consumers on feed", "consumers", len(consumers), "feed", feed.Key, "url", feed.BaseURL) }
// The redriver replays events that failed every in-line retry against the // same handlers, so a transient failure (a Postgres blip, say) self-heals // instead of silently losing the event. redriver := jetstream.NewDeadLetterRedriver(jetstream.DeadLetterRedriveStore{ Source: app.jetstreamState, Mutator: app.jetstreamState, Pruner: app.jetstreamState, }, handlers, jetstream.WithRedriveInterval(app.cfg.Jetstream.RedriveInterval), // This makes REDRIVE_INTERVAL > IDENTITY_NEGATIVE_CACHE_TTL hold per row, not only per tick. jetstream.WithRedriveMinimumAge(app.cfg.Identity.NegativeCacheTTL)) wg.Add(1) go func() { defer wg.Done() redriver.Run(ctx) }() slog.Info("started Jetstream dead letter redriver")
return set, nil}
// start launches one connector with cursor persistence, retry plus dead// letter, and graceful shutdown.func (s *consumerSet) start(ctx context.Context, wg *sync.WaitGroup, app *application, name, wsURL string, handler jetstream.EventHandler) { connector := jetstream.NewConnector(name, wsURL, handler, jetstream.WithCursorStore(app.jetstreamState), jetstream.WithDeadLetterWriter(app.jetstreamState), ) s.connectors = append(s.connectors, connector)
wg.Add(1) go func() { defer wg.Done() if err := connector.Start(ctx); err != nil && !errors.Is(err, context.Canceled) { slog.Error("Jetstream consumer stopped", "consumer", name, "error", err) } }()}
// registerFeedConsumers builds every consumer, in a fixed order. Each runs// once per configured feed, with its collection filters appended by// jetstream.WantedCollections.func (a *application) registerFeedConsumers() ([]feedConsumer, error) { var consumers []feedConsumer
// One lock instance serializes identity refreshes of a DID for the user // and community consumers alike. identityRefreshLock := jetstream.NewPostgresIdentityRefreshLock(a.identityRefreshLockDB, jetstream.IdentityRefreshLockWaitTimeout)
// Users: actor profiles and actor blocks. // // A trusted bridge hosts many virtual repos, and its profile records may // be the first time Coves sees those identities. Relay scheduling can // deliver profiles and posts in either order, so the user, post, and // comment consumers share one provenance gate. userOpts := []jetstream.ConsumerOption{ jetstream.WithUserBridgeTrust(a.bridgeTrust), jetstream.WithUserRevGate(a.revGate), jetstream.WithUserBlockRepo(a.userBlockRepo), jetstream.WithUserUncachedIdentityResolver(a.uncachedIdentityResolver), jetstream.WithUserIdentityRefreshFailureTTL(a.cfg.Identity.NegativeCacheTTL), jetstream.WithUserIdentityRefreshLock(identityRefreshLock), } // Statically typed rather than asserted at runtime: a type assertion here // would silently disable handle sync if the store's method set ever // drifted, leaving OAuth sessions pinned to stale handles with no error. // The compiler now catches that instead. if sessionUpdater := a.oauthStore.UnwrapPostgresStore(); sessionUpdater != nil { userOpts = append(userOpts, jetstream.WithSessionHandleUpdater(sessionUpdater)) slog.Info("OAuth session handle sync enabled for identity changes") } consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerUsers, handler: jetstream.NewUserEventConsumer(a.userService, a.identityResolver, userOpts...), })
// Communities: profiles (in the community's own repo), plus subscriptions // and community blocks (in the subscribing user's repo). The identity // resolver supplies PLC handle resolution, which is the source of truth. if a.cfg.Instance.SkipDIDWebVerification { slog.Warn("did:web domain verification is DISABLED; this must never be set in production") } // A community's stored handle comes ONLY from resolving its DID // (the create-path-handle-spoofing fix). Without a resolver the consumer falls // back to the record's own handle — a T0-only mode that would let any repo // name itself c-<anything>.<our domain>. buildIdentity always constructs one, // so this is the assertion that keeps a future refactor from starting the // consumer in that mode. if a.identityResolver == nil { return nil, fmt.Errorf("community consumer requires an identity resolver: a consumer without one " + "indexes communities under whatever handle their own record claims") } // The SSRF hatch is open only in dev, where a developer's own instance is // what the .well-known DID document is fetched from. In production that // fetch dials whatever domain a community record published by anyone on the // federated network names, from inside this AppView's own network. // BridgeTrust here gates the self-asserted `origin` on profile records: a // repo on a trusted bridge PDS may claim a foreign origin (lemmy.world); // anyone else may claim only their own handle's domain. communityOpts := append( []jetstream.CommunityConsumerOption{ jetstream.WithCommunityRevGate(a.revGate), jetstream.WithCommunityBridgeTrust(a.bridgeTrust), jetstream.WithCommunityUncachedIdentityResolver(a.uncachedIdentityResolver), jetstream.WithCommunityIdentityRefreshFailureTTL(a.cfg.Identity.NegativeCacheTTL), jetstream.WithCommunityIdentityRefreshLock(identityRefreshLock), }, jetstream.PrivateHostOptions(a.allowPrivateHosts())...) consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerCommunities, handler: jetstream.NewCommunityEventConsumer( a.communityRepo, a.cfg.Instance.DID, a.cfg.Instance.SkipDIDWebVerification, a.identityResolver, communityOpts...), })
// Posts, in both shapes, plus the community records that decide about // them: the deprecated community-repo post, the author-repo postv2, and // the acceptance/removal pair. One consumer, because they write the same // admission row and an acceptance is meaningless without the post it pins. // The warning stays here, because it is about this process rather than about // the option — a helper that logged would fire once per test that builds a // hatched fetcher, and this line has to mean "this server is running // unguarded". if a.allowPrivateHosts() { slog.Warn("direct post fetch has SSRF protection DISABLED (IS_DEV_ENV); this must never be set in production") } consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerPosts, handler: a.buildPostConsumer(), })
// Aggregators: service declarations and authorization records, following // Bluesky's feed generator and labeler pattern. consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerAggregators, handler: jetstream.NewAggregatorEventConsumer(a.aggregatorRepo, jetstream.WithAggregatorRevGate(a.revGate)), })
// Votes from user repositories, with atomic post/comment count updates. // // The same erasure marker the post consumer reads: a vote whose subject was // swept by an account erasure has no row to wait for. Without the lookup, the // ordering gate refuses it as merely early — it dead-letters at once, the // redriver replays it until the attempt budget is spent, and the row then sits // retired in the queue alongside events that represent a real backlog. consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerVotes, handler: a.buildVoteConsumer(), })
// Comments from user repositories, with atomic parent count updates. consumers = append(consumers, feedConsumer{ name: jetstream.ConsumerComments, handler: a.buildCommentConsumer(), })
return consumers, nil}
// warnIfNoPrimaryFeed flags a topology with no primary-key feed. This is// expected in local dev (a self-only feed), but in production it usually means// cursor continuity from the single-feed era is being forfeited.func warnIfNoPrimaryFeed(feeds []jetstream.Feed) { for _, feed := range feeds { if feed.Key == jetstream.PrimaryFeedKey { return } } slog.Warn("no JETSTREAM_FEEDS entry uses the primary feed key: every consumer name will be "+ "suffixed \"@<feedKey>\", so cursors persisted under the bare legacy names will NOT be used", "primary_feed_key", jetstream.PrimaryFeedKey)}
// buildVoteConsumer builds the vote consumer registered on the feed.func (a *application) buildVoteConsumer() *jetstream.VoteEventConsumer { return jetstream.NewVoteEventConsumer(a.voteRepo, a.userService, a.db, jetstream.WithVoteDeletedAccounts(postgresRepo.NewDeletedAccountRepository(a.db)), jetstream.WithVoteBridgeTrust(a.bridgeTrust), jetstream.WithVoteNotifications(a.notificationRepo))}
// buildCommentConsumer builds the comment consumer registered on the feed.func (a *application) buildCommentConsumer() *jetstream.CommentEventConsumer { return jetstream.NewCommentEventConsumer(a.commentRepo, a.db, jetstream.WithCommentBridgeTrust(a.bridgeTrust), jetstream.WithCommentMediaReconciler(a.mediaReconciler), jetstream.WithCommentNotifications(a.notificationRepo), jetstream.WithCommentBlobReferences(a.blobReferenceRepo))}
// buildPostConsumer builds the post consumer registered on the feed.func (a *application) buildPostConsumer() *jetstream.PostEventConsumer { // The direct fetcher is what makes acceptance-before-post converge without // full relay coverage (docs/POSTS.md §5.4). It dials a PDS named by a DID document // anyone can publish, so its SSRF guard stays on in production and the hatch // opens only under IS_DEV_ENV — where the hermetic stack's PDS is a private // address the guard would otherwise refuse. // // The decision goes through jetstream.PrivatePostFetcherOptions rather than // an inline conditional: `.env.ci:140` sets IS_DEV_ENV=true, so `make ci` // takes the permissive branch. The production branch is testable in T0. postFetcher := jetstream.NewDirectPostFetcher(a.identityResolver, jetstream.PrivatePostFetcherOptions(a.allowPrivateHosts())...) return jetstream.NewPostEventConsumer(a.postRepo, a.communityRepo, a.userService, a.db, jetstream.WithPostBridgeTrust(a.bridgeTrust), jetstream.WithPostIdentityResolver(a.identityResolver), jetstream.WithAdmissions(a.admissionRepo), jetstream.WithDeletedAccounts(postgresRepo.NewDeletedAccountRepository(a.db)), jetstream.WithPostRecordFetcher(postFetcher), jetstream.WithPostMediaReconciler(a.mediaReconciler), // The host-side half of an author's own deletion (§5.3): when the // author tombstones a post this instance's community accepted, the // acceptance in that community's repo is withdrawn. It refuses // itself for every community this AppView does not host, which on // most instances is all of them. jetstream.WithAcceptanceCleanup(a.communityWriter), jetstream.WithPostNotifications(a.notificationRepo), jetstream.WithPostBlobReferences(a.blobReferenceRepo))}