Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
35 kB · 926 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927package atproto
import ( "bytes" "context" "crypto/tls" "fmt" "net/http" "net/url" "runtime" "strconv" "strings" "sync" "sync/atomic" "time"
indigoatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/events/schedulers/parallel" "github.com/bluesky-social/indigo/repo" "github.com/bluesky-social/indigo/repomgr" "github.com/ipfs/go-cid" glex "github.com/streamplace/glex/runtime" "github.com/streamplace/oatproxy/pkg/oatproxy" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqhttp" "stream.place/streamplace/pkg/aqtime" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/model" notificationpkg "stream.place/streamplace/pkg/notifications" "stream.place/streamplace/pkg/spmetrics" "stream.place/streamplace/pkg/statedb"
"slices"
"github.com/gorilla/websocket")
// dedupWindow bounds how long a key is remembered. Relays run within seconds of// each other, so a few minutes covers all realistic skew.const dedupWindow = 5 * time.Minute
type ATProtoSynchronizer struct { CLI *config.CLI Model model.Model StatefulDB *statedb.StatefulDB Noter notificationpkg.Notifier Bus *bus.Bus // The identity directories, built on first use behind dirMu; read them // through [ATProtoSynchronizer.directory] rather than directly. Set them // before the synchronizer is used and they are taken as given. PLCDirectory identity.Directory CachedPLCDirectory identity.Directory dirMu sync.Mutex OATProxy *oatproxy.OATProxy
// firehose liveness, written from every relay consumer concurrently // (unix nanos). lastSeen atomic.Int64 lastEvent atomic.Int64 // events seen across all relays, pre-dedup (for the periodic in-sync ping). seenEvents atomic.Int64
// cross-relay dedup, shared by every relay consumer. Initialized at the // top of StartFirehose. commitDedup *firehoseDeduper identityDedup *firehoseDeduper
// sweeping is held for the length of a sweep, so the periodic ticker // cannot start a second one on top of the first. sweeping atomic.Bool}
func (atsync *ATProtoSynchronizer) markSeen() { atsync.lastSeen.Store(time.Now().UnixNano()) atsync.seenEvents.Add(1)}
func (atsync *ATProtoSynchronizer) markEvent(t time.Time) { atsync.lastEvent.Store(t.UnixNano())}
func sinceNanos(ns int64) time.Duration { if ns == 0 { return 0 } return time.Since(time.Unix(0, ns))}
// relayHosts returns the ordered, de-duplicated list of relay websocket URLs to// consume: the comma-separated --relay-host list, plus our own PDS firehose so// records published to our built-in PDS are always indexed immediately (rather// than waiting for an external relay to crawl them back to us). We only add// ourselves when we actually have an HTTP listener to dial — i.e. never in// tests that run without the server.func (atsync *ATProtoSynchronizer) relayHosts() []string { seen := map[string]struct{}{} var hosts []string add := func(h string) { h = strings.TrimSpace(h) if h == "" { return } if _, ok := seen[h]; ok { return } seen[h] = struct{}{} hosts = append(hosts, h) } for _, h := range strings.Split(atsync.CLI.RelayHost, ",") { add(h) } if self := atsync.selfRelayURL(); self != "" { add(self) } return hosts}
// selfRelayURL is the websocket URL of our own firehose listener, or ""// when there's no HTTP listener to dial (e.g. tests run without the// server). Used both to add ourselves to the relay set and to recognize// that connection in connectRelay so we can present the right Host header.func (atsync *ATProtoSynchronizer) selfRelayURL() string { if atsync.CLI.HTTPAddr == "" { return "" } return httpToWSURL(atsync.CLI.OwnPublicURL())}
func httpToWSURL(u string) string { if strings.HasPrefix(u, "https://") { return "wss://" + strings.TrimPrefix(u, "https://") } if strings.HasPrefix(u, "http://") { return "ws://" + strings.TrimPrefix(u, "http://") } return u}
// StartFirehose subscribes to every configured relay (plus our own PDS, if// enabled) concurrently, de-duplicating the overlapping event streams so each// commit is indexed exactly once. It returns only when ctx is cancelled; a// single relay flapping is logged and retried forever rather than taking the// node down, since surviving relay outages is the whole point of running more// than one.func (atsync *ATProtoSynchronizer) StartFirehose(ctx context.Context) error { ctx = log.WithLogValues(ctx, "func", "StartFirehose")
relays := atsync.relayHosts() if len(relays) == 0 { return fmt.Errorf("no relay hosts configured") }
atsync.commitDedup = newFirehoseDeduper(dedupWindow) atsync.identityDedup = newFirehoseDeduper(dedupWindow) atsync.markSeen() atsync.markEvent(time.Now())
log.Log(ctx, "starting firehose consumers", "relays", relays)
g, ctx := errgroup.WithContext(ctx) for _, relay := range relays { relay := relay spmetrics.TouchRelay(relay, relayProtocol(relay)) g.Go(func() error { atsync.consumeRelay(ctx, relay) return nil }) } g.Go(func() error { atsync.monitorFirehose(ctx) return nil }) return g.Wait()}
// consumeRelay keeps a single relay's firehose connected, reconnecting with// capped exponential backoff. It never returns an error: an unhealthy relay// must not disturb the others (or crash the node), so it just loops until ctx// is cancelled.func (atsync *ATProtoSynchronizer) consumeRelay(ctx context.Context, relay string) { ctx = log.WithLogValues(ctx, "relay", relay)
cursor := atsync.newRelayCursor(ctx, relay) // Persist progress on a timer and once more on the way out, so a restart // resumes near where we left off rather than re-tailing from live. flushDone := make(chan struct{}) go func() { defer close(flushDone) ticker := time.NewTicker(cursorFlushInterval) defer ticker.Stop() for { select { case <-ctx.Done(): cursor.flush(ctx) return case <-ticker.C: cursor.flush(ctx) } } }() defer func() { <-flushDone }()
const ( minBackoff = time.Second maxBackoff = 30 * time.Second ) backoff := minBackoff for { if ctx.Err() != nil { return } start := time.Now() err := atsync.connectRelay(ctx, relay, cursor) if ctx.Err() != nil { return } if err != nil { log.Error(ctx, "relay firehose disconnected; reconnecting", "err", err, "backoff", backoff) } else { log.Warn(ctx, "relay firehose closed; reconnecting", "backoff", backoff) } // A connection that stayed healthy for a while earns a fresh backoff. if time.Since(start) > time.Minute { backoff = minBackoff } select { case <-ctx.Done(): return case <-time.After(backoff): } if backoff < maxBackoff { backoff *= 2 if backoff > maxBackoff { backoff = maxBackoff } } }}
// firehoseReplayWindow is how stale a stored relay cursor may be and still be// worth replaying from. Zero (or negative) disables the cap, so every connect// resumes from whatever is stored.func (atsync *ATProtoSynchronizer) firehoseReplayWindow() time.Duration { if atsync.CLI == nil { return config.DefaultFirehoseReplayWindow } return atsync.CLI.FirehoseReplayWindow}
// dropStaleCursor abandons a relay cursor too old to be worth replaying from,// and orders a sweep to heal the gap that leaves.//// The kick matters: without it the gap sits there until the next scheduled// sweep, up to --sweep-interval away, which is the one way dropping the cursor// could be worse than replaying it. sweepOnce skips if a sweep is already// running, so several relays deciding this at once costs nothing.func (atsync *ATProtoSynchronizer) dropStaleCursor(ctx context.Context, cursor *relayCursor) { if !cursor.dropIfStale(ctx, atsync.firehoseReplayWindow()) { return } // The skipped span may hold #identity events as well as commits. Commits // the head check finds by asking hosts, but identity drift it can only see // by resolving — and a cached resolution from before the gap agrees with // the equally-stale repo row, hiding the change until the cache entry ages // out. Starting the cache over makes the healing sweep's resolutions // authoritative. atsync.resetIdentCache() if atsync.StatefulDB == nil { // No index to sweep (tests that exercise the firehose on its own). return } log.Log(ctx, "kicking off a sweep to heal the skipped firehose replay") go atsync.sweepOnce(ctx, atsync.Sweep)}
// connectRelay dials one relay and pumps its firehose until the connection// drops or ctx is cancelled. Event handlers run on the parent ctx (not the// per-connection one) so an in-flight commit keeps indexing across a reconnect// — important because dedup has already claimed it, so no other relay will// re-deliver it to us.func (atsync *ATProtoSynchronizer) connectRelay(ctx context.Context, relay string, cursor *relayCursor) error { u, err := url.Parse(relay) if err != nil { return fmt.Errorf("invalid relay URI %q: %w", relay, err) } // A cursor whose newest event is hours old asks the relay to re-send hours // of the whole network's traffic, at full speed, to a node that indexes a // fraction of a percent of it. Checked here rather than once at load // because a long backoff stretch ages a cursor that was fresh when we // started. atsync.dropStaleCursor(ctx, cursor) // MoQ relays (moqt:// and aliases) are consumed over QUIC instead of // WebSocket. Everything downstream of frame-decode — dedup, cursor, // handlers, backoff — is shared, so this is just a transport swap. switch u.Scheme { case "moqt", "moql", "moq", "moqs": return atsync.connectRelayMoq(ctx, relay, cursor) } u.Path = "xrpc/com.atproto.sync.subscribeRepos" if seq, ok := cursor.param(); ok { q := u.Query() q.Set("cursor", strconv.FormatInt(seq, 10)) u.RawQuery = q.Encode() }
header := http.Header{"User-Agent": []string{aqhttp.UserAgent}} // Our own loopback listener serves both the broadcaster and the // server-repo firehoses on one port and disambiguates them by the // request Host (see Server.isServerPDS). Dialing loopback sends // Host: 127.0.0.1, which lands us on the broadcaster firehose, so the // server repo's own records — most importantly place.stream.media.origin, // which getVideoList's "can this node serve it" filter depends on — // would never get indexed locally. Force Host to ServerHost so the // self-subscription gets the server-PDS firehose, matching what a // single-node (ServerHost == BroadcasterHost) deployment gets for free. // gorilla/websocket pulls the "Host" header out and uses it as the // HTTP Host while still dialing the loopback address in u. isSelf := relay == atsync.selfRelayURL() if isSelf && atsync.CLI.ServerHost != "" { header.Set("Host", atsync.CLI.ServerHost) }
dialer := websocket.DefaultDialer if isSelf && u.Scheme == "wss" { // Under --secure the self-subscription dials wss://127.0.0.1:<https-port>, // but our cert is issued for ServerHost, not for the loopback IP we dial // (and in dev it's frequently self-signed on top of that), so verification // would fail on hostname every time. Skipping it is not a trust decision: // the peer on the other end of this loopback socket is this same process. d := *websocket.DefaultDialer d.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} dialer = &d } con, _, err := dialer.Dial(u.String(), header) if err != nil { return fmt.Errorf("subscribing to firehose failed (dialing): %w", err) } defer con.Close()
protocol := relayProtocol(relay) spmetrics.FirehoseRelaysConnected.WithLabelValues(relay, protocol).Set(1) defer spmetrics.FirehoseRelaysConnected.WithLabelValues(relay, protocol).Set(0)
streamCtx, cancel := context.WithCancel(ctx) defer cancel()
rsc := atsync.repoStreamCallbacks(ctx, relay, cursor, cancel) scheduler := parallel.NewScheduler(10, 100, relay, rsc.EventHandler)
log.Log(ctx, "connected to relay firehose") return events.HandleRepoStream(streamCtx, con, scheduler, nil)}
// relayProtocol classifies a relay URL as "moq" (moqt:// and aliases) or// "websocket" (everything else), for per-protocol metric labels.func relayProtocol(relay string) string { if u, err := url.Parse(relay); err == nil { switch u.Scheme { case "moqt", "moql", "moq", "moqs": return "moq" } } return "websocket"}
// repoStreamCallbacks builds the event callbacks shared by the WebSocket and// MoQ transports: per-relay metrics, liveness marking, cursor advance,// cross-relay dedup, and running the indexing handlers. cancel ends the current// connection when the relay sends an error frame.//// The handlers run INLINE, on the callback's own goroutine, which is one of// indigo's parallel-scheduler workers. That is deliberate and load-bearing.// Spawning a goroutine per event here — which this used to do — returns from// the callback instantly and so defeats every bound the scheduler exists to// provide. Running them inline gets all three back://// - Concurrency is capped at the scheduler's worker count, instead of one// goroutine per event. A relay replaying a backlog at full send rate is// what turned that into millions of goroutines all queued behind a single// sqlite write connection.// - Events for one repo are handled in the order the relay sent them: the// scheduler chains same-DID tasks through a per-repo queue, so at most one// is in flight per repo. The per-event goroutines had silently dropped that// guarantee. It is per-relay, not global — each relay has its own scheduler,// so see revCASAttempts for the cross-relay race that remains.// - Backpressure. The scheduler's feeder channel is unbuffered, so once every// worker is busy, AddWork for a new repo blocks, HandleRepoStream (or the// moq read loop) stops reading the socket, and TCP tells the relay to slow// down. Overload degrades to "the firehose lags", which the sweep covers,// instead of to an OOM.//// Both transports feed the same scheduler (connectRelayMoq → dispatchMoqFrame →// scheduler.AddWork), so this holds for websocket and moq alike.//// The handlers keep using the captured ctx rather than the one the scheduler// hands them: parallel workers call the event handler with context.TODO(), and// the captured ctx is where our log values and shutdown cancellation live. It// is the parent ctx and not the per-connection one so an in-flight commit keeps// indexing across a reconnect.//// The tradeoff, accepted: on disconnect both transports defer// scheduler.Shutdown(), which lets in-flight handlers finish but drops tasks// still queued behind them — whose dedup claim has already been made, so no// other relay will re-deliver them. That is a small silent gap, and healing// exactly this kind of gap is what the head check and sweep are for.func (atsync *ATProtoSynchronizer) repoStreamCallbacks(ctx context.Context, relay string, cursor *relayCursor, cancel context.CancelFunc) *events.RepoStreamCallbacks { protocol := relayProtocol(relay) // observeSeq advances the cursor and republishes the per-relay high-water // seq gauge; both relays carry the upstream's seq, so the gauge's cross-relay // difference measures how far apart the relays are. // // On a moq relay only the gauge gets the upstream seq: there the resume // cursor holds MoQ group sequences (observeGroup in connectRelayMoq), and // upstream at-seqs are orders of magnitude larger, so feeding them into the // same max-wins cursor would bury the group cursor — the next SubscribeFrom // would then wait forever on a group the relay will never produce. moq := protocol == "moq" var gaugeHigh highWater // upstream-seq high-water for the moq gauge only observeSeq := func(seq int64) { if moq { gaugeHigh.observe(seq) spmetrics.FirehoseRelayHighSeq.WithLabelValues(relay, protocol).Set(float64(gaugeHigh.get())) return } cursor.observe(seq) spmetrics.FirehoseRelayHighSeq.WithLabelValues(relay, protocol).Set(float64(cursor.highSeq())) } // observeTime records how current this relay is. Unlike the seq it applies // to every transport — an event's stamp means the same thing whoever // carried it — and it is recorded before dedup, because a duplicate is // still proof of how far along this relay is. One small parse per event; an // unparsable stamp is simply not an observation. observeTime := func(t string) { aqt, err := aqtime.FromString(t) if err != nil { return } cursor.observeTime(aqt.Time()) } return &events.RepoStreamCallbacks{ RepoCommit: func(evt *indigoatproto.SyncSubscribeRepos_Commit) error { atsync.markSeen() observeSeq(evt.Seq) observeTime(evt.Time) spmetrics.FirehoseEventsReceivedTotal.WithLabelValues(relay, protocol, "commit").Inc() if atsync.commitDedup.seen(evt.Commit.String()) { spmetrics.FirehoseEventsDedupedTotal.WithLabelValues(relay, protocol, "commit").Inc() return nil } atsync.handleCommitEventOps(ctx, evt) return nil }, RepoIdentity: func(evt *indigoatproto.SyncSubscribeRepos_Identity) error { atsync.markSeen() observeSeq(evt.Seq) observeTime(evt.Time) spmetrics.FirehoseEventsReceivedTotal.WithLabelValues(relay, protocol, "identity").Inc() if atsync.identityDedup.seen(identityDedupKey(evt)) { spmetrics.FirehoseEventsDedupedTotal.WithLabelValues(relay, protocol, "identity").Inc() return nil } atsync.handleIdentityEventOps(ctx, evt) return nil }, Error: func(evt *events.ErrorFrame) error { cancel() return fmt.Errorf("firehose error: %s: %s", evt.Error, evt.Message) }, }}
// identityDedupKey distinguishes a single identity broadcast (so the same one// relayed by several relays collapses) from genuinely separate updates to the// same DID (a later handle change must NOT be dropped). The originating Time is// preserved as the event is relayed, so (did, time, handle) is stable for one// broadcast yet distinct across real changes. seq is deliberately excluded — it// is assigned per-relay and so differs for the very duplicates we want to fold.func identityDedupKey(evt *indigoatproto.SyncSubscribeRepos_Identity) string { handle := "" if evt.Handle != nil { handle = *evt.Handle } return evt.Did + "\x00" + evt.Time + "\x00" + handle}
// firehoseSyncPingInterval is how often monitorFirehose emits its Info-level// "in sync" heartbeat while healthy. Trouble is still reported within the 5s// tick; this only spaces out the all-is-well line so it doesn't spam.const firehoseSyncPingInterval = 60 * time.Second
func (atsync *ATProtoSynchronizer) monitorFirehose(ctx context.Context) { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() lastPing := time.Now() lastPingSeen := atsync.seenEvents.Load() for { select { case <-ctx.Done(): return case <-ticker.C: since := sinceNanos(atsync.lastEvent.Load()) goroutines := runtime.NumGoroutine() if since > 10*time.Second { log.Warn(ctx, fmt.Sprintf("firehose is %s behind real time", since), "goroutines", goroutines) } else { log.Debug(ctx, fmt.Sprintf("firehose is %s behind real time", since), "goroutines", goroutines) } dry := sinceNanos(atsync.lastSeen.Load()) if dry > 10*time.Second { log.Warn(ctx, fmt.Sprintf("firehose dry; no new events for %s", dry)) } // Periodic in-sync ping: healthy operation is otherwise Debug-only, // so surface a heartbeat at Info once a minute. An unhealthy stretch // (warns above) just delays the next ping; the ping never lies. if since <= 10*time.Second && dry <= 10*time.Second && time.Since(lastPing) >= firehoseSyncPingInterval { seen := atsync.seenEvents.Load() log.Log(ctx, "firehose in sync", "behind", since, "events", seen-lastPingSeen, "goroutines", goroutines) lastPing = time.Now() lastPingSeen = seen } } }}
var CollectionFilter = []string{ constants.APP_BSKY_GRAPH_FOLLOW, constants.APP_BSKY_FEED_POST, constants.APP_BSKY_GRAPH_BLOCK, constants.APP_BSKY_ACTOR_PROFILE, constants.PLACE_STREAM_LIVE_RECOMMENDATIONS,}
// indexedCollection reports whether we index anything in this collection: the// explicit [CollectionFilter] list plus everything under place.stream., which is// our own namespace and always ours to index. An empty filter means "index// everything", as it always has.func indexedCollection(collection string) bool { if len(CollectionFilter) == 0 { return true } if slices.Contains(CollectionFilter, collection) { return true } return strings.HasPrefix(collection, "place.stream.")}
// commitHasIndexedOps reports whether a commit touches anything we index, from// the op paths alone — no CAR parsing.//// This is what lets handleCommitEventOps skip the CAR parse for the large// majority of firehose traffic (the whole network's posts, likes and follows on// repos we have never heard of). Handlers run on a bounded worker pool now, so// that parse is no longer paid by a throwaway goroutine: it is paid out of the// pool, and directly caps how fast the node can catch up.//// An op whose path does not parse counts as indexed, so that the main loop// reaches it and logs the same error it always has.func commitHasIndexedOps(evt *indigoatproto.SyncSubscribeRepos_Commit) bool { for _, op := range evt.Ops { collection, _, err := syntax.ParseRepoPath(op.Path) if err != nil { return true } if indexedCollection(collection.String()) { return true } } return false}
func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt *indigoatproto.SyncSubscribeRepos_Commit) { ctx = log.WithLogValues(ctx, "event", "commit", "did", evt.Repo, "rev", evt.Rev, "seq", fmt.Sprintf("%d", evt.Seq), "func", "handleCommitEventOps")
if evt.TooBig { log.Warn(ctx, "skipping tooBig events for now") return }
// A commit with nothing we index still has to be tracked: trackCommitRev // follows the rev chain for repos we DO track, and a tracked repo posting a // like is a commit whose ops are all filtered. Skipping it there would // leave our stored rev behind, and the next commit we did care about would // look like a hole and order a pointless repair. So the ops are what get // skipped, not the event — and trackCommitRev never needed the CAR anyway. if commitHasIndexedOps(evt) && !atsync.handleIndexedOps(ctx, evt) { return }
// Every op in this commit is indexed, so the index can claim this commit. // Only reached on a clean pass: an event we bailed out of half-applied // leaves the stored rev where it was, and the next commit for that repo // notices the hole and orders a repair. atsync.trackCommitRev(ctx, evt)}
// handleIndexedOps reads the commit's CAR and applies the ops we index,// reporting whether it got all the way through. It is split out of// handleCommitEventOps so that a commit with nothing for us can skip the CAR// parse entirely while still having its rev tracked; a bail-out here (an// unreadable CAR, an unparsable path) reports false and so leaves the repo's// stored rev where it was, exactly as before, so the next commit finds the hole.func (atsync *ATProtoSynchronizer) handleIndexedOps(ctx context.Context, evt *indigoatproto.SyncSubscribeRepos_Commit) bool { rr, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks)) if err != nil { log.Error(ctx, "failed to read repo from car", "err", err) return false }
for _, op := range evt.Ops { collection, rkey, err := syntax.ParseRepoPath(op.Path) uri := fmt.Sprintf("at://%s/%s", evt.Repo, op.Path) if err != nil { log.Error(ctx, "invalid path in repo op", "eventKind", op.Action, "path", op.Path) return false } ctx := log.WithLogValues(ctx, "eventKind", op.Action, "collection", collection.String(), "rkey", rkey.String())
if !indexedCollection(collection.String()) { continue }
aqt, err := aqtime.FromString(evt.Time) if err != nil { log.Error(ctx, "failed to parse time", "err", err) continue } opTime := aqt.Time() atsync.markEvent(opTime)
r, err := atsync.Model.GetRepo(evt.Repo) if err != nil { log.Error(ctx, "failed to get repo", "err", err) continue } if r != nil { // One enrichment here gives every line under this op a handle to go // with the DID -- the dispatch switch below, the contiguity checks, // and everything reviveRepo and handleCreateUpdate log through ctx // -- without any of them doing a lookup of their own. ctx = log.WithLogValues(ctx, "handle", r.Handle) } atsync.reviveRepo(ctx, r) // log.Warn(ctx, "got record we care about", "collection", collection, "rkey", rkey)
ek := repomgr.EventKind(op.Action) switch ek { case repomgr.EvtKindCreateRecord, repomgr.EvtKindUpdateRecord: // read the record bytes from blocks, and verify CID rc, recCBOR, err := rr.GetRecordBytes(ctx, op.Path) if err != nil { log.Error(ctx, "reading record from event blocks (CAR)", "err", err) break } if !op.Cid.Defined() || glex.Link(rc) != glex.Link(cid.Cid(*op.Cid)) { log.Error(ctx, "mismatch between commit op CID and record block", "recordCID", rc, "opCID", op.Cid) break }
// Defense-in-depth: handleCreateUpdate decodes arbitrary CBOR // and dispatches across every record type, so a malformed record // or a bug in any one handler can panic. Recover it so one bad // event can't crash the firehose consumer goroutine (which would // tear down event processing for the whole node). The error is // logged and the loop moves on to the next op. err = log.Recover(ctx, func() error { return atsync.handleCreateUpdate(ctx, evt.Repo, rkey, recCBOR, op.Cid.String(), collection, ek == repomgr.EvtKindUpdateRecord, false) }) if err != nil { log.Error(ctx, "failed to handle create update", "err", err) continue }
case repomgr.EvtKindDeleteRecord: if collection.String() == constants.APP_BSKY_GRAPH_FOLLOW { if r == nil { log.Debug(ctx, "no repo found for follow", "userDID", evt.Repo, "subjectDID", rkey.String()) continue } log.Debug(ctx, "deleting follow", "userDID", evt.Repo, "subjectDID", rkey.String()) err := atsync.Model.DeleteFollow(ctx, evt.Repo, rkey.String()) if err != nil { log.Debug(ctx, "failed to delete follow", "err", err) } }
if collection.String() == constants.APP_BSKY_GRAPH_BLOCK { if r == nil { log.Debug(ctx, "no repo found for block", "userDID", evt.Repo, "subjectDID", rkey.String()) continue } log.Warn(ctx, "deleting block", "userDID", evt.Repo, "subjectDID", rkey.String()) err := atsync.Model.DeleteBlock(ctx, rkey.String()) if err != nil { log.Error(ctx, "failed to delete block", "err", err) } }
if collection.String() == constants.PLACE_STREAM_KEY { log.Warn(ctx, "revoking stream key", "userDID", evt.Repo, "rkey", rkey.String()) key, err := atsync.Model.GetSigningKeyByRKey(ctx, rkey.String()) if err != nil { log.Error(ctx, "failed to get signing key", "err", err) continue } if key == nil { log.Warn(ctx, "no signing key found for stream key", "userDID", evt.Repo, "rkey", rkey.String()) continue } now := time.Now() key.RevokedAt = &now err = atsync.Model.UpdateSigningKey(key) if err != nil { log.Error(ctx, "failed to revoke signing key", "err", err) } atsync.Bus.Publish(evt.Repo, key) }
if collection.String() == constants.PLACE_STREAM_CHAT_MESSAGE { msg, err := atsync.Model.GetChatMessage(uri) if err != nil { log.Error(ctx, "failed to get chat message", "err", err) continue } if msg == nil { log.Warn(ctx, "no chat message found for uri", "uri", uri) continue } log.Warn(ctx, "deleting chat message", "userDID", evt.Repo, "uri", uri) err = atsync.Model.DeleteChatMessage(ctx, uri, &opTime) if err != nil { log.Error(ctx, "failed to delete chat message", "err", err) continue } mv, err := msg.ToStreamplaceMessageView() if err != nil { log.Error(ctx, "failed to convert chat message to streamplace message view", "err", err) continue } isTrue := true mv.Deleted = &isTrue atsync.Bus.Publish(msg.StreamerRepoDID, mv) }
if collection.String() == constants.PLACE_STREAM_LIVE_TELEPORT { err := atsync.Model.DeleteTeleport(ctx, uri) if err != nil { log.Error(ctx, "failed to delete teleport", "err", err) } }
if collection.String() == constants.PLACE_STREAM_MODERATION_PERMISSION { log.Debug(ctx, "deleting moderation delegation", "userDID", evt.Repo, "rkey", rkey.String()) err := atsync.Model.DeleteModerationDelegation(ctx, rkey.String()) if err != nil { log.Error(ctx, "failed to delete moderation delegation", "err", err) } // Publish deletion to WebSocket bus for real-time updates // Create a deleted record marker to notify frontend deletedRecord := map[string]any{ "$type": "place.stream.moderation.permission", "deleted": true, "rkey": rkey.String(), "streamer": evt.Repo, } go atsync.Bus.Publish(evt.Repo, deletedRecord) }
if collection.String() == constants.PLACE_STREAM_CHAT_GATE { log.Debug(ctx, "deleting gate", "userDID", evt.Repo, "rkey", rkey.String()) err := atsync.Model.DeleteGate(ctx, rkey.String()) if err != nil { log.Error(ctx, "failed to delete gate", "err", err) } }
if collection.String() == constants.PLACE_STREAM_CHAT_PINNED_RECORD { log.Debug(ctx, "deleting pinned record", "userDID", evt.Repo, "rkey", rkey.String()) err := atsync.Model.DeletePinnedRecord(ctx, rkey.String()) if err != nil { log.Error(ctx, "failed to delete pinned record", "err", err) } deletedPin := map[string]any{ "$type": constants.PLACE_STREAM_CHAT_PINNED_RECORD, "rkey": rkey.String(), "deleted": true, } go atsync.Bus.Publish(evt.Repo, deletedPin) }
if collection.String() == constants.PLACE_STREAM_BADGE_DEF { log.Debug(ctx, "deleting badge def", "uri", uri) err := atsync.Model.DeleteBadgeDef(ctx, uri) if err != nil { log.Error(ctx, "failed to delete badge def", "err", err) } }
if collection.String() == constants.PLACE_STREAM_BADGE_ISSUANCE { log.Debug(ctx, "deleting badge issuance", "uri", uri) err := atsync.Model.DeleteBadgeIssuance(ctx, uri) if err != nil { log.Error(ctx, "failed to delete badge issuance", "err", err) } }
if collection.String() == constants.PLACE_STREAM_VIDEO { log.Debug(ctx, "deleting video", "uri", uri) if err := atsync.Model.DeleteVideo(ctx, uri); err != nil { log.Error(ctx, "failed to delete video", "err", err) } }
if collection.String() == constants.PLACE_STREAM_MEDIA_TRACK { log.Debug(ctx, "deleting media track", "uri", uri) if err := atsync.Model.DeleteMediaTrack(ctx, uri); err != nil { log.Error(ctx, "failed to delete media track", "err", err) } }
if collection.String() == constants.PLACE_STREAM_MEDIA_ORIGIN { log.Debug(ctx, "deleting media origin", "uri", uri) if err := atsync.Model.DeleteMediaOrigin(ctx, uri); err != nil { log.Error(ctx, "failed to delete media origin", "err", err) } }
if collection.String() == constants.PLACE_STREAM_BETA_INVITE { log.Debug(ctx, "deleting beta invite", "uri", uri) if err := atsync.Model.DeleteBetaInvite(ctx, uri); err != nil { log.Error(ctx, "failed to delete beta invite", "err", err) } }
if collection.String() == constants.PLACE_STREAM_BETA_REQUEST { log.Debug(ctx, "deleting beta request", "uri", uri) if err := atsync.Model.DeleteBetaRequest(ctx, uri); err != nil { log.Error(ctx, "failed to delete beta request", "err", err) } }
if collection.String() == constants.PLACE_STREAM_MEDIA_VIEW_COUNT { log.Debug(ctx, "deleting media view count", "uri", uri) if err := atsync.Model.DeleteMediaViewCount(ctx, uri); err != nil { log.Error(ctx, "failed to delete media view count", "err", err) } }
if collection.String() == constants.PLACE_STREAM_VOD_COMMENT { log.Debug(ctx, "deleting VOD comment", "uri", uri) if err := atsync.Model.DeleteVodComment(ctx, uri, &opTime); err != nil { log.Error(ctx, "failed to delete VOD comment", "err", err) } }
if collection.String() == constants.PLACE_STREAM_LIKE { log.Debug(ctx, "deleting like", "uri", uri) if err := atsync.Model.DeleteLike(ctx, uri); err != nil { log.Error(ctx, "failed to delete like", "err", err) } }
if collection.String() == constants.PLACE_STREAM_VOD_GATE { log.Debug(ctx, "deleting VOD gate", "userDID", evt.Repo, "rkey", rkey.String()) if err := atsync.Model.DeleteVodGate(ctx, rkey.String()); err != nil { log.Error(ctx, "failed to delete VOD gate", "err", err) } }
default: log.Error(ctx, "unexpected record op kind") } }
return true}
// reviveRepo un-parks a repo we had written off. A commit event is proof the// account is back -- a deactivated, suspended, or deleted repo cannot write --// so the terminal status goes away and the ordinary wedge logic (an empty// Version means "backfill me") takes it from there.//// It is called from the commit path with the row that path already loaded, so// the common case costs one comparison and no query at all.func (atsync *ATProtoSynchronizer) reviveRepo(ctx context.Context, r *model.Repo) { if !r.TerminalStatus() { return } log.Log(ctx, "repo committed while parked, clearing terminal status", "did", r.DID, "handle", r.Handle, "status", r.Status) if err := atsync.Model.SetRepoStatus(ctx, r.DID, model.RepoStatusOK); err != nil { log.Error(ctx, "failed to clear repo status", "did", r.DID, "handle", r.Handle, "err", err) return } r.Status = model.RepoStatusOK}
func (atsync *ATProtoSynchronizer) handleIdentityEventOps(ctx context.Context, evt *indigoatproto.SyncSubscribeRepos_Identity) { handle := "" if evt.Handle != nil { handle = *evt.Handle } ctx = log.WithLogValues(ctx, "event", "identity", "did", evt.Did, "handle", handle, "func", "handleIdentityEventOps") r, err := atsync.Model.GetRepo(evt.Did) if err != nil { log.Error(ctx, "failed to get repo", "err", err) return } if r == nil { log.Debug(ctx, "no repo found for identity", "did", evt.Did) return } _, err = atsync.RefreshIdentity(ctx, evt.Did) if err != nil { log.Error(ctx, "failed to refresh ident", "err", err) return }}