fix webhook backlink intake stalling on prior-reference capacity
1599 unmodified linesExpand all160016011602160316041604160516061607160826 unmodified lines163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916602 unmodified lines16631664166516661649165016671668166916701653167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171016681711171217131714171517161717167516761718171917201679168016811682168316841685172117221723172417251726172717281729173017311732168716881689169016911692169316941695169616971698173317341735173617371738173917401741174217001743174417451746174717481749175017511752175317541755175617571758175917601709171017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001714180118021803180418051 unmodified line18071808180918101724181118121813181418155 unmodified lines182118221823182418251826173818271828182918301831573 unmodified linesExpand all1599 unmodified linesExpand all throw new Error('Invalid Jetstream consumer') await saveJetstreamCursor(streamId, relay, cursor)}
const MAX_BACKLINK_REFERENCE_ROWS = 100_000export const MAX_BACKLINK_REFERENCE_ROWS = 100_000export const MAX_BACKLINK_REFERENCES_PER_EVENT = 100const MAX_BACKLINK_REFERENCE_BYTES = 32 * 1024const BACKLINK_REFERENCE_TTL_MS = 7 * 24 * 60 * 60_000
26 unmodified lines timeUs: number rev: string}
/** * Decode a stored reference list. Rows written before the jsonb binding fix * hold the encoded array as a jsonb *string*, so both shapes are accepted; * anything else is absent state rather than trusted state. */function decodeReferences(value: unknown): string[] | undefined { let decoded = value if (typeof decoded === 'string') { try { decoded = JSON.parse(decoded) } catch { return undefined } } if (!Array.isArray(decoded) || !decoded.every((reference) => typeof reference === 'string')) return undefined return decoded}
/** Load bounded prior relevant backlink references after restart. */export async function loadPriorReferenceIndex(eventAtUri: string): Promise<PriorReferenceIndexEntry | undefined> { const key = normalizeAtUri(eventAtUri) const rows = await db<Array<{ refs: unknown; last_seq: number | string | bigint; last_rev: string }>>`2 unmodified lines WHERE event_at_uri = ${key} AND expires_at > NOW() LIMIT 1 ` const row = rows[0] if (!row || !Array.isArray(row.refs) || !row.refs.every((reference) => typeof reference === 'string')) return undefined const references = row ? decodeReferences(row.refs) : undefined if (!row || !references) return undefined return { eventAtUri: key, references: row.refs.map((reference) => normalizeAtUri(reference)), references: references.map((reference) => normalizeAtUri(reference)), timeUs: parseSafeSequence(row.last_seq, 'Invalid backlink reference'), rev: row.last_rev, }}
/** * The durable outcome of one prior-reference write. Intake keeps an in-memory * index of the keys that own durable state, so a write reports what the row * now holds rather than only that a statement ran. */export type PriorReferenceWrite = /** References are durable for this key at this event version. */ | 'stored' /** The row survives as an empty tombstone; no reference remains. */ | 'cleared' /** A newer event already owns this key, or no row exists to clear. */ | 'stale' /** Global capacity refused a new key. Existing keys are never refused. */ | 'rejected'
const CAPACITY_WARN_INTERVAL_MS = 60_000let lastCapacityWarnAt = 0
/** * Persist refs only after filtering them to active backlink scopes. Event order * is monotonic: a replayed older create/update cannot replace newer references. * * One statement, no advisory lock. The backlink consumer sees every record in * the network, so a per-event transaction round trip is the intake throughput * ceiling. An empty set is a tombstone over known state and never creates a * row, which keeps this bounded table holding references rather than the relay. * Capacity is admitted per statement and converged by prunePriorReferenceIndex, * so concurrent writers may overshoot the bound by their own concurrency. */export async function savePriorReferenceIndex( eventAtUri: string, references: readonly string[], timeUs: number, rev: string,): Promise<void> {): Promise<PriorReferenceWrite> { const key = normalizeAtUri(eventAtUri) if (!Number.isSafeInteger(timeUs) || timeUs < 0) throw new Error('Invalid backlink reference') assertAtprotoRevision(rev) if (references.length > MAX_BACKLINK_REFERENCES_PER_EVENT) throw new Error('Too many backlink references') const normalized = [...new Set(references.map(normalizeAtUri))] if (normalized.length > MAX_BACKLINK_REFERENCES_PER_EVENT) throw new Error('Too many backlink references') const encoded = JSON.stringify(normalized) if (new TextEncoder().encode(encoded).byteLength > MAX_BACKLINK_REFERENCE_BYTES) if (new TextEncoder().encode(JSON.stringify(normalized)).byteLength > MAX_BACKLINK_REFERENCE_BYTES) throw new Error('Backlink references are too large')
await db.begin(async (tx) => { // Serialize only the capacity decision. This prevents a burst of distinct // records from exceeding the global bounded-reference admission cap. await tx`SET LOCAL lock_timeout = '1000ms'` await tx`SELECT pg_advisory_xact_lock(814732193)` const existing = await tx<Array<{ event_at_uri: string }>>` SELECT event_at_uri FROM webhook_backlink_references WHERE event_at_uri = ${key} FOR UPDATE if (normalized.length === 0) { const cleared = await db<Array<{ event_at_uri: string }>>` UPDATE webhook_backlink_references SET refs = '[]'::jsonb, last_seq = ${timeUs}::bigint, last_rev = ${rev}::text, expires_at = NOW() + (${BACKLINK_REFERENCE_TTL_MS} * INTERVAL '1 millisecond'), updated_at = NOW() WHERE event_at_uri = ${key} AND (last_seq < ${timeUs}::bigint OR (last_seq = ${timeUs}::bigint AND last_rev <= ${rev}::text)) RETURNING event_at_uri ` if (existing.length === 0) { const capacity = await tx<Array<{ count: number | string | bigint }>>` SELECT count(*) AS count FROM webhook_backlink_references ` if (parseSafeSequence(capacity[0]?.count, 'Invalid backlink reference state') >= MAX_BACKLINK_REFERENCE_ROWS) { // The current event can still complete direct matching/delivery. It // simply has no durable backlink prior-state entry until capacity frees. logger.warn('[DB] backlink reference capacity reached') return } } await tx` return cleared.length > 0 ? 'cleared' : 'stale' }
const rows = await db<Array<{ admitted: boolean; written: boolean }>>` WITH admission AS ( SELECT EXISTS (SELECT 1 FROM webhook_backlink_references WHERE event_at_uri = ${key}) OR (SELECT count(*) FROM webhook_backlink_references) < ${MAX_BACKLINK_REFERENCE_ROWS} AS admitted ), written AS ( INSERT INTO webhook_backlink_references (event_at_uri, refs, last_seq, last_rev, expires_at, updated_at) VALUES (${key}, ${encoded}::jsonb, ${timeUs}, ${rev}, NOW() + (${BACKLINK_REFERENCE_TTL_MS} * INTERVAL '1 millisecond'), NOW()) SELECT ${key}::text, -- The array binds as jsonb. A pre-encoded string would bind as a jsonb -- *string*, which no reader can distinguish from a one-element list. ${normalized}::jsonb, ${timeUs}::bigint, ${rev}::text, NOW() + (${BACKLINK_REFERENCE_TTL_MS} * INTERVAL '1 millisecond'), NOW() FROM admission WHERE admission.admitted ON CONFLICT (event_at_uri) DO UPDATE SET refs = EXCLUDED.refs, last_seq = EXCLUDED.last_seq, last_rev = EXCLUDED.last_rev, expires_at = EXCLUDED.expires_at, updated_at = NOW() WHERE webhook_backlink_references.last_seq < EXCLUDED.last_seq OR (webhook_backlink_references.last_seq = EXCLUDED.last_seq AND webhook_backlink_references.last_rev <= EXCLUDED.last_rev) ` }) RETURNING event_at_uri ) SELECT admission.admitted, EXISTS (SELECT 1 FROM written) AS written FROM admission ` const outcome = rows[0] if (outcome?.written === true) return 'stored' if (outcome?.admitted === false) { // The event still completes direct matching and delivery; it simply has no // durable backlink prior state until maintenance frees capacity. One // warning per interval: this path is reached once per relay event. const now = Date.now() if (now - lastCapacityWarnAt >= CAPACITY_WARN_INTERVAL_MS) { lastCapacityWarnAt = now logger.warn('[DB] backlink reference capacity reached') } return 'rejected' } return 'stale'}
/** Bounded hydration of the keys that own durable references after a restart. */export async function loadPriorReferenceKeys( limit = MAX_BACKLINK_REFERENCE_ROWS,): Promise<{ keys: string[]; complete: boolean }> { if (!Number.isSafeInteger(limit) || limit < 1 || limit > MAX_BACKLINK_REFERENCE_ROWS) { throw new Error('Invalid backlink reference key limit') } const rows = await db<Array<{ event_at_uri: string }>>` SELECT event_at_uri FROM webhook_backlink_references -- '"[]"' is the legacy double-encoded empty list. Both spellings mean the -- key owns no reference and therefore needs no in-memory entry. WHERE expires_at > NOW() AND refs::text NOT IN ('[]', '"[]"') ORDER BY expires_at DESC LIMIT ${limit + 1} ` // An incomplete index makes intake load every key instead of trusting it. return { keys: rows.slice(0, limit).map((row) => row.event_at_uri), complete: rows.length <= limit }}
/** Delete only if a later event has not already repopulated the same record. */export async function deletePriorReferenceIndex(eventAtUri: string, timeUs: number, rev: string): Promise<void> {export async function deletePriorReferenceIndex(eventAtUri: string, timeUs: number, rev: string): Promise<boolean> { const key = normalizeAtUri(eventAtUri) if (!Number.isSafeInteger(timeUs) || timeUs < 0) throw new Error('Invalid backlink reference') assertAtprotoRevision(rev) const rows = await db<Array<{ event_at_uri: string }>>`1 unmodified line WHERE event_at_uri = ${key} AND (last_seq < ${timeUs} OR (last_seq = ${timeUs} AND last_rev <= ${rev})) RETURNING event_at_uri ` void rows return rows.length > 0}
/** Periodic bounded maintenance for TTL and global capacity. */export async function prunePriorReferenceIndex(batchSize = 1_000): Promise<number> {5 unmodified lines SELECT count(*) AS total FROM webhook_backlink_references ), victims AS ( SELECT refs.event_at_uri FROM webhook_backlink_references refs CROSS JOIN stats -- Admission refuses a new key at the bound, so eviction must start there -- too. A strictly greater test deadlocks the table at exactly capacity. WHERE refs.expires_at <= NOW() OR stats.total > ${MAX_BACKLINK_REFERENCE_ROWS} WHERE refs.expires_at <= NOW() OR stats.total >= ${MAX_BACKLINK_REFERENCE_ROWS} ORDER BY refs.expires_at ASC, refs.updated_at ASC LIMIT ${batchSize} FOR UPDATE OF refs SKIP LOCKED )573 unmodified linesExpand all5 unmodified lines678910111213141515161617181920212264 unmodified lines87888990899091929394919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140108 unmodified linesExpand all2492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413425 unmodified lines// Firehose imports its database namespace at module load. Keep unit tests fully// in-memory; the real DB behavior is covered by db/backfill tests.mock.module('./db', () => ({ MAX_BACKLINK_REFERENCES_PER_EVENT: 100, MAX_BACKLINK_REFERENCE_ROWS: 100_000, loadCursor: async () => undefined, saveCursor: async () => undefined, loadCursorForStream: async () => undefined, saveCursorForStream: async () => undefined, loadPriorReferenceIndex: async () => undefined, savePriorReferenceIndex: async () => undefined, deletePriorReferenceIndex: async () => undefined, savePriorReferenceIndex: async () => 'stored', deletePriorReferenceIndex: async () => false, loadPriorReferenceKeys: async () => ({ keys: [], complete: true }), enqueueWebhookDeliveries: async () => ({ enqueued: 0, deduplicated: 0 }), recordWebhookIntakeQuarantine: async () => undefined, getWebhookRecord: async () => undefined, upsertWebhookRecord: async () => false,64 unmodified lines commit: { rev: REV, operation: 'create', collection: 'app.bsky.feed.post', rkey: 'one', record: { text: 'ok' } }, }}
function testDependencies(created: FakeJetstream[]) { const cursor: CursorRepository = { load: async () => undefined, save: async () => undefined } const references: PriorReferenceRepository = { load: async () => undefined, save: async () => undefined, delete: async () => undefined,/** In-memory stand-in for the durable reference table that records its traffic. */interface ReferenceProbe extends PriorReferenceRepository { readonly rows: Map<string, readonly string[]> readonly loads: string[] readonly writes: string[] complete: boolean}
function referenceProbe(seed: Iterable<[string, readonly string[]]> = []): ReferenceProbe { const rows = new Map<string, readonly string[]>(seed) const probe: ReferenceProbe = { rows, loads: [], writes: [], complete: true, async load(key) { probe.loads.push(key) const references = rows.get(key) return references ? { references, timeUs: 0, rev: REV } : undefined }, async save(key, references) { probe.writes.push(key) if (references.length === 0) { if (!rows.has(key)) return 'stale' rows.set(key, []) return 'cleared' } rows.set(key, [...references]) return 'stored' }, async delete(key) { probe.writes.push(key) return rows.delete(key) }, async keys() { return { keys: [...rows].filter(([, references]) => references.length > 0).map(([key]) => key), complete: probe.complete, } }, } return probe}
function testDependencies(created: FakeJetstream[], references: PriorReferenceRepository = referenceProbe()) { const cursor: CursorRepository = { load: async () => undefined, save: async () => undefined } return { cursorRepository: cursor, referenceRepository: references, createJetstreamClient: (options: JetstreamOptions) => {108 unmodified linesExpand all await direct?.options.onEvent(commit(40)) expect(candidateCount).toBe(1) })})
const VISITOR = 'did:plc:cccccccccccccccccccccccc'const VISITOR_KEY = `at://${VISITOR}/app.bsky.feed.like/like`const SUBJECT = `at://${SCOPE}/app.bsky.feed.post/abc`
function backlinkRecord(): WhRecord { return { $type: 'place.wisp.v2.wh', scope: { aturi: `at://${SCOPE}`, backlinks: true }, url: 'https://receiver.example/hook', createdAt: '2025-01-01T00:00:00.000Z', }}
function likeCommit(timeUs: number, references: readonly string[], operation: 'create' | 'update' = 'create') { return { did: VISITOR, time_us: timeUs, kind: 'commit' as const, commit: { rev: REV, operation, collection: 'app.bsky.feed.like', rkey: 'like', record: references.length > 0 ? { subject: { uri: references[0] } } : { text: 'unrelated' }, }, }}
async function startBacklinkIntake(created: FakeJetstream[], probe: ReferenceProbe) { initScopeDids([{ did: OWNER, rkey: 'hook', record: backlinkRecord() }]) await startFirehose({ ...testDependencies(created, probe), enqueueWebhookDeliveries: async (entries) => ({ enqueued: entries.length, deduplicated: 0 }), recordWebhookIntakeQuarantine: async () => undefined, }) const backlink = created.find((client) => !client.options.wantedDids && !client.options.wantedCollections) expect(backlink).toBeDefined() return backlink as FakeJetstream}
describe('backlink prior-reference index', () => { test('a relay record that references nothing in scope costs no durable read or write', async () => { const created: FakeJetstream[] = [] const probe = referenceProbe() const backlink = await startBacklinkIntake(created, probe) await backlink.options.onEvent(likeCommit(10, [])) expect(probe.loads).toEqual([]) expect(probe.writes).toEqual([]) })
test('remembers a referencing record and forgets it once its references are gone', async () => { const created: FakeJetstream[] = [] const probe = referenceProbe() const backlink = await startBacklinkIntake(created, probe)
await backlink.options.onEvent(likeCommit(10, [SUBJECT])) expect(probe.loads).toEqual([]) expect(probe.writes).toEqual([VISITOR_KEY]) expect(probe.rows.get(VISITOR_KEY)).toEqual([SUBJECT])
// The same key may now own durable state, so its next version is read. await backlink.options.onEvent(likeCommit(11, [], 'update')) expect(probe.loads).toEqual([VISITOR_KEY]) expect(probe.rows.get(VISITOR_KEY)).toEqual([])
// Cleared state is forgotten again: no further round trip for this key. await backlink.options.onEvent(likeCommit(12, [], 'update')) expect(probe.loads).toEqual([VISITOR_KEY]) expect(probe.writes).toEqual([VISITOR_KEY, VISITOR_KEY]) })
test('reads durable state for every key when hydration is truncated', async () => { const created: FakeJetstream[] = [] const probe = referenceProbe() probe.complete = false const backlink = await startBacklinkIntake(created, probe) await backlink.options.onEvent(likeCommit(10, [])) expect(probe.loads).toEqual([VISITOR_KEY]) })
test('hydrates keys that already own references so their next event is matched', async () => { const created: FakeJetstream[] = [] const probe = referenceProbe([[VISITOR_KEY, [SUBJECT]]]) const backlink = await startBacklinkIntake(created, probe) await backlink.options.onEvent(likeCommit(10, [], 'update')) expect(probe.loads).toEqual([VISITOR_KEY]) expect(probe.rows.get(VISITOR_KEY)).toEqual([]) })})12344567848 unmodified lines575859606162616263646566676869201 unmodified linesExpand all2712722732742722732752762772782792802816 unmodified lines288289290291292293294295296297298299529 unmodified linesExpand all8298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109118328338349129139149159169179189199209219229238 unmodified lines932933934935856936937938860939940941942943865944945946947948867949950951952953954873955956957958959314 unmodified linesExpand all127412751276127712781279128012811282120 unmodified linesExpand all14031404140514061407140814091410141120 unmodified linesimport type { Main as WhRecord } from '@wispplace/lexicons/types/place/wisp/v2/wh'import { createLogger } from '@wispplace/observability'import { config } from '../config'import type { WebhookEntry } from './db'import type { PriorReferenceWrite, WebhookEntry } from './db'import * as database from './db'import * as delivery from './delivery'import { buildJetstreamSubscriptionUrl,48 unmodified lines
/** Durable, bounded backlink state. It deliberately stores references, never whole records. */export interface PriorReferenceRepository { load(eventAtUri: string): Promise<PriorReferenceSnapshot | undefined> save(eventAtUri: string, references: readonly string[], timeUs: number, rev: string): Promise<void> delete(eventAtUri: string, timeUs: number, rev: string): Promise<void> save(eventAtUri: string, references: readonly string[], timeUs: number, rev: string): Promise<PriorReferenceWrite> /** True only when this event removed the durable row. */ delete(eventAtUri: string, timeUs: number, rev: string): Promise<boolean> /** Bounded hydration of the keys that own durable references, when available. */ keys?(): Promise<{ keys: readonly string[]; complete: boolean }>}
interface DurableDeliveryEvent { relay: string201 unmodified linesExpand all eventAtUri: string, references: readonly string[], timeUs: number, rev: string, ) => Promise<void> deletePriorReferenceIndex?: (eventAtUri: string, timeUs: number, rev: string) => Promise<void> ) => Promise<PriorReferenceWrite> deletePriorReferenceIndex?: (eventAtUri: string, timeUs: number, rev: string) => Promise<boolean> loadPriorReferenceKeys?: () => Promise<{ keys: readonly string[]; complete: boolean }> } return { async load(key) { if (!api.loadPriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable')6 unmodified lines async delete(key, timeUs, rev) { if (!api.deletePriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable') return api.deletePriorReferenceIndex(key, timeUs, rev) }, async keys() { if (!api.loadPriorReferenceKeys) throw new Error('Durable prior-reference storage is unavailable') return api.loadPriorReferenceKeys() }, }}
function defaultEnqueueWebhookDeliveries(): EnqueueWebhookDeliveries {529 unmodified linesExpand all readonly matchRecord: unknown readonly tooComplex: boolean}
/** * The keys that own durable prior references. The backlink consumer reads every * record in the network, so without this index each irrelevant event costs a * database round trip and intake can never keep pace with the relay. * * `null` means the index is not trustworthy - never hydrated, hydration was * truncated, or it outgrew the durable bound - and every key is then loaded * from the database, which is slow but always correct. * * It mirrors a table this process is the only writer of. A second intake * replica would need its own hydration, which startup already performs. */let priorReferenceKeys: Set<string> | null = null
/** Does this key possibly own durable prior state? Unknown answers "yes". */function mayOwnPriorReferences(key: string): boolean { return priorReferenceKeys === null || priorReferenceKeys.has(key)}
/** What a durable write leaves behind; `undefined` means it changed nothing. */const PRIOR_STATE_AFTER_WRITE: Record<PriorReferenceWrite, boolean | undefined> = { stored: true, cleared: false, rejected: false, stale: undefined,}
function trackPriorReferences(key: string, owned: boolean): void { if (!priorReferenceKeys) return if (!owned) { priorReferenceKeys.delete(key) return } if (priorReferenceKeys.size >= database.MAX_BACKLINK_REFERENCE_ROWS && !priorReferenceKeys.has(key)) { // Past the durable bound the set can no longer mirror the table. Loading // every key stays correct, so degrade instead of matching on a lie. priorReferenceKeys = null logger.warn('[backlink] prior-reference index exceeded its bound; reading durable state for every event') return } priorReferenceKeys.add(key)}
async function hydratePriorReferenceKeys(): Promise<void> { const repository = currentDependencies().referenceRepository if (!repository.keys) { priorReferenceKeys = null return } try { const { keys, complete } = await repository.keys() priorReferenceKeys = complete ? new Set(keys) : null } catch { // Never fail startup over an optimization; fall back to durable reads. priorReferenceKeys = null logger.warn('[backlink] prior-reference index hydration failed; reading durable state for every event') }}
/** Persist references, skipping the round trip when nothing is or becomes durable. */async function writePriorReferences( key: string, references: readonly string[], timeUs: number, rev: string,): Promise<void> { if (references.length === 0 && !mayOwnPriorReferences(key)) return const owned = PRIOR_STATE_AFTER_WRITE[await currentDependencies().referenceRepository.save(key, references, timeUs, rev)] if (owned !== undefined) trackPriorReferences(key, owned)}
async function loadReferenceLifecycle( event: JetstreamEvent, deliveryRecord: unknown,): Promise<ReferenceLifecycle | undefined> { if (event.kind !== 'commit' || !event.commit || backlinkScopeDids.size === 0) return undefined const { did, commit } = event const key = eventAtUri(did, commit.collection, commit.rkey) const repository = currentDependencies().referenceRepository const snapshot = await repository.load(key) const oldReferences = normalizeReferences(snapshot?.references ?? []) if (commit.operation !== 'delete' && deliveryRecord === undefined) { throw new Error('Non-delete commit has no validated record') } const current = commit.operation === 'delete' ? { references: [], tooComplex: false } : currentReferences(deliveryRecord) // Only a key that may own durable state is worth reading it. const snapshot = mayOwnPriorReferences(key) ? await currentDependencies().referenceRepository.load(key) : undefined const oldReferences = normalizeReferences(snapshot?.references ?? []) const combined = [...new Set([...oldReferences, ...current.references])] return { key, oldReferences,8 unmodified lines lifecycle: ReferenceLifecycle | undefined,): Promise<void> { if (!lifecycle || event.kind !== 'commit' || !event.commit) return const { time_us: timeUs, commit } = event const repository = currentDependencies().referenceRepository if (lifecycle.tooComplex) { // Preserve a monotonic empty tombstone at this event version. A later delete // therefore cannot resurrect stale references from an earlier known state. await repository.save(lifecycle.key, [], timeUs, commit.rev) await writePriorReferences(lifecycle.key, [], timeUs, commit.rev) return } if (commit.operation === 'delete') { // Delete only after all deliveries using the old durable snapshot are enqueued. await repository.delete(lifecycle.key, timeUs, commit.rev) if (!mayOwnPriorReferences(lifecycle.key)) return if (await currentDependencies().referenceRepository.delete(lifecycle.key, timeUs, commit.rev)) { trackPriorReferences(lifecycle.key, false) } } else { await repository.save(lifecycle.key, lifecycle.newReferences, timeUs, commit.rev) await writePriorReferences(lifecycle.key, lifecycle.newReferences, timeUs, commit.rev) }}
async function clearPriorReferenceState(event: JetstreamEvent): Promise<void> { if (event.kind !== 'commit' || !event.commit) return await currentDependencies().referenceRepository.save( await writePriorReferences( eventAtUri(event.did, event.commit.collection, event.commit.rkey), [], event.time_us, event.commit.rev,314 unmodified linesExpand all createJetstreamClient: options.createJetstreamClient ?? ((clientOptions) => new JetstreamClient(clientOptions)), } firehoseStarted = true try { await hydratePriorReferenceKeys() await startRegistry(options.cursors?.[REGISTRY_STREAM]) await startDirect(options.cursors?.[DIRECT_STREAM]) await startBacklink(options.cursors?.[BACKLINK_STREAM]) } catch (error) {120 unmodified linesExpand all directJetstream = null backlinkJetstream = null registryJetstream = null dependencies = null priorReferenceKeys = null stopping = false intakePaused = false lastEventTime = 0 totalEvents = 020 unmodified lines