Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364import { describe, expect, it } from 'vitest';import { readJetstream, REWIND_US, type ReadOptions } from '../src/jetstream';import { commit, deadJetstream, fakeJetstream } from './events';
const events = Array.from({ length: 30 }, (_, i) => commit(100 + i, 'did:plc:alice', 'sh.tangled.repo', `repo-${i}`));
function options(overrides: Partial<ReadOptions> = {}): ReadOptions { return { url: 'https://index.tangled.network', cursor: 0, collections: ['sh.tangled.repo'], maxEvents: 1000, idleMs: 5, closeMs: 5, untilUs: Number.MAX_SAFE_INTEGER, maxReconnects: 3, seen: new Set(), ...overrides, };}
describe('readJetstream', () => { it('rewinds on reconnect and drops the replayed events', async () => { const jetstream = fakeJetstream(() => events, [12]); const res = await readJetstream(options(), jetstream.subscribe);
expect(res.error).toBeUndefined(); expect(res.events.map((e) => e.time_us)).toEqual(events.map((e) => e.time_us)); expect(jetstream.cursors).toEqual([0, events[11].time_us - REWIND_US]); expect(res.cursor).toBe(events[29].time_us); });
it('starts every read behind the saved cursor', async () => { const jetstream = fakeJetstream(() => events); const res = await readJetstream(options({ cursor: events[20].time_us }), jetstream.subscribe);
expect(jetstream.cursors).toEqual([events[20].time_us - REWIND_US]); expect(res.events).toHaveLength(20); });
it('stops once events reach the time the run started', async () => { const jetstream = fakeJetstream(() => events); const res = await readJetstream(options({ untilUs: events[4].time_us }), jetstream.subscribe);
expect(res.events).toHaveLength(5); expect(res.full).toBe(false); });
it('says when it stopped because the batch filled up', async () => { const jetstream = fakeJetstream(() => events); const res = await readJetstream(options({ maxEvents: 10 }), jetstream.subscribe);
expect(res.events).toHaveLength(10); expect(res.full).toBe(true); });
it('gives up after too many disconnects', async () => { const res = await readJetstream(options({ maxReconnects: 2 }), deadJetstream);
expect(res.events).toHaveLength(0); expect(res.error).toBe('jetstream disconnected 3 times'); });});