import { isCanonicalWebhookDid } from '@wispplace/atproto-utils'
import { createLogger } from '@wispplace/observability'
import { config } from '../config'
import {
MAX_JETSTREAM_SUBSCRIPTION_URL_BYTES,
MAX_JETSTREAM_WANTED_COLLECTIONS,
MAX_JETSTREAM_WANTED_DIDS,
} from './admission'
import { isValidAtprotoRevision } from './atproto-revision'
const logger = createLogger('webhook-service:jetstream')
export interface JetstreamCommit {
rev: string
operation: 'create' | 'update' | 'delete'
collection: string
rkey: string
record?: unknown
cid?: string
}
export interface JetstreamEvent {
did: string
time_us: number
kind: 'commit' | 'identity' | 'account'
commit?: JetstreamCommit
}
export interface JetstreamOptions {
url: string
wantedDids?: readonly string[]
wantedCollections?: readonly string[]
cursor?: number
/**
* Resolves only after every durable effect for this batch is committed,
* including the stream cursor. Rejecting replays the whole batch.
*/
onEvents: (events: readonly JetstreamEvent[]) => void | Promise<void>
onConnect?: () => void
onDisconnect?: () => void
onError?: (err: Error) => void
maxQueue?: number
/** Events handed to onEvents at once. One batch is one durable transaction. */
batchMax?: number
maxEventBytes?: number
reconnectMinMs?: number
reconnectMaxMs?: number
reconnectMaxExponent?: number
/** How long a recovering stream may go without durable progress before it is stalled. */
progressStaleMs?: number
/** Handler attempts for one batch before the whole stream replays. */
batchMaxAttempts?: number
batchRetryDelayMs?: number
}
function isStableDid(value: string): boolean {
return isCanonicalWebhookDid(value)
}
/** A recovering stream stays healthy only while it keeps acknowledging work. */
const DEFAULT_PROGRESS_STALE_MS = 60_000
/** A transient handler failure is the batch's problem, not the queue's. */
const DEFAULT_BATCH_MAX_ATTEMPTS = 3
const DEFAULT_BATCH_RETRY_DELAY_MS = 250
const COLLECTION_RE = /^[A-Za-z0-9.-]{1,253}$/
const RKEY_RE = /^[A-Za-z0-9._~:%@+-]{1,512}$/
function callback(callback: (() => void) | undefined): void {
try {
callback?.()
} catch {
// Lifecycle observers cannot be allowed to break cursor handling.
}
}
function report(options: JetstreamOptions, err: Error): void {
try {
options.onError?.(err)
} catch {
// Error observers are diagnostic only.
}
}
function validCursor(value: unknown): value is number {
return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0
}
/** Validate a relay message before it can influence a durable cursor. */
export function parseJetstreamEvent(value: unknown): JetstreamEvent | null {
if (!value || typeof value !== 'object' || Array.isArray(value)) return null
const event = value as Record<string, unknown>
const did = event.did
const timeUs = event.time_us
const kind = event.kind
if (typeof did !== 'string' || !isStableDid(did)) return null
if (!validCursor(timeUs)) return null
if (kind !== 'commit' && kind !== 'identity' && kind !== 'account') return null
if (kind !== 'commit') {
return { did, time_us: timeUs, kind }
}
if (!event.commit || typeof event.commit !== 'object' || Array.isArray(event.commit)) return null
const commit = event.commit as Record<string, unknown>
if (
typeof commit.rev !== 'string' ||
!isValidAtprotoRevision(commit.rev) ||
(commit.operation !== 'create' && commit.operation !== 'update' && commit.operation !== 'delete') ||
typeof commit.collection !== 'string' ||
!COLLECTION_RE.test(commit.collection) ||
typeof commit.rkey !== 'string' ||
!RKEY_RE.test(commit.rkey) ||
(commit.operation !== 'delete' && commit.record === undefined) ||
(commit.cid !== undefined && (typeof commit.cid !== 'string' || commit.cid.length === 0 || commit.cid.length > 512))
) {
return null
}
return {
did,
time_us: timeUs,
kind: 'commit',
commit: {
rev: commit.rev,
operation: commit.operation,
collection: commit.collection,
rkey: commit.rkey,
...(commit.record === undefined ? {} : { record: commit.record }),
...(commit.cid === undefined ? {} : { cid: commit.cid }),
},
}
}
/** A credential-free, query-free relay identity for cursors and delivery identities. */
export function normalizeRelayIdentity(value: string): string {
const url = new URL(value)
url.username = ''
url.password = ''
url.search = ''
url.hash = ''
const pathname = url.pathname.replace(/\/+$/, '') || '/'
return `${url.protocol}//${url.host.toLowerCase()}${pathname}`
}
function messageText(data: unknown, maxBytes: number): string | null {
if (typeof data === 'string') return Buffer.byteLength(data) <= maxBytes ? data : null
if (data instanceof ArrayBuffer) {
if (data.byteLength > maxBytes) return null
return new TextDecoder().decode(data)
}
if (ArrayBuffer.isView(data)) {
if (data.byteLength > maxBytes) return null
return Buffer.from(data.buffer, data.byteOffset, data.byteLength).toString('utf8')
}
return null
}
/** Build and hard-limit the exact encoded relay subscription URL. */
export function buildJetstreamSubscriptionUrl(
options: Pick<JetstreamOptions, 'url' | 'wantedDids' | 'wantedCollections'>,
cursor?: number,
): string {
const url = new URL(options.url)
// Only this client owns subscription query parameters. Never preserve a
// caller-supplied cursor/filter that could broaden replay scope.
url.search = ''
let didCount = 0
for (const did of options.wantedDids ?? []) {
if (!isStableDid(did)) continue
if (++didCount > MAX_JETSTREAM_WANTED_DIDS) throw new Error('Jetstream wanted-DID limit exceeded')
url.searchParams.append('wantedDids', did)
}
let collectionCount = 0
for (const collection of options.wantedCollections ?? []) {
if (!COLLECTION_RE.test(collection)) continue
if (++collectionCount > MAX_JETSTREAM_WANTED_COLLECTIONS)
throw new Error('Jetstream wanted-collection limit exceeded')
url.searchParams.append('wantedCollections', collection)
}
if (cursor !== undefined) url.searchParams.set('cursor', String(cursor))
const result = url.toString()
if (Buffer.byteLength(result) > MAX_JETSTREAM_SUBSCRIPTION_URL_BYTES)
throw new Error('Jetstream subscription URL limit exceeded')
return result
}
/**
* A bounded Jetstream consumer. Events are not considered acknowledged until the
* async handler and optional durable-cursor callback both resolve. Queue overflow,
* parse errors, and handler errors deliberately reconnect from the last acked cursor.
*/
export class JetstreamClient {
private ws: WebSocket | null = null
private destroyed = false
private accepting = true
private paused = false
private connected = false
private reconnectTimer: ReturnType<typeof setTimeout> | null = null
private reconnectAttempts = 0
private lastAckedCursor: number | undefined
private queue: JetstreamEvent[] = []
private running = false
private attempts = 0
private failed = false
// Queue pressure is recoverable. Keep accepted events and reconnect only
// after they have drained and advanced the durable cursor.
private backpressured = false
private reconnectWhenDrained = false
private quarantined = false
private protocolFailures = 0
private lastProgressAt = 0
private lastFailureAt = 0
private lastFailureKind: 'protocol' | 'queue' | 'handler' | 'connect' | undefined
private lastReportedErrorAt = 0
private generation = 0
private readonly drainWaiters = new Set<() => void>()
private readonly maxQueue: number
private readonly batchMax: number
private readonly maxEventBytes: number
private readonly reconnectMinMs: number
private readonly reconnectMaxMs: number
private readonly reconnectMaxExponent: number
private readonly progressStaleMs: number
private readonly batchMaxAttempts: number
private readonly batchRetryDelayMs: number
private readonly retryTimers = new Set<ReturnType<typeof setTimeout>>()
constructor(private readonly opts: JetstreamOptions) {
let relay: URL
try {
relay = new URL(opts.url)
} catch {
throw new Error('Invalid Jetstream relay URL')
}
const loopback = ['127.0.0.1', '[::1]', 'localhost'].includes(relay.hostname)
if (
relay.username ||
relay.password ||
relay.search ||
relay.hash ||
!relay.hostname ||
(relay.protocol !== 'wss:' && !(config.allowInsecureDevelopment && loopback && relay.protocol === 'ws:'))
) {
throw new Error('Unsafe Jetstream relay URL')
}
this.lastAckedCursor = opts.cursor
this.maxQueue = opts.maxQueue ?? config.intakeQueueMax
// A batch is drawn from the queue, so it can never usefully exceed it.
this.batchMax = Math.min(opts.batchMax ?? config.intakeBatchMax, this.maxQueue)
this.maxEventBytes = opts.maxEventBytes ?? config.intakeEventMaxBytes
this.reconnectMinMs = opts.reconnectMinMs ?? config.jetstreamReconnectMinMs
this.reconnectMaxMs = opts.reconnectMaxMs ?? config.jetstreamReconnectMaxMs
this.reconnectMaxExponent = opts.reconnectMaxExponent ?? config.jetstreamReconnectMaxExponent
this.progressStaleMs = opts.progressStaleMs ?? DEFAULT_PROGRESS_STALE_MS
this.batchMaxAttempts = opts.batchMaxAttempts ?? DEFAULT_BATCH_MAX_ATTEMPTS
this.batchRetryDelayMs = opts.batchRetryDelayMs ?? DEFAULT_BATCH_RETRY_DELAY_MS
if (
!Number.isSafeInteger(this.maxQueue) ||
this.maxQueue < 1 ||
!Number.isSafeInteger(this.batchMax) ||
this.batchMax < 1 ||
!Number.isSafeInteger(this.maxEventBytes) ||
this.maxEventBytes < 1
) {
throw new Error('Invalid bounded Jetstream queue configuration')
}
}
start(): void {
if (
this.destroyed ||
!this.accepting ||
this.ws ||
this.reconnectTimer ||
this.backpressured ||
this.reconnectWhenDrained
)
return
this.connect()
}
private buildUrl(cursor = this.lastAckedCursor): string {
return buildJetstreamSubscriptionUrl(this.opts, cursor)
}
private connect(): void {
if (this.destroyed || !this.accepting || this.ws || this.backpressured || this.reconnectWhenDrained) return
this.failed = false
const generation = ++this.generation
let ws: WebSocket
try {
ws = new WebSocket(this.buildUrl())
} catch {
this.scheduleReconnect(new Error('Jetstream connection setup failed'), 'connect')
return
}
this.ws = ws
ws.onopen = () => {
if (this.ws !== ws || generation !== this.generation || this.destroyed || !this.accepting) return
this.connected = true
callback(this.opts.onConnect)
}
ws.onmessage = (message) => {
if (
this.ws !== ws ||
generation !== this.generation ||
this.destroyed ||
!this.accepting ||
this.failed ||
this.backpressured
)
return
const text = messageText(message.data, this.maxEventBytes)
if (!text) {
this.quarantine(new Error('Jetstream message exceeded the intake limit'))
return
}
let event: JetstreamEvent | null
try {
event = parseJetstreamEvent(JSON.parse(text))
} catch {
event = null
}
if (!event) {
this.quarantine(new Error('Jetstream protocol event was invalid'))
return
}
if (this.queue.length >= this.maxQueue) {
// A close cannot put the already received frame back on the wire.
// This should only be reachable after an event-loop race; retain the
// bounded queue and let the relay replay this frame from the last ack.
this.applyBackpressure(new Error('Jetstream intake queue is full'))
return
}
this.queue.push(event)
this.pump()
if (this.queue.length >= this.maxQueue) this.applyBackpressure(new Error('Jetstream intake queue is full'))
}
ws.onerror = () => {
if (this.ws !== ws || generation !== this.generation || this.destroyed || !this.accepting) return
// onclose schedules reconnect. This only records a safe diagnostic state.
this.report(new Error('Jetstream WebSocket error'))
}
ws.onclose = () => {
if (this.ws !== ws || generation !== this.generation) return
this.ws = null
const wasConnected = this.connected
this.connected = false
if (wasConnected) callback(this.opts.onDisconnect)
if (!this.destroyed && this.accepting) {
// Never replay from a stale cursor while accepted work is still
// draining, even when the relay (rather than queue pressure) closed
// the socket.
if (this.backpressured || this.reconnectWhenDrained || this.queue.length !== 0 || this.running) {
this.reconnectWhenDrained = true
this.reconnectAfterDrain()
} else this.scheduleReconnect()
}
}
}
private pump(): void {
if (this.destroyed || this.failed || this.paused || this.running || this.queue.length === 0) return
this.running = true
void this.run(this.queue.slice(0, this.batchMax))
}
/**
* One batch is one durable unit: the handler resolves only after every effect
* for these events, including the stream cursor, is committed. Acknowledging
* is therefore just dropping the batch from the queue - there is no window in
* which the cursor is ahead of the work it claims.
*/
private async run(batch: readonly JetstreamEvent[]): Promise<void> {
const last = batch[batch.length - 1]
try {
if (!last) return
await this.opts.onEvents(batch)
if (this.destroyed || this.failed) return
this.queue.splice(0, batch.length)
this.lastAckedCursor = last.time_us
// A durable acknowledgement proves the connection carried useful work;
// only then reset exponential reconnect backoff.
this.attempts = 0
this.reconnectAttempts = 0
this.protocolFailures = 0
this.quarantined = false
this.lastProgressAt = Date.now()
} catch {
this.retryOrFail()
} finally {
this.running = false
this.pump()
this.notifyDrained()
}
}
/**
* A transient failure belongs to its batch. Retrying it in place keeps every
* event this connection already accepted; only a batch that keeps failing
* costs the stream a replay from its last durable acknowledgement.
*/
private retryOrFail(): void {
if (this.destroyed || this.failed) return
this.attempts++
if (this.attempts >= this.batchMaxAttempts) {
this.fail(new Error('Jetstream event handler failed'), 'handler')
return
}
this.report(new Error('Jetstream event handler failed'))
const timer = setTimeout(() => {
this.retryTimers.delete(timer)
if (this.destroyed || this.failed) return
this.pump()
}, this.batchRetryDelayMs * this.attempts)
this.retryTimers.add(timer)
}
private report(error: Error): void {
const now = Date.now()
if (now - this.lastReportedErrorAt < 60_000) return
this.lastReportedErrorAt = now
report(this.opts, error)
}
private applyBackpressure(error: Error): void {
if (this.destroyed || this.failed || this.backpressured) return
this.backpressured = true
this.lastFailureKind = 'queue'
this.lastFailureAt = Date.now()
this.report(error)
try {
this.ws?.close()
} catch {
// A close race is benign.
}
// onclose schedules only after this bounded queue drains. If a custom
// socket closes synchronously, this remains a no-op until then.
this.reconnectAfterDrain()
}
private reconnectAfterDrain(): void {
if (
(!this.backpressured && !this.reconnectWhenDrained) ||
this.destroyed ||
!this.accepting ||
this.ws ||
this.queue.length !== 0 ||
this.running
)
return
this.backpressured = false
this.reconnectWhenDrained = false
this.scheduleReconnect()
}
private quarantine(error: Error): void {
if (this.destroyed || this.failed) return
this.backpressured = false
this.reconnectWhenDrained = false
this.quarantined = true
this.protocolFailures = Math.min(this.protocolFailures + 1, this.reconnectMaxExponent + 1)
this.lastFailureKind = 'protocol'
this.lastFailureAt = Date.now()
this.failed = true
this.report(error)
this.queue = []
this.notifyDrained()
try {
this.ws?.close()
} catch {
// A close race is benign.
}
// A malformed frame never advances the cursor. onclose probes again from
// the last durable ack with capped full-jitter backoff. If close races and
// no socket remains, schedule the half-open probe here.
if (!this.ws && this.accepting) this.scheduleReconnect()
}
private fail(error: Error, kind: 'queue' | 'handler' = 'handler'): void {
if (this.destroyed || this.failed) return
this.backpressured = false
// A disconnected socket can fail while sibling handlers still own work.
// Drop their unacknowledged queue entries, but do not replay over those
// side effects until every in-flight handler and cursor flush has settled.
this.reconnectWhenDrained = this.running
this.lastFailureKind = kind
this.lastFailureAt = Date.now()
this.failed = true
this.report(error)
// Do not retain unacknowledged items. Their cursor was never persisted and the
// relay replay is deduplicated by the durable delivery/event state.
this.queue = []
this.notifyDrained()
try {
this.ws?.close()
} catch {
// A close race is benign.
}
if (!this.ws && this.accepting) this.scheduleReconnect()
}
private scheduleReconnect(error?: Error, kind?: 'connect'): void {
if (this.destroyed || !this.accepting || this.reconnectTimer || this.backpressured || this.reconnectWhenDrained)
return
if (error) {
if (kind) {
this.lastFailureKind = kind
this.lastFailureAt = Date.now()
}
this.report(error)
}
this.reconnectAttempts = Math.min(this.reconnectAttempts + 1, this.reconnectMaxExponent + 1)
const exponent = Math.min(
Math.max(this.reconnectAttempts - 1, this.protocolFailures - 1),
this.reconnectMaxExponent,
)
const ceiling = Math.min(this.reconnectMaxMs, this.reconnectMinMs * 2 ** exponent)
// Full jitter prevents synchronized reconnect storms. After repeated protocol
// poison, probe only in the slow half of the capped interval.
const lower = this.protocolFailures >= this.reconnectMaxExponent + 1 ? Math.ceil(ceiling / 2) : 1
const delay = lower + Math.floor(Math.random() * Math.max(1, ceiling - lower + 1))
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null
this.connect()
}, delay)
}
get cursor(): number | undefined {
return this.lastAckedCursor
}
/**
* How far the durable cursor trails the present, in milliseconds. A relay
* only retains a few days of replay, so an age approaching that window means
* events are about to be lost silently rather than merely late.
*/
get cursorAgeMs(): number | undefined {
if (this.lastAckedCursor === undefined) return undefined
return Math.max(0, Date.now() - Math.floor(this.lastAckedCursor / 1000))
}
get queued(): number {
return this.queue.length
}
get isConnected(): boolean {
return this.connected && !this.failed && !this.quarantined && !this.backpressured
}
/**
* Intake health is durable progress, not socket state. A bounded queue that
* closes the socket, drains, and replays from its last acknowledgement is
* working exactly as designed; only a quarantined stream, or one that has
* acknowledged nothing inside the window, has actually stopped serving.
*/
get isProgressing(): boolean {
if (this.destroyed || this.quarantined) return false
if (this.isConnected) return true
return this.lastProgressAt !== 0 && Date.now() - this.lastProgressAt < this.progressStaleMs
}
get isQuarantined(): boolean {
return this.quarantined
}
get protocolFailureCount(): number {
return this.protocolFailures
}
get lastProgressTime(): number | undefined {
return this.lastProgressAt || undefined
}
get lastFailureTime(): number | undefined {
return this.lastFailureAt || undefined
}
get failureKind(): 'protocol' | 'queue' | 'handler' | 'connect' | undefined {
return this.lastFailureKind
}
pause(): void {
this.paused = true
}
resume(): void {
if (this.destroyed) return
this.paused = false
this.pump()
}
/** Stop reading the socket, retain accepted work, and let drain() finish it. */
stopAccepting(): void {
if (!this.accepting) return
this.accepting = false
this.paused = false
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer)
this.reconnectTimer = null
}
try {
this.ws?.close()
} catch {
// A close race is benign.
}
this.pump()
}
async drain(): Promise<void> {
if (this.queue.length === 0 && !this.running) return
await new Promise<void>((resolve) => this.drainWaiters.add(resolve))
}
private notifyDrained(): void {
if (this.queue.length !== 0 || this.running) return
this.reconnectAfterDrain()
for (const resolve of this.drainWaiters) resolve()
this.drainWaiters.clear()
}
/** Permanently close the websocket and discard work that was not accepted durably. */
destroy(): void {
if (this.destroyed) return
this.destroyed = true
this.accepting = false
this.paused = false
this.backpressured = false
this.reconnectWhenDrained = false
if (this.reconnectTimer) clearTimeout(this.reconnectTimer)
this.reconnectTimer = null
for (const timer of this.retryTimers) clearTimeout(timer)
this.retryTimers.clear()
this.queue = []
try {
this.ws?.close()
} catch {
// A close race is benign.
}
this.ws = null
this.connected = false
this.notifyDrained()
}
}
export function getJetstreamLoggerForTests(): typeof logger {
return logger
}