Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263import type { JetstreamEvent } from '@atcute/jetstream';import type { Subscribe } from '../src/jetstream';
const SECOND = 1_000_000;
export function commit( seconds: number, did: string, collection: string, rkey: string, operation: 'create' | 'update' | 'delete' = 'create', record: Record<string, unknown> = {},): JetstreamEvent { const base = { rev: `rev-${seconds}`, collection, rkey, operation }; return { did, time_us: seconds * SECOND, kind: 'commit', commit: operation === 'delete' ? base : { ...base, cid: 'bafy', record }, } as JetstreamEvent;}
export function identity(seconds: number, did: string, handle: string): JetstreamEvent { return { did, time_us: seconds * SECOND, kind: 'identity', identity: { did, handle, seq: seconds, time: new Date(0).toISOString() }, } as JetstreamEvent;}
const parked = () => new Promise<never>(() => {});
// replays whatever source() holds from the cursor on, then goes quiet like a// caught up jetstream. dropAfter[n] closes the nth connection after that many eventsexport function fakeJetstream(source: () => JetstreamEvent[], dropAfter: (number | undefined)[] = []) { const cursors: number[] = []; const subscribe: Subscribe = (_url, cursor, _collections, onClose) => { const connection = cursors.push(cursor) - 1; const limit = dropAfter[connection]; const pending = source().filter((e) => e.time_us >= cursor); return { async *[Symbol.asyncIterator]() { for (const [i, evt] of pending.entries()) { if (i === limit) break; yield evt; } if (limit !== undefined) onClose(); await parked(); }, }; }; return { subscribe, cursors };}
// a jetstream that refuses every connectionexport const deadJetstream: Subscribe = (_url, _cursor, _collections, onClose) => ({ async *[Symbol.asyncIterator]() { queueMicrotask(onClose); await parked(); },});