import { afterEach, describe, expect, mock, test } from 'bun:test'
import type { Main as WhRecord } from '@wispplace/lexicons/types/place/wisp/v2/wh'
import type { CursorRepository, PriorReferenceRepository } from './firehose'
import type { JetstreamClient, JetstreamEvent, JetstreamOptions } from './jetstream'
// 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 () => 'stored',
deletePriorReferenceIndex: async () => false,
loadPriorReferenceKeys: async () => ({ keys: [], complete: true }),
runIntakeBatch: async (work: (sql: unknown) => Promise<unknown>) => work(undefined),
enqueueWebhookDeliveries: async () => ({ enqueued: 0, deduplicated: 0 }),
recordWebhookIntakeQuarantine: async () => undefined,
getWebhookRecord: async () => undefined,
upsertWebhookRecord: async () => false,
deleteWebhookRecord: async () => false,
loadActiveWebhooks: async () => ({ rows: [], overflow: false }),
loadActiveWebhooksForOwner: async () => ({ rows: [], overflow: false }),
}))
const firehose = await import('./firehose')
const delivery = await import('./delivery')
const { getEventStats, initScopeDids, resetFirehoseForTests, startFirehose } = firehose
const OWNER = 'did:plc:aaaaaaaaaaaaaaaaaaaaaaaa'
const SCOPE = 'did:plc:bbbbbbbbbbbbbbbbbbbbbbbb'
const REV = '3lq6x5f2abcde'
class FakeJetstream {
started = false
paused = false
constructor(readonly options: JetstreamOptions) {}
get isConnected(): boolean {
return true
}
get isQuarantined(): boolean {
return false
}
get protocolFailureCount(): number {
return 0
}
get queued(): number {
return 0
}
get failureKind(): undefined {
return undefined
}
get lastProgressTime(): undefined {
return undefined
}
start(): void {
this.started = true
}
pause(): void {
this.paused = true
}
resume(): void {
this.paused = false
}
stopAccepting(): void {}
async drain(): Promise<void> {}
destroy(): void {}
}
function record(scopeDid = SCOPE, enabled = true): WhRecord {
return {
$type: 'place.wisp.v2.wh',
scope: { aturi: `at://${scopeDid}` },
url: 'https://receiver.example/hook',
createdAt: '2025-01-01T00:00:00.000Z',
...(enabled ? {} : { enabled: false }),
}
}
function commit(timeUs: number): JetstreamEvent {
return {
did: SCOPE,
time_us: timeUs,
kind: 'commit',
commit: { rev: REV, operation: 'create', collection: 'app.bsky.feed.post', rkey: 'one', record: { text: 'ok' } },
}
}
/** 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) => {
const client = new FakeJetstream(options)
created.push(client)
return client as unknown as JetstreamClient
},
}
}
afterEach(() => {
resetFirehoseForTests()
})
describe('firehose admitted in-memory candidate index', () => {
test('subscribes owner and scope DID, then matches owner A scoped to B without a DB candidate scan', async () => {
initScopeDids([{ did: OWNER, rkey: 'hook', record: record() }])
const created: FakeJetstream[] = []
const enqueued: number[] = []
await startFirehose({
...testDependencies(created),
enqueueWebhookDeliveries: async (entries) => {
enqueued.push(entries.length)
return { enqueued: entries.length, deduplicated: 0 }
},
recordWebhookIntakeQuarantine: async () => undefined,
})
const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE))
expect(direct?.options.wantedDids).toEqual(expect.arrayContaining([OWNER, SCOPE]))
await direct?.options.onEvents([commit(10)])
expect(enqueued).toEqual([1])
})
test('quarantines deterministic poison, commits its cursor path, and continues with a valid event', async () => {
initScopeDids([{ did: OWNER, rkey: 'hook', record: record() }])
const created: FakeJetstream[] = []
const quarantines: string[] = []
const saved: number[] = []
let poison = true
await startFirehose({
...testDependencies(created),
cursorRepository: {
load: async () => undefined,
save: async (_stream, cursor) => {
saved.push(cursor)
},
},
enqueueWebhookDeliveries: async (entries) => {
if (poison) {
poison = false
throw new delivery.WebhookDeliveryInputError('payload_too_large', 'too large')
}
return { enqueued: entries.length, deduplicated: 0 }
},
recordWebhookIntakeQuarantine: async (input) => {
quarantines.push(input.reason)
},
})
const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE))
expect(direct).toBeDefined()
await direct?.options.onEvents([commit(20)])
await direct?.options.onEvents([commit(21)])
expect(quarantines).toEqual(['payload_too_large'])
// Each batch commits its own cursor, quarantined poison included.
expect(saved).toEqual([20, 21])
})
test('commits deliveries and the stream cursor in one batch, and neither when it fails', async () => {
initScopeDids([{ did: OWNER, rkey: 'hook', record: record() }])
const created: FakeJetstream[] = []
const effects: string[] = []
let failing = false
await startFirehose({
...testDependencies(created),
runIntakeBatch: async (work) => {
try {
const result = await work(undefined)
effects.push('commit')
return result
} catch (error) {
effects.push('rollback')
throw error
}
},
cursorRepository: {
load: async () => undefined,
save: async (_stream, cursor) => {
effects.push(`cursor:${cursor}`)
},
},
enqueueWebhookDeliveries: async (entries) => {
effects.push('enqueue')
if (failing) throw new Error('outbox unavailable')
return { enqueued: entries.length, deduplicated: 0 }
},
recordWebhookIntakeQuarantine: async () => undefined,
})
const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE))
expect(direct).toBeDefined()
await direct?.options.onEvents([commit(10)])
// The cursor is the last statement of the same transaction as the delivery.
expect(effects).toEqual(['enqueue', 'cursor:10', 'commit'])
failing = true
effects.length = 0
await expect(direct?.options.onEvents([commit(11)])).rejects.toThrow('Intake event processing failed')
// A rolled back batch never advances the cursor past work it lost.
expect(effects).toEqual(['enqueue', 'rollback'])
})
test('bounds a direct-PDS owner flood deterministically at the per-owner admission cap', async () => {
const rows = Array.from({ length: 101 }, (_, index) => ({
did: OWNER,
rkey: `hook-${String(index).padStart(3, '0')}`,
record: record(),
}))
initScopeDids(rows)
const created: FakeJetstream[] = []
let candidateCount = 0
await startFirehose({
...testDependencies(created),
enqueueWebhookDeliveries: async (entries) => {
candidateCount = entries.length
return { enqueued: entries.length, deduplicated: 0 }
},
recordWebhookIntakeQuarantine: async () => undefined,
})
const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE))
await direct?.options.onEvents([commit(30)])
expect(candidateCount).toBe(100)
expect(getEventStats().rejectedSubscriptionAdmissions).toBeGreaterThanOrEqual(1)
})
test('disabled records do not consume active-owner admission capacity', async () => {
const disabled = Array.from({ length: 500 }, (_, index) => ({
did: OWNER,
rkey: `disabled-${index}`,
record: record(SCOPE, false),
}))
initScopeDids([...disabled, { did: OWNER, rkey: 'active', record: record() }])
const created: FakeJetstream[] = []
let candidateCount = 0
await startFirehose({
...testDependencies(created),
enqueueWebhookDeliveries: async (entries) => {
candidateCount = entries.length
return { enqueued: entries.length, deduplicated: 0 }
},
recordWebhookIntakeQuarantine: async () => undefined,
})
const direct = created.find((client) => client.options.wantedDids?.includes(SCOPE))
await direct?.options.onEvents([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.onEvents([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.onEvents([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.onEvents([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.onEvents([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.onEvents([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.onEvents([likeCommit(10, [], 'update')])
expect(probe.loads).toEqual([VISITOR_KEY])
expect(probe.rows.get(VISITOR_KEY)).toEqual([])
})
})