import type { Main as WhRecord } from '@wispplace/lexicons/types/place/wisp/v2/wh'
import { createLogger } from '@wispplace/observability'
import { config } from '../config'
import type { PriorReferenceWrite, WebhookEntry } from './db'
import * as database from './db'
import * as delivery from './delivery'
import {
buildJetstreamSubscriptionUrl,
JetstreamClient,
type JetstreamEvent,
type JetstreamOptions,
normalizeRelayIdentity,
} from './jetstream'
import {
collectAtUriReferences,
collectRelevantAtUriReferences,
type EventKind,
isDid,
matchWebhooks,
parseAtUri,
scopeDid,
validateWebhookRecord,
} from './matcher'
import {
BACKLINK_CACHE_KEY,
clearRegistryCache,
getCached,
getCacheGeneration,
invalidate,
invalidateMany,
setCachedIfCurrent,
} from './registry'
import { assertSafeWebhookUrlSyntax } from './webhook-url'
const logger = createLogger('webhook-service:firehose')
const WH_COLLECTION = 'place.wisp.v2.wh'
const MAX_RELEVANT_REFERENCES =
typeof database.MAX_BACKLINK_REFERENCES_PER_EVENT === 'number' && database.MAX_BACKLINK_REFERENCES_PER_EVENT > 0
? database.MAX_BACKLINK_REFERENCES_PER_EVENT
: 100
const DIRECT_STREAM = 'direct'
const BACKLINK_STREAM = 'backlink'
const REGISTRY_STREAM = 'registry'
type StreamName = typeof DIRECT_STREAM | typeof BACKLINK_STREAM | typeof REGISTRY_STREAM
/** The executor of the current intake batch; every durable effect joins it. */
type BatchSql = database.SqlExecutor | undefined
/** Runs one batch so its effects and its cursor commit together. */
type IntakeBatchRunner = <T>(work: (sql: BatchSql) => Promise<T>) => Promise<T>
export interface CursorRepository {
load(stream: StreamName, relay: string): Promise<number | undefined>
save(stream: StreamName, cursor: number, relay: string, sql?: BatchSql): Promise<void>
}
interface PriorReferenceSnapshot {
references: readonly string[]
timeUs?: number
rev?: string
}
/** Durable, bounded backlink state. It deliberately stores references, never whole records. */
export interface PriorReferenceRepository {
load(eventAtUri: string, sql?: BatchSql): Promise<PriorReferenceSnapshot | undefined>
save(
eventAtUri: string,
references: readonly string[],
timeUs: number,
rev: string,
sql?: BatchSql,
): Promise<PriorReferenceWrite>
/** True only when this event removed the durable row. */
delete(eventAtUri: string, timeUs: number, rev: string, sql?: BatchSql): Promise<boolean>
/** Bounded hydration of the keys that own durable references, when available. */
keys?(): Promise<{ keys: readonly string[]; complete: boolean }>
}
interface DurableDeliveryEvent {
relay: string
timeUs: number
rev: string
operation: EventKind
did: string
collection: string
rkey: string
cid?: string
record?: unknown
}
type EnqueueWebhookDeliveries = (
entries: readonly WebhookEntry[],
event: DurableDeliveryEvent,
sql?: BatchSql,
) => Promise<{ enqueued: number; deduplicated: number }>
interface FirehoseStartOptions {
/** Test seam. Production uses the durable DB cursor repository. */
cursorRepository?: CursorRepository
/** Test seam. Production requires the durable DB prior-reference repository. */
referenceRepository?: PriorReferenceRepository
enqueueWebhookDeliveries?: EnqueueWebhookDeliveries
/** Test seam. Production writes an idempotent redacted intake quarantine row. */
recordWebhookIntakeQuarantine?: typeof database.recordWebhookIntakeQuarantine
/** Test seam. Production wraps each batch in one database transaction. */
runIntakeBatch?: IntakeBatchRunner
createJetstreamClient?: (options: JetstreamOptions) => JetstreamClient
/** Compatibility seam for tests; each stream remains independent in production. */
cursors?: Partial<Record<StreamName, number>>
/** Capture relay events without invoking handlers until startup reconciliation finishes. */
paused?: boolean
}
interface RuntimeDependencies {
cursorRepository: CursorRepository
referenceRepository: PriorReferenceRepository
enqueueWebhookDeliveries: EnqueueWebhookDeliveries
recordWebhookIntakeQuarantine: typeof database.recordWebhookIntakeQuarantine
runIntakeBatch: IntakeBatchRunner
createJetstreamClient: (options: JetstreamOptions) => JetstreamClient
}
interface WhMutationResult {
deliveryRecord?: WhRecord
selfKey: string
/** Deterministic malformed/unsafe registration: revoke and acknowledge, never replay forever. */
ackableInvalid?: true
}
let lastEventTime = 0
let totalEvents = 0
let totalMatched = 0
let tooComplexBacklinkEvents = 0
let invalidRegistryRecords = 0
let invalidDeliveryInputEvents = 0
let rejectedSubscriptionAdmissions = 0
let registrySnapshotOverflow = false
let firehoseStarted = false
let stopping = false
let reconfigureScheduled = false
let reconfiguring: Promise<void> | null = null
let directRestartCursorHint: number | undefined
let backlinkRestartCursorHint: number | undefined
let forceStreamRestart = false
let refreshRegistryPromise: Promise<void> | null = null
let intakePaused = false
const CURSOR_SAFETY_REWIND_US = 2_000_000
let dependencies: RuntimeDependencies | null = null
let scopeDids = new Set<string>()
let ownerDids = new Set<string>()
let backlinkScopeDids = new Set<string>()
let directWantedDids = new Set<string>()
let directJetstream: JetstreamClient | null = null
let backlinkJetstream: JetstreamClient | null = null
// Always-on, collection-filtered discovery stream. It is the sole mutation path
// for webhook registrations, including an owner's first-ever record.
let registryJetstream: JetstreamClient | null = null
const allClients = new Set<JetstreamClient>()
function safeError(stream: StreamName): void {
logger.warn(`[${stream}] intake event failed; cursor was not advanced`)
}
function sortedDids(values: ReadonlySet<string>): string[] {
return [...values].sort()
}
function eventAtUri(did: string, collection: string, rkey: string): string {
return `at://${did}/${collection}/${rkey}`
}
function normalizeReferences(references: readonly string[]): string[] {
const result = new Set<string>()
for (const reference of references) {
const parsed = parseAtUri(reference)
if (!parsed || !backlinkScopeDids.has(parsed.did)) continue
result.add(
`at://${parsed.did}${parsed.collection ? `/${parsed.collection}` : ''}${parsed.rkey ? `/${parsed.rkey}` : ''}`,
)
if (result.size >= 256) break
}
return [...result]
}
function currentReferences(record: unknown): { references: string[]; tooComplex: boolean } {
const collected = collectRelevantAtUriReferences(
record,
(reference) => {
const parsed = parseAtUri(reference)
return parsed !== null && backlinkScopeDids.has(parsed.did)
},
MAX_RELEVANT_REFERENCES,
)
return { references: normalizeReferences(collected.references), tooComplex: collected.tooComplex }
}
function referencesIntersectActiveScopes(references: readonly string[]): boolean {
for (const reference of references) {
const parsed = parseAtUri(reference)
if (parsed && backlinkScopeDids.has(parsed.did)) return true
}
return false
}
function mergeEntries(direct: readonly WebhookEntry[], backlink: readonly WebhookEntry[]): WebhookEntry[] {
const result = [...direct]
const seen = new Set(result.map((entry) => `${entry.ownerDid}/${entry.rkey}`))
for (const entry of backlink) {
const key = `${entry.ownerDid}/${entry.rkey}`
if (seen.has(key)) continue
seen.add(key)
result.push(entry)
}
return result
}
function entriesForTrackedKeys(keys: Iterable<string> | undefined): WebhookEntry[] {
if (!keys) return []
const entries: WebhookEntry[] = []
for (const key of keys) {
const tracked = trackedWebhooks.get(key)
if (!tracked) continue
entries.push({ ownerDid: tracked.ownerDid, rkey: tracked.rkey, record: tracked.record })
}
return entries
}
function cachedTrackedEntries(key: string, keys: Iterable<string> | undefined): readonly WebhookEntry[] {
const cached = getCached(key)
if (cached !== undefined) return cached
const generation = getCacheGeneration(key)
const entries = entriesForTrackedKeys(keys)
// This is synchronous today, but retain the generation fence so future async
// candidate sources cannot repopulate a mutation-invalidated entry.
if (setCachedIfCurrent(key, entries, generation)) return getCached(key) ?? entries
return entriesForTrackedKeys(keys)
}
async function getWebhooksForEvent(eventDid: string, referenceRecord: unknown): Promise<WebhookEntry[]> {
const direct = cachedTrackedEntries(eventDid, scopeWebhookKeys.get(eventDid))
const backlink = cachedTrackedEntries(BACKLINK_CACHE_KEY, backlinkWebhookKeys)
if (backlink.length === 0 || !referencesIntersectActiveScopes(collectAtUriReferences(referenceRecord))) {
return [...direct]
}
return mergeEntries(direct, backlink)
}
function defaultCursorRepository(): CursorRepository {
const api = database as typeof database & {
loadCursorForStream?: (stream: string, relay: string) => Promise<number | undefined>
saveCursorForStream?: (stream: string, cursor: number, relay: string, sql?: BatchSql) => Promise<void>
saveCursor?:
| ((cursor: number, relay: string) => Promise<void>)
| ((stream: string, cursor: number, relay: string) => Promise<void>)
}
return {
async load(stream, relay) {
if (api.loadCursorForStream) return api.loadCursorForStream(stream, relay)
// Legacy installs have only one cursor. Never reuse it for unrelated streams.
if (stream === DIRECT_STREAM) return database.loadCursor(relay)
throw new Error('Independent durable cursor storage is unavailable')
},
async save(stream, cursor, relay, sql) {
if (api.saveCursorForStream) return api.saveCursorForStream(stream, cursor, relay, sql)
if (api.saveCursor && api.saveCursor.length >= 3) {
return (api.saveCursor as (stream: string, cursor: number, relay: string) => Promise<void>)(
stream,
cursor,
relay,
)
}
if (stream === DIRECT_STREAM) return database.saveCursor(cursor, relay)
throw new Error('Independent durable cursor storage is unavailable')
},
}
}
function defaultReferenceRepository(): PriorReferenceRepository {
const api = database as typeof database & {
loadPriorReferenceIndex?: (eventAtUri: string, sql?: BatchSql) => Promise<PriorReferenceSnapshot | undefined>
savePriorReferenceIndex?: (
eventAtUri: string,
references: readonly string[],
timeUs: number,
rev: string,
sql?: BatchSql,
) => Promise<PriorReferenceWrite>
deletePriorReferenceIndex?: (eventAtUri: string, timeUs: number, rev: string, sql?: BatchSql) => Promise<boolean>
loadPriorReferenceKeys?: () => Promise<{ keys: readonly string[]; complete: boolean }>
}
return {
async load(key, sql) {
if (!api.loadPriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable')
return api.loadPriorReferenceIndex(key, sql)
},
async save(key, references, timeUs, rev, sql) {
if (!api.savePriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable')
return api.savePriorReferenceIndex(key, references, timeUs, rev, sql)
},
async delete(key, timeUs, rev, sql) {
if (!api.deletePriorReferenceIndex) throw new Error('Durable prior-reference storage is unavailable')
return api.deletePriorReferenceIndex(key, timeUs, rev, sql)
},
async keys() {
if (!api.loadPriorReferenceKeys) throw new Error('Durable prior-reference storage is unavailable')
return api.loadPriorReferenceKeys()
},
}
}
function defaultEnqueueWebhookDeliveries(): EnqueueWebhookDeliveries {
const api = delivery as typeof delivery & {
enqueueWebhookDeliveries?: (
entries: readonly WebhookEntry[],
event: DurableDeliveryEvent,
options?: delivery.EnqueueWebhookDeliveriesOptions,
) => Promise<{ enqueued: number; deduplicated: number }>
}
return async (entries, event, sql) => {
if (!api.enqueueWebhookDeliveries) throw new Error('Durable webhook delivery queue is unavailable')
// Outbox rows join the intake batch, so the cursor can never advance past
// a delivery that was rolled back.
const options = sql
? {
enqueueOutbox: (
outboxEvent: Parameters<typeof database.enqueueWebhookDeliveryOutbox>[0],
rows: Parameters<typeof database.enqueueWebhookDeliveryOutbox>[1],
ensureEvent?: boolean,
) => database.enqueueWebhookDeliveryOutbox(outboxEvent, rows, ensureEvent, sql),
}
: undefined
return api.enqueueWebhookDeliveries(entries, event, options)
}
}
function currentDependencies(): RuntimeDependencies {
if (!dependencies) throw new Error('Firehose was not started')
return dependencies
}
/** A small mutex for registry-state snapshots versus live WH mutations. */
class AsyncMutex {
private tail: Promise<void> = Promise.resolve()
async run<T>(operation: () => Promise<T>): Promise<T> {
const previous = this.tail
let release: (() => void) | undefined
this.tail = new Promise<void>((resolve) => {
release = resolve
})
await previous.catch(() => undefined)
try {
return await operation()
} finally {
release?.()
}
}
}
const registryStateMutex = new AsyncMutex()
class KeyedSerialExecutor {
private readonly tails = new Map<string, Promise<void>>()
constructor(private readonly maxKeys: number) {}
async run(key: string, task: () => Promise<void>): Promise<void> {
const previous = this.tails.get(key)
if (!previous && this.tails.size >= this.maxKeys) throw new Error('Intake record-key limit reached')
const current = previous ? previous.then(task) : task()
this.tails.set(key, current)
try {
await current
} finally {
if (this.tails.get(key) === current) this.tails.delete(key)
}
}
get size(): number {
return this.tails.size
}
}
const recordExecutor = new KeyedSerialExecutor(config.intakeRecordKeyMax)
/**
* Rebuild tracked owner/scope DID sets. Owner DIDs are always subscribed so a
* webhook owned by A and scoped to B still receives A's future update/delete.
*/
interface TrackedWebhook {
readonly ownerDid: string
readonly rkey: string
readonly scopeDid: string
readonly backlinks: boolean
readonly record: WhRecord
}
const trackedWebhooks = new Map<string, TrackedWebhook>()
const scopeWebhookKeys = new Map<string, Set<string>>()
const backlinkWebhookKeys = new Set<string>()
const ownerRefCounts = new Map<string, number>()
const scopeRefCounts = new Map<string, number>()
const backlinkScopeRefCounts = new Map<string, number>()
const DIRECT_SUBSCRIPTION_PARAMETER_BYTES = Buffer.byteLength('&wantedDids=')
const DIRECT_SUBSCRIPTION_CURSOR_RESERVE_BYTES = Buffer.byteLength(`&cursor=${Number.MAX_SAFE_INTEGER}`)
let directSubscriptionUrlBytes = Buffer.byteLength(new URL(config.jetstreamUrl).toString())
function wantedDidBytes(did: string): number {
return DIRECT_SUBSCRIPTION_PARAMETER_BYTES + Buffer.byteLength(encodeURIComponent(did))
}
function projectedCount(
counts: ReadonlyMap<string, number>,
did: string,
prior: TrackedWebhook | undefined,
next: TrackedWebhook | undefined,
field: keyof TrackedWebhook,
): number {
let value = counts.get(did) ?? 0
if (prior?.[field] === did) value--
if (next?.[field] === did) value++
return value
}
function projectedBooleanCount(
counts: ReadonlyMap<string, number>,
did: string,
prior: TrackedWebhook | undefined,
next: TrackedWebhook | undefined,
): number {
let value = counts.get(did) ?? 0
if (prior?.backlinks && prior.scopeDid === did) value--
if (next?.backlinks && next.scopeDid === did) value++
return value
}
function canAdmitTrackedWebhook(key: string, next: TrackedWebhook): boolean {
const prior = trackedWebhooks.get(key)
if (!prior && trackedWebhooks.size >= config.registryActiveSubscriptionsMax) return false
if (!prior && (ownerRefCounts.get(next.ownerDid) ?? 0) >= config.registryOwnerActiveRecordsMax) return false
const affected = affectedDids(prior, next)
let projectedScopeDids = scopeDids.size
let projectedBacklinkScopes = backlinkScopeDids.size
let projectedDirectDids = directWantedDids.size
let projectedUrlBytes = directSubscriptionUrlBytes
for (const did of affected) {
const oldScope = (scopeRefCounts.get(did) ?? 0) > 0
const newScope = projectedCount(scopeRefCounts, did, prior, next, 'scopeDid') > 0
if (!oldScope && newScope) projectedScopeDids++
if (oldScope && !newScope) projectedScopeDids--
const oldBacklink = (backlinkScopeRefCounts.get(did) ?? 0) > 0
const newBacklink = projectedBooleanCount(backlinkScopeRefCounts, did, prior, next) > 0
if (!oldBacklink && newBacklink) projectedBacklinkScopes++
if (oldBacklink && !newBacklink) projectedBacklinkScopes--
const oldDirect = directWantedDids.has(did)
const newDirect =
projectedCount(ownerRefCounts, did, prior, next, 'ownerDid') +
projectedCount(scopeRefCounts, did, prior, next, 'scopeDid') >
0
if (!oldDirect && newDirect) {
projectedDirectDids++
projectedUrlBytes += wantedDidBytes(did)
}
if (oldDirect && !newDirect) {
projectedDirectDids--
projectedUrlBytes -= wantedDidBytes(did)
}
}
return (
projectedScopeDids <= config.registryDirectScopeDidsMax &&
projectedBacklinkScopes <= config.registryBacklinkScopeDidsMax &&
projectedDirectDids <= config.registryDirectScopeDidsMax &&
projectedUrlBytes + DIRECT_SUBSCRIPTION_CURSOR_RESERVE_BYTES <= config.registrySubscriptionUrlBytesMax
)
}
function canAdmitSubscription(ownerDid: string, rkey: string, record: WhRecord): boolean {
const next = trackedWebhook(ownerDid, rkey, record)
return next !== undefined && canAdmitTrackedWebhook(`${ownerDid}/${rkey}`, next)
}
function adjustCount(counts: Map<string, number>, values: Set<string>, value: string, delta: 1 | -1): void {
const next = (counts.get(value) ?? 0) + delta
if (next <= 0) {
counts.delete(value)
values.delete(value)
} else {
counts.set(value, next)
values.add(value)
}
}
function refreshDirectDid(did: string): void {
const wasTracked = directWantedDids.has(did)
const shouldTrack = (ownerRefCounts.get(did) ?? 0) + (scopeRefCounts.get(did) ?? 0) > 0
if (!wasTracked && shouldTrack) {
directWantedDids.add(did)
directSubscriptionUrlBytes += wantedDidBytes(did)
} else if (wasTracked && !shouldTrack) {
directWantedDids.delete(did)
directSubscriptionUrlBytes -= wantedDidBytes(did)
}
}
function trackedWebhook(ownerDid: string, rkey: string, record: unknown): TrackedWebhook | undefined {
if (!isDid(ownerDid)) {
invalidRegistryRecords++
return undefined
}
const validated = validateWebhookRecord(record)
if (!validated) {
invalidRegistryRecords++
return undefined
}
// Disabled records remain durable state/history, but have no active scope or
// delivery effect and therefore consume no admission/index capacity.
if (validated.enabled === false) return undefined
const scope = parseAtUri(validated.scope.aturi)
if (!scope) return undefined
return { ownerDid, rkey, scopeDid: scope.did, backlinks: validated.scope.backlinks === true, record: validated }
}
function addTrackedWebhook(key: string, entry: TrackedWebhook): void {
trackedWebhooks.set(key, entry)
let scoped = scopeWebhookKeys.get(entry.scopeDid)
if (!scoped) {
scoped = new Set<string>()
scopeWebhookKeys.set(entry.scopeDid, scoped)
}
scoped.add(key)
if (entry.backlinks) backlinkWebhookKeys.add(key)
adjustCount(ownerRefCounts, ownerDids, entry.ownerDid, 1)
adjustCount(scopeRefCounts, scopeDids, entry.scopeDid, 1)
if (entry.backlinks) adjustCount(backlinkScopeRefCounts, backlinkScopeDids, entry.scopeDid, 1)
refreshDirectDid(entry.ownerDid)
refreshDirectDid(entry.scopeDid)
}
function removeTrackedWebhook(key: string): TrackedWebhook | undefined {
const entry = trackedWebhooks.get(key)
if (!entry) return undefined
trackedWebhooks.delete(key)
const scoped = scopeWebhookKeys.get(entry.scopeDid)
scoped?.delete(key)
if (scoped?.size === 0) scopeWebhookKeys.delete(entry.scopeDid)
if (entry.backlinks) backlinkWebhookKeys.delete(key)
adjustCount(ownerRefCounts, ownerDids, entry.ownerDid, -1)
adjustCount(scopeRefCounts, scopeDids, entry.scopeDid, -1)
if (entry.backlinks) adjustCount(backlinkScopeRefCounts, backlinkScopeDids, entry.scopeDid, -1)
refreshDirectDid(entry.ownerDid)
refreshDirectDid(entry.scopeDid)
return entry
}
function affectedDids(...entries: Array<TrackedWebhook | undefined>): Set<string> {
const result = new Set<string>()
for (const entry of entries) {
if (!entry) continue
result.add(entry.ownerDid)
result.add(entry.scopeDid)
}
return result
}
function applyTrackedWebhookMutation(
ownerDid: string,
rkey: string,
newRecord: WhRecord | undefined,
registrationCursorHint?: number,
schedule = true,
): void {
const key = `${ownerDid}/${rkey}`
const prior = trackedWebhooks.get(key)
const next = newRecord ? trackedWebhook(ownerDid, rkey, newRecord) : undefined
const affected = affectedDids(prior, next)
const previousDirect = new Map([...affected].map((did) => [did, directWantedDids.has(did)]))
const previousBacklinks = new Map([...affected].map((did) => [did, backlinkScopeDids.has(did)]))
removeTrackedWebhook(key)
if (next) addTrackedWebhook(key, next)
let changed = false
for (const did of affected) {
if (previousDirect.get(did) !== directWantedDids.has(did)) changed = true
if (previousBacklinks.get(did) !== backlinkScopeDids.has(did)) changed = true
}
if (schedule && changed && firehoseStarted && !stopping) scheduleReconfigure(registrationCursorHint)
}
function sameDidSet(left: ReadonlySet<string>, right: ReadonlySet<string>): boolean {
if (left.size !== right.size) return false
for (const did of left) {
if (!right.has(did)) return false
}
return true
}
/** Rebuild the bounded in-memory subscription index from reconciliation-filtered DB rows. */
export function initScopeDids(webhooks: ReadonlyArray<{ did: string; rkey?: string; record: unknown }>): void {
clearRegistryCache()
trackedWebhooks.clear()
scopeWebhookKeys.clear()
backlinkWebhookKeys.clear()
ownerRefCounts.clear()
scopeRefCounts.clear()
backlinkScopeRefCounts.clear()
scopeDids = new Set<string>()
ownerDids = new Set<string>()
backlinkScopeDids = new Set<string>()
directWantedDids = new Set<string>()
directSubscriptionUrlBytes = Buffer.byteLength(new URL(config.jetstreamUrl).toString())
const ordered = [...webhooks].sort((left, right) =>
`${left.did}/${left.rkey ?? ''}`.localeCompare(`${right.did}/${right.rkey ?? ''}`),
)
for (const webhook of ordered) {
if (!webhook.rkey || !isDid(webhook.did)) continue
const entry = trackedWebhook(webhook.did, webhook.rkey, webhook.record)
if (!entry) continue
const key = `${webhook.did}/${webhook.rkey}`
if (!canAdmitTrackedWebhook(key, entry)) {
rejectedSubscriptionAdmissions++
continue
}
addTrackedWebhook(key, entry)
}
}
async function findExistingWebhook(ownerDid: string, rkey: string): Promise<WebhookEntry | undefined> {
// This is a primary-key lookup, not a full registry scan: the global registry
// consumer must have strictly bounded work for each WH mutation.
return database.getWebhookRecord(ownerDid, rkey)
}
async function loadTrackableWebhooks(): Promise<database.ActiveWebhookLoadResult> {
return database.loadActiveWebhooks()
}
function replaceTrackedOwner(
ownerDid: string,
rows: readonly { did: string; rkey: string; record: WhRecord }[],
registrationCursorHint?: number,
): void {
clearRegistryCache()
// A reconciliation swap may remove and immediately re-add the same DID. Batch
// those mutations so a no-op owner refresh cannot churn live subscriptions.
const priorDirectDids = new Set(directWantedDids)
const priorBacklinkDids = new Set(backlinkScopeDids)
const priorKeys = [...trackedWebhooks.entries()]
.filter(([, entry]) => entry.ownerDid === ownerDid)
.map(([key]) => key)
.sort()
for (const key of priorKeys) {
const rkey = key.slice(ownerDid.length + 1)
applyTrackedWebhookMutation(ownerDid, rkey, undefined, undefined, false)
}
const ordered = rows
.filter((row) => row.did === ownerDid)
.slice()
.sort((left, right) => left.rkey.localeCompare(right.rkey))
for (const row of ordered) applyTrackedWebhookMutation(ownerDid, row.rkey, row.record, undefined, false)
if (
firehoseStarted &&
!stopping &&
(!sameDidSet(priorDirectDids, directWantedDids) || !sameDidSet(priorBacklinkDids, backlinkScopeDids))
) {
// Replay from the registry stream's acknowledged boundary. This closes the
// interval between a PDS snapshot becoming active and its direct stream swap.
scheduleReconfigure(registrationCursorHint)
}
}
/** Incremental, bounded owner swap used for reconciliation transitions. */
export function refreshRegistryOwnerFromDatabase(ownerDid: string): Promise<void> {
return registryStateMutex.run(async () => {
const snapshot = await database.loadActiveWebhooksForOwner(ownerDid)
registrySnapshotOverflow ||= snapshot.overflow
replaceTrackedOwner(ownerDid, snapshot.rows, registryJetstream?.cursor)
})
}
async function reloadRegistry(
oldRecord: WhRecord | undefined,
newRecord: WhRecord | undefined,
ownerDid: string,
rkey: string,
registrationCursorHint?: number,
): Promise<void> {
const keys = new Set<string>([ownerDid])
const oldScope = oldRecord ? scopeDid(oldRecord.scope.aturi) : undefined
const newScope = newRecord ? scopeDid(newRecord.scope.aturi) : undefined
if (oldScope) keys.add(oldScope)
if (newScope) keys.add(newScope)
invalidateMany(keys)
invalidate(BACKLINK_CACHE_KEY)
// Apply just this mutation to the in-memory sets. The global registry stream
// never performs a load-all scan per event.
applyTrackedWebhookMutation(ownerDid, rkey, newRecord, registrationCursorHint)
}
/**
* Coalesced reconciliation-transition hook. It clears potentially stale cached
* candidates and reloads only status-eligible, sanitized records. A force restart
* replays each stream from its durable cursor; no callback can restart after stop.
*/
export function refreshRegistryFromDatabase(): Promise<void> {
if (refreshRegistryPromise) return refreshRegistryPromise
refreshRegistryPromise = registryStateMutex
.run(async () => {
// Holding the mutex across this bounded status-filtered snapshot prevents a
// live WH mutation from being overwritten by an older DB result. The
// synchronous init swap invalidates caches only once the snapshot arrived.
const snapshot = await loadTrackableWebhooks()
registrySnapshotOverflow = snapshot.overflow
initScopeDids(snapshot.rows)
if (firehoseStarted && !stopping && !intakePaused) scheduleReconfigure(undefined, true)
})
.finally(() => {
refreshRegistryPromise = null
})
return refreshRegistryPromise
}
async function disableInvalidWebhook(
ownerDid: string,
rkey: string,
oldRecord: WhRecord | undefined,
source: { revision: string; cid?: string; timeUs: number },
): Promise<void> {
const applied = await database.deleteWebhookRecord(ownerDid, rkey, source)
if (applied && oldRecord) await reloadRegistry(oldRecord, undefined, ownerDid, rkey, source.timeUs)
}
async function handleWhRecord(
op: EventKind,
ownerDid: string,
rkey: string,
record: unknown,
source: { revision: string; cid?: string; timeUs: number },
): Promise<WhMutationResult> {
const existing = await findExistingWebhook(ownerDid, rkey)
const oldRecord = existing?.record
const selfKey = `${ownerDid}/${rkey}`
if (op === 'delete') {
// Delete is source-fenced before invalidating caches or acknowledging the cursor.
const applied = await database.deleteWebhookRecord(ownerDid, rkey, source)
if (applied && oldRecord) await reloadRegistry(oldRecord, undefined, ownerDid, rkey, source.timeUs)
return { selfKey }
}
const validated = validateWebhookRecord(record)
if (!validated) {
invalidRegistryRecords++
// An invalid update must revoke an older valid endpoint rather than leave it
// active indefinitely. The malformed record itself is never stored or used.
await disableInvalidWebhook(ownerDid, rkey, oldRecord, source)
return { selfKey, ackableInvalid: true }
}
try {
assertSafeWebhookUrlSyntax(validated.url)
} catch {
invalidRegistryRecords++
await disableInvalidWebhook(ownerDid, rkey, oldRecord, source)
return { selfKey, ackableInvalid: true }
}
if (validated.enabled !== false && !canAdmitSubscription(ownerDid, rkey, validated)) {
rejectedSubscriptionAdmissions++
// Do not retain an old scope when a valid update cannot be admitted: that
// would keep delivering to a stale endpoint forever. A new over-cap record
// is omitted; an existing one is durably revoked before acknowledgement.
if (oldRecord) await disableInvalidWebhook(ownerDid, rkey, oldRecord, source)
await database.recordWebhookIntakeQuarantine({
relay: normalizeRelayIdentity(config.jetstreamUrl),
timeUs: source.timeUs,
rev: source.revision,
did: ownerDid,
collection: WH_COLLECTION,
rkey,
reason: 'invalid_subscription',
})
logger.warn('[registry] active subscription admission limit reached; event quarantined')
return { selfKey, ackableInvalid: true }
}
const changed = await database.upsertWebhookRecord(ownerDid, rkey, validated, source)
if (changed) await reloadRegistry(oldRecord, validated, ownerDid, rkey, source.timeUs)
return { deliveryRecord: validated, selfKey }
}
async function enqueueMatched(
event: JetstreamEvent,
backlinkRecord: unknown,
deliveryRecord: unknown,
excludedWebhookKey?: string,
sql?: BatchSql,
): Promise<void> {
if (event.kind !== 'commit' || !event.commit) return
const { did, time_us: timeUs, commit } = event
const candidates = await getWebhooksForEvent(did, backlinkRecord)
let matched = matchWebhooks(
candidates,
did,
commit.collection,
commit.rkey,
commit.operation,
deliveryRecord,
backlinkRecord,
)
// A subscription never receives the mutation of its own WH record. Other
// subscriptions may intentionally watch place.wisp.v2.wh and still receive it.
if (excludedWebhookKey) {
matched = matched.filter((entry) => `${entry.ownerDid}/${entry.rkey}` !== excludedWebhookKey)
}
if (matched.length === 0) return
const durableEvent: DurableDeliveryEvent = {
relay: normalizeRelayIdentity(config.jetstreamUrl),
timeUs,
rev: commit.rev,
operation: commit.operation,
did,
collection: commit.collection,
rkey: commit.rkey,
...(commit.cid ? { cid: commit.cid } : {}),
...(deliveryRecord === undefined ? {} : { record: deliveryRecord }),
}
try {
const result = await currentDependencies().enqueueWebhookDeliveries(matched, durableEvent, sql)
totalMatched += result.enqueued + result.deduplicated
} catch (error) {
if (error instanceof delivery.WebhookDeliveryInputError) {
// Deterministic untrusted payload/subscription input is ackable. Store a
// bounded redacted identity first; a failure to persist that row remains
// retryable and deliberately keeps the cursor behind.
await currentDependencies().recordWebhookIntakeQuarantine(
{
relay: normalizeRelayIdentity(config.jetstreamUrl),
timeUs,
rev: commit.rev,
did,
collection: commit.collection,
rkey: commit.rkey,
reason: error.kind,
},
sql,
)
invalidDeliveryInputEvents++
logger.warn('[delivery] invalid durable input; event quarantined')
return
}
throw error
}
}
interface ReferenceLifecycle {
readonly key: string
readonly oldReferences: readonly string[]
readonly newReferences: readonly string[]
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,
sql?: BatchSql,
): Promise<void> {
if (references.length === 0 && !mayOwnPriorReferences(key)) return
const owned =
PRIOR_STATE_AFTER_WRITE[await currentDependencies().referenceRepository.save(key, references, timeUs, rev, sql)]
if (owned !== undefined) trackPriorReferences(key, owned)
}
async function loadReferenceLifecycle(
event: JetstreamEvent,
deliveryRecord: unknown,
sql?: BatchSql,
): 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)
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, sql)
: undefined
const oldReferences = normalizeReferences(snapshot?.references ?? [])
const combined = [...new Set([...oldReferences, ...current.references])]
return {
key,
oldReferences,
newReferences: current.references,
matchRecord: current.tooComplex ? [] : combined,
tooComplex: current.tooComplex,
}
}
async function commitReferenceLifecycle(
event: JetstreamEvent,
lifecycle: ReferenceLifecycle | undefined,
sql?: BatchSql,
): Promise<void> {
if (!lifecycle || event.kind !== 'commit' || !event.commit) return
const { time_us: timeUs, commit } = event
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 writePriorReferences(lifecycle.key, [], timeUs, commit.rev, sql)
return
}
if (commit.operation === 'delete') {
// Delete only after all deliveries using the old durable snapshot are enqueued.
if (!mayOwnPriorReferences(lifecycle.key)) return
if (await currentDependencies().referenceRepository.delete(lifecycle.key, timeUs, commit.rev, sql)) {
trackPriorReferences(lifecycle.key, false)
}
} else {
await writePriorReferences(lifecycle.key, lifecycle.newReferences, timeUs, commit.rev, sql)
}
}
async function clearPriorReferenceState(event: JetstreamEvent, sql?: BatchSql): Promise<void> {
if (event.kind !== 'commit' || !event.commit) return
await writePriorReferences(
eventAtUri(event.did, event.commit.collection, event.commit.rkey),
[],
event.time_us,
event.commit.rev,
sql,
)
}
async function processCommit(stream: StreamName, event: JetstreamEvent, sql?: BatchSql): Promise<void> {
if (event.kind !== 'commit' || !event.commit) return
const { did, commit } = event
if (commit.operation !== 'create' && commit.operation !== 'update' && commit.operation !== 'delete') return
const isWebhookRegistration = commit.collection === WH_COLLECTION
// The dedicated registry stream is the only mutator for WH records. Direct
// and backlink consumers still acknowledge their duplicate WH copies, while
// registry delivers the WH event to legitimate non-self subscriptions.
if (isWebhookRegistration && stream !== REGISTRY_STREAM) return
if (!isWebhookRegistration && stream === REGISTRY_STREAM) return
// The backlink consumer sees the full relay. Directly subscribed owners/scopes
// are handled by the direct stream so identical events do not race each other.
if (stream === BACKLINK_STREAM && directWantedDids.has(did)) return
if (stream === BACKLINK_STREAM && backlinkScopeDids.size === 0) return
let deliveryRecord = commit.record
let selfKey: string | undefined
let ackableInvalid = false
if (isWebhookRegistration) {
const mutation = await registryStateMutex.run(() =>
handleWhRecord(commit.operation, did, commit.rkey, commit.record, {
revision: commit.rev,
timeUs: event.time_us,
...(commit.cid ? { cid: commit.cid } : {}),
}),
)
deliveryRecord = mutation.deliveryRecord
selfKey = mutation.selfKey
ackableInvalid = mutation.ackableInvalid === true
}
if (ackableInvalid) {
await clearPriorReferenceState(event, sql)
totalEvents++
return
}
const lifecycle = await loadReferenceLifecycle(event, deliveryRecord, sql)
if (lifecycle?.tooComplex) {
tooComplexBacklinkEvents++
logger.warn('[backlink] record references too complex; backlink matching skipped')
// Passing an empty backlink snapshot preserves direct matching while ensuring
// no truncated backlink result is ever partially delivered or indexed.
await enqueueMatched(event, [], deliveryRecord, selfKey, sql)
await commitReferenceLifecycle(event, lifecycle, sql)
totalEvents++
return
}
// A full-backlink relay event with neither prior nor current relevant reference
// cannot match any backlink subscription. Direct candidates still matter only
// when this event came from a direct subscription, which was handled above.
if (
stream === BACKLINK_STREAM &&
lifecycle &&
lifecycle.oldReferences.length === 0 &&
lifecycle.newReferences.length === 0
) {
await commitReferenceLifecycle(event, lifecycle, sql)
return
}
await enqueueMatched(event, lifecycle?.matchRecord ?? deliveryRecord, deliveryRecord, selfKey, sql)
await commitReferenceLifecycle(event, lifecycle, sql)
totalEvents++
}
/**
* One relay batch is one durable transaction: deliveries, backlink state, and
* the stream cursor commit together. A crash therefore replays the batch rather
* than resuming past effects the outbox never received.
*
* Registry record mutations keep their own short transactions. They are rare
* and monotonic by source revision, and they take table-wide advisory locks
* that do not belong inside a relay-paced batch.
*/
async function handleStreamEvents(stream: StreamName, events: readonly JetstreamEvent[]): Promise<void> {
const last = events[events.length - 1]
if (!last) return
lastEventTime = Date.now()
const relay = normalizeRelayIdentity(config.jetstreamUrl)
const dependencies = currentDependencies()
try {
await dependencies.runIntakeBatch(async (sql) => {
for (const event of events) {
if (event.kind !== 'commit' || !event.commit) continue
const key = `${event.did}/${event.commit.collection}/${event.commit.rkey}`
await recordExecutor.run(key, () => processCommit(stream, event, sql))
}
await dependencies.cursorRepository.save(stream, last.time_us, relay, sql)
})
} catch {
safeError(stream)
throw new Error('Intake event processing failed')
}
}
function createClient(
stream: StreamName,
cursor: number | undefined,
wantedDids?: readonly string[],
wantedCollections?: readonly string[],
): JetstreamClient {
const client = currentDependencies().createJetstreamClient({
url: config.jetstreamUrl,
...(wantedDids ? { wantedDids } : {}),
...(wantedCollections ? { wantedCollections } : {}),
cursor,
maxQueue: config.intakeQueueMax,
batchMax: config.intakeBatchMax,
maxEventBytes: config.intakeEventMaxBytes,
reconnectMinMs: config.jetstreamReconnectMinMs,
reconnectMaxMs: config.jetstreamReconnectMaxMs,
reconnectMaxExponent: config.jetstreamReconnectMaxExponent,
onEvents: (events) => handleStreamEvents(stream, events),
onConnect: () => logger.info(`[${stream}] Jetstream connected`),
onDisconnect: () => logger.warn(`[${stream}] Jetstream disconnected`),
onError: () => safeError(stream),
})
if (intakePaused) client.pause()
allClients.add(client)
return client
}
function safelyRewoundCursor(cursor: number | undefined): number | undefined {
if (cursor === undefined || !Number.isSafeInteger(cursor) || cursor < 0) return undefined
return Math.max(0, cursor - CURSOR_SAFETY_REWIND_US)
}
function earliestCursor(...cursors: Array<number | undefined>): number | undefined {
let earliest: number | undefined
for (const cursor of cursors) {
if (cursor === undefined) continue
earliest = earliest === undefined ? cursor : Math.min(earliest, cursor)
}
return earliest
}
async function streamCursor(
stream: StreamName,
initialCursor?: number,
registrationHint?: number,
): Promise<number | undefined> {
const durable =
initialCursor ??
(await currentDependencies().cursorRepository.load(stream, normalizeRelayIdentity(config.jetstreamUrl)))
return earliestCursor(durable, safelyRewoundCursor(registrationHint))
}
interface StreamPlan {
readonly cursor: number | undefined
readonly wantedDids?: readonly string[]
readonly wantedCollections?: readonly string[]
}
function assertPlanUrl(plan: StreamPlan): void {
buildJetstreamSubscriptionUrl(
{
url: config.jetstreamUrl,
...(plan.wantedDids ? { wantedDids: plan.wantedDids } : {}),
...(plan.wantedCollections ? { wantedCollections: plan.wantedCollections } : {}),
},
plan.cursor,
)
}
async function planRegistry(initialCursor?: number): Promise<StreamPlan> {
const durable =
initialCursor ??
(await currentDependencies().cursorRepository.load(REGISTRY_STREAM, normalizeRelayIdentity(config.jetstreamUrl)))
// A fresh registry cursor begins at zero. Jetstream clamps it to retained
// history, avoiding any correctness decision based on local wall-clock skew.
const plan: StreamPlan = { cursor: durable ?? 0, wantedCollections: [WH_COLLECTION] }
assertPlanUrl(plan)
return plan
}
async function planDirect(initialCursor?: number, registrationHint?: number): Promise<StreamPlan | undefined> {
if (directWantedDids.size === 0) return undefined
const plan: StreamPlan = {
cursor: await streamCursor(DIRECT_STREAM, initialCursor, registrationHint),
wantedDids: sortedDids(directWantedDids),
}
assertPlanUrl(plan)
return plan
}
async function planBacklink(initialCursor?: number, registrationHint?: number): Promise<StreamPlan | undefined> {
if (backlinkScopeDids.size === 0) return undefined
const plan: StreamPlan = { cursor: await streamCursor(BACKLINK_STREAM, initialCursor, registrationHint) }
assertPlanUrl(plan)
return plan
}
function startPlanned(stream: StreamName, plan: StreamPlan): JetstreamClient {
const client = createClient(stream, plan.cursor, plan.wantedDids, plan.wantedCollections)
client.start()
return client
}
async function startRegistry(initialCursor?: number): Promise<void> {
if (!firehoseStarted || stopping || registryJetstream) return
registryJetstream = startPlanned(REGISTRY_STREAM, await planRegistry(initialCursor))
}
async function startDirect(initialCursor?: number, registrationHint?: number): Promise<void> {
if (!firehoseStarted || stopping || directJetstream) return
const plan = await planDirect(initialCursor, registrationHint)
if (plan) directJetstream = startPlanned(DIRECT_STREAM, plan)
}
async function startBacklink(initialCursor?: number, registrationHint?: number): Promise<void> {
if (!firehoseStarted || stopping || backlinkJetstream) return
const plan = await planBacklink(initialCursor, registrationHint)
if (plan) backlinkJetstream = startPlanned(BACKLINK_STREAM, plan)
}
async function retireClient(client: JetstreamClient): Promise<void> {
client.stopAccepting()
await client.drain()
client.destroy()
allClients.delete(client)
}
function scheduleReconfigure(registrationCursorHint?: number, force = false): void {
const rewound = safelyRewoundCursor(registrationCursorHint)
if (rewound !== undefined) {
directRestartCursorHint = earliestCursor(directRestartCursorHint, rewound)
backlinkRestartCursorHint = earliestCursor(backlinkRestartCursorHint, rewound)
}
if (force) forceStreamRestart = true
if (reconfigureScheduled || stopping || !firehoseStarted) return
reconfigureScheduled = true
queueMicrotask(() => {
reconfigureScheduled = false
if (stopping || !firehoseStarted) return
if (!reconfiguring) {
reconfiguring = reconfigure()
.catch(() => {
// Plans are validated before retiring active clients. Keep the old
// subscription alive and avoid exposing configuration details.
logger.warn('[registry] subscription reconfiguration was rejected')
})
.finally(() => {
reconfiguring = null
// Hints arriving while a prior reconfiguration drained are serviced
// afterwards, still from durable/rewound cursors.
if (
firehoseStarted &&
!stopping &&
(directRestartCursorHint !== undefined || backlinkRestartCursorHint !== undefined || forceStreamRestart)
) {
scheduleReconfigure()
}
})
}
})
}
async function reconfigure(): Promise<void> {
if (!firehoseStarted || stopping) return
const force = forceStreamRestart
forceStreamRestart = false
const directHint = directRestartCursorHint
const backlinkHint = backlinkRestartCursorHint
directRestartCursorHint = undefined
backlinkRestartCursorHint = undefined
// Validate/load every replacement before touching an active subscription. A
// config/admission error leaves the old direct stream serving safely.
const nextRegistry = force && registryJetstream ? await planRegistry() : undefined
const nextDirect = await planDirect(undefined, directHint)
const nextBacklink =
backlinkScopeDids.size > 0 && (force || !backlinkJetstream)
? await planBacklink(undefined, backlinkHint)
: undefined
if (force && registryJetstream) {
const oldRegistry = registryJetstream
registryJetstream = null
await retireClient(oldRegistry)
if (!stopping && nextRegistry) registryJetstream = startPlanned(REGISTRY_STREAM, nextRegistry)
}
const oldDirect = directJetstream
if (oldDirect) {
directJetstream = null
await retireClient(oldDirect)
}
if (!stopping && nextDirect) directJetstream = startPlanned(DIRECT_STREAM, nextDirect)
if (backlinkScopeDids.size === 0) {
const oldBacklink = backlinkJetstream
backlinkJetstream = null
if (oldBacklink) await retireClient(oldBacklink)
} else if (force && backlinkJetstream) {
const oldBacklink = backlinkJetstream
backlinkJetstream = null
await retireClient(oldBacklink)
if (!stopping && nextBacklink) backlinkJetstream = startPlanned(BACKLINK_STREAM, nextBacklink)
} else if (!backlinkJetstream && !stopping && nextBacklink) {
backlinkJetstream = startPlanned(BACKLINK_STREAM, nextBacklink)
}
}
export async function startFirehose(options: FirehoseStartOptions = {}): Promise<void> {
if (firehoseStarted) return
stopping = false
intakePaused = options.paused === true
dependencies = {
cursorRepository: options.cursorRepository ?? defaultCursorRepository(),
referenceRepository: options.referenceRepository ?? defaultReferenceRepository(),
enqueueWebhookDeliveries: options.enqueueWebhookDeliveries ?? defaultEnqueueWebhookDeliveries(),
recordWebhookIntakeQuarantine: options.recordWebhookIntakeQuarantine ?? database.recordWebhookIntakeQuarantine,
runIntakeBatch: options.runIntakeBatch ?? database.runIntakeBatch,
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) {
firehoseStarted = false
stopping = true
for (const client of allClients) client.destroy()
allClients.clear()
directJetstream = null
backlinkJetstream = null
registryJetstream = null
throw error
}
}
/** Stop accepting new socket messages. Call drainFirehose before closing the DB. */
export function stopFirehose(): void {
if (!firehoseStarted && stopping) return
firehoseStarted = false
stopping = true
intakePaused = false
reconfigureScheduled = false
for (const client of allClients) client.stopAccepting()
}
export async function drainFirehose(timeoutMs = config.shutdownTimeoutMs): Promise<boolean> {
const clients = [...allClients]
// A subscription reconfiguration can be awaiting a cursor query or retiring
// a previous client while shutdown begins. It is intake work too: do not let
// callers close DB/Redis until it settles or this bounded drain expires.
const pendingReconfigure = reconfiguring
let timer: ReturnType<typeof setTimeout> | undefined
try {
const completed = await Promise.race([
Promise.all([
...clients.map((client) => client.drain()),
...(pendingReconfigure ? [pendingReconfigure] : []),
]).then(() => true),
new Promise<boolean>((resolve) => {
timer = setTimeout(() => resolve(false), timeoutMs)
}),
])
return completed
} finally {
if (timer) clearTimeout(timer)
for (const client of clients) {
client.destroy()
allClients.delete(client)
}
directJetstream = null
backlinkJetstream = null
registryJetstream = null
dependencies = null
}
}
export function getFirehoseHealth() {
const directRequired = directWantedDids.size > 0
const backlinkRequired = backlinkScopeDids.size > 0
const directConnected = !directRequired || directJetstream?.isConnected === true
const backlinkConnected = !backlinkRequired || backlinkJetstream?.isConnected === true
const registryConnected = registryJetstream?.isConnected === true
const connected = registryConnected && directConnected && backlinkConnected
// Readiness follows durable progress, not socket state. A stream draining its
// bounded queue after backpressure is serving; one that acknowledges nothing
// is not. Both are reported so the socket view stays visible.
const directProgressing = !directRequired || directJetstream?.isProgressing === true
const backlinkProgressing = !backlinkRequired || backlinkJetstream?.isProgressing === true
const registryProgressing = registryJetstream?.isProgressing === true
return {
connected,
directConnected,
backlinkConnected,
registryConnected,
directProgressing,
backlinkProgressing,
registryProgressing,
cursorAgeMs: {
direct: directJetstream?.cursorAgeMs,
backlink: backlinkJetstream?.cursorAgeMs,
registry: registryJetstream?.cursorAgeMs,
},
started: firehoseStarted,
paused: intakePaused,
queued: (directJetstream?.queued ?? 0) + (backlinkJetstream?.queued ?? 0) + (registryJetstream?.queued ?? 0),
activeRecordKeys: recordExecutor.size,
tooComplexBacklinkEvents,
invalidRegistryRecords,
invalidDeliveryInputEvents,
rejectedSubscriptionAdmissions,
registrySnapshotOverflow,
admissionLimits: {
activeSubscriptions: config.registryActiveSubscriptionsMax,
wantedDids: config.registryDirectScopeDidsMax,
subscriptionUrlBytes: config.registrySubscriptionUrlBytesMax,
},
lastEventTime: lastEventTime || undefined,
timeSinceLastEvent: lastEventTime ? Date.now() - lastEventTime : undefined,
streamFailures: {
direct: {
quarantined: directJetstream?.isQuarantined ?? false,
protocolFailures: directJetstream?.protocolFailureCount ?? 0,
failureKind: directJetstream?.failureKind,
lastProgressAt: directJetstream?.lastProgressTime,
},
backlink: {
quarantined: backlinkJetstream?.isQuarantined ?? false,
protocolFailures: backlinkJetstream?.protocolFailureCount ?? 0,
failureKind: backlinkJetstream?.failureKind,
lastProgressAt: backlinkJetstream?.lastProgressTime,
},
registry: {
quarantined: registryJetstream?.isQuarantined ?? false,
protocolFailures: registryJetstream?.protocolFailureCount ?? 0,
failureKind: registryJetstream?.failureKind,
lastProgressAt: registryJetstream?.lastProgressTime,
},
},
healthy:
firehoseStarted && registryProgressing && directProgressing && backlinkProgressing && !registrySnapshotOverflow,
}
}
export function getEventStats() {
return {
events: totalEvents,
matched: totalMatched,
tooComplexBacklinkEvents,
invalidRegistryRecords,
invalidDeliveryInputEvents,
rejectedSubscriptionAdmissions,
}
}
/** Test-only reset after a fully stopped client set. */
export function resetFirehoseForTests(): void {
stopFirehose()
for (const client of allClients) client.destroy()
allClients.clear()
directJetstream = null
backlinkJetstream = null
registryJetstream = null
dependencies = null
priorReferenceKeys = null
stopping = false
intakePaused = false
lastEventTime = 0
totalEvents = 0
totalMatched = 0
tooComplexBacklinkEvents = 0
invalidRegistryRecords = 0
invalidDeliveryInputEvents = 0
rejectedSubscriptionAdmissions = 0
registrySnapshotOverflow = false
trackedWebhooks.clear()
scopeWebhookKeys.clear()
backlinkWebhookKeys.clear()
ownerRefCounts.clear()
scopeRefCounts.clear()
backlinkScopeRefCounts.clear()
scopeDids = new Set<string>()
ownerDids = new Set<string>()
backlinkScopeDids = new Set<string>()
directWantedDids = new Set<string>()
directSubscriptionUrlBytes = Buffer.byteLength(new URL(config.jetstreamUrl).toString())
clearRegistryCache()
}