import { 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');
  });
});