import { zstdCompressSync, zstdDecompressSync } from 'node:zlib';
import { afterEach, describe, expect, it } from 'vitest';
import { WebSocketServer } from 'ws';
import { subscribeJetstream } from '../src/jetstream.js';

const COLLECTION = 'dev.pdsjs.git.pull';
const DID = 'did:plc:stranger';

/**
 * @param {Partial<Record<string, unknown>>} commit
 * @param {Record<string, unknown>} [extra]
 */
const frame = (commit, extra = {}) => ({
  $type: 'message',
  payload: {
    $type: 'network.bsky.jetstream.subscribeEvents#commit',
    did: DID,
    seq: 1000,
    time: '2026-08-13T06:47:43.959305Z',
    operation: 'create',
    collection: COLLECTION,
    rkey: '3kpull',
    rev: '3krev',
    record: { ref: 'refs/heads/fix' },
    cid: 'bafypull',
    ...commit,
  },
  ...extra,
});

describe('subscribeJetstream', () => {
  /** @type {WebSocketServer} */
  let server;

  afterEach(() => {
    server?.close();
  });

  it('hands each event to the handler and advances the sequence', async () => {
    server = new WebSocketServer({ port: 0 });
    const port = /** @type {import('node:net').AddressInfo} */ (
      server.address()
    ).port;
    server.on('connection', (socket) => {
      socket.send(JSON.stringify(frame({})));
      socket.send(JSON.stringify(frame({ operation: 'delete', seq: 2000 })));
      socket.send(
        JSON.stringify({
          $type: 'message',
          payload: {
            $type: 'network.bsky.jetstream.subscribeEvents#identity',
            did: DID,
            seq: 3000,
            time: '2026-08-13T06:47:43.959305Z',
            identity: { handle: 'stranger.example.com' },
          },
        }),
      );
    });

    const controller = new AbortController();
    /** @type {any[]} */
    const events = [];
    /** @type {number[]} */
    const seqs = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onEvent: (event) => {
        events.push(event);
      },
      onSeq: (seq) => {
        seqs.push(seq);
        if (seq === 3000) controller.abort();
      },
    });
    expect(events).toHaveLength(3);
    expect(events[0].cid).toBe('bafypull');
    expect(events[1].deleted).toBe(true);
    expect(events[2]).toEqual({
      kind: 'identity',
      did: DID,
      handle: 'stranger.example.com',
      seq: 3000,
    });
    expect(seqs).toEqual([1000, 2000, 3000]);
  });

  it('reconnects from the last sequence after a drop', async () => {
    /** @type {WebSocketServer} */
    const sockets = new WebSocketServer({ port: 0 }, () => {});
    server = sockets;
    const port = /** @type {import('node:net').AddressInfo} */ (
      sockets.address()
    ).port;
    /** @type {import('ws').WebSocket|undefined} */
    let client;
    sockets.on('connection', (socket) => {
      client = socket;
      socket.send(JSON.stringify(frame({ seq: 500 })));
      // The drop lands once the first event is out.
      setTimeout(() => socket.close(), 50);
    });

    const controller = new AbortController();
    /** @type {string[]} */
    const notices = [];
    const done = subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onNotice: (message) => notices.push(message),
      onEvent: () => {},
      onSeq: () => {
        // The second connection carries the resume cursor.
        if (notices.some((message) => message.includes('from 500'))) {
          controller.abort();
        }
      },
    });
    await done;
    expect(notices.some((message) => message.includes('from 500'))).toBe(true);
    client?.close();
  });

  it('asks the caller how to resume a refused cursor and tails from its answer', async () => {
    const sockets = new WebSocketServer({ port: 0 });
    server = sockets;
    const port = /** @type {import('node:net').AddressInfo} */ (
      sockets.address()
    ).port;
    let refused = true;
    sockets.on('connection', (socket) => {
      if (refused) {
        socket.close(1008, 'CursorTooOld');
      } else {
        socket.send(JSON.stringify(frame({ seq: 200 })));
      }
    });

    const controller = new AbortController();
    /** @type {number[]} */
    const refusedSeqs = [];
    /** @type {string[]} */
    const notices = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      cursor: 42,
      signal: controller.signal,
      onNotice: (message) => notices.push(message),
      onEvent: () => {},
      onSeq: (seq) => {
        if (seq === 200) controller.abort();
      },
      onCursorTooOld: (refusedSeq) => {
        refusedSeqs.push(refusedSeq);
        refused = false;
        return 100;
      },
    });
    expect(refusedSeqs).toEqual([42]);
    expect(notices.some((message) => message.includes('refused'))).toBe(true);
  });

  it('reads compressed frames through the injected decompressor', async () => {
    const sockets = new WebSocketServer({
      port: 0,
      handleProtocols: () => 'zstd',
    });
    server = sockets;
    const port = /** @type {import('node:net').AddressInfo} */ (
      sockets.address()
    ).port;
    /** @type {string[]} */
    const protocols = [];
    /** @type {URL[]} */
    const asked = [];
    sockets.on('connection', (socket, request) => {
      protocols.push(socket.protocol);
      asked.push(new URL(request.url ?? '', 'ws://127.0.0.1'));
      socket.send(zstdCompressSync(Buffer.from(JSON.stringify(frame({})))));
    });

    const controller = new AbortController();
    /** @type {any[]} */
    const events = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      decompress: async (bytes) =>
        new Uint8Array(zstdDecompressSync(Buffer.from(bytes))),
      onEvent: (event) => {
        events.push(event);
      },
      onSeq: () => controller.abort(),
    });
    expect(events).toHaveLength(1);
    expect(events[0].cid).toBe('bafypull');
    expect(protocols).toEqual(['zstd']);
    expect(asked[0].searchParams.get('requireCompression')).toBe('true');
  });

  it('reads a binary frame that is plain JSON', async () => {
    server = new WebSocketServer({ port: 0 });
    const port = /** @type {import('node:net').AddressInfo} */ (
      server.address()
    ).port;
    server.on('connection', (socket) => {
      socket.send(Buffer.from(JSON.stringify(frame({}))), { binary: true });
    });

    const controller = new AbortController();
    /** @type {any[]} */
    const events = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onEvent: (event) => {
        events.push(event);
      },
      onSeq: () => controller.abort(),
    });
    expect(events).toHaveLength(1);
  });

  it('takes the sequence from the envelope, then from the v1 timestamp', async () => {
    server = new WebSocketServer({ port: 0 });
    const port = /** @type {import('node:net').AddressInfo} */ (
      server.address()
    ).port;
    server.on('connection', (socket) => {
      socket.send(
        JSON.stringify(frame({ seq: undefined, rkey: '3kone' }, { cursor: 7 })),
      );
      socket.send(
        JSON.stringify(frame({ seq: undefined, time_us: 8, rkey: '3ktwo' })),
      );
      // No sequence anywhere: the frame indexes, the cursor holds.
      socket.send(JSON.stringify(frame({ seq: undefined, rkey: '3kthree' })));
    });

    const controller = new AbortController();
    /** @type {any[]} */
    const events = [];
    /** @type {number[]} */
    const seqs = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onEvent: (event) => {
        const write = /** @type {any} */ (event);
        events.push(write);
        if (write.rkey === '3kthree') controller.abort();
      },
      onSeq: (seq) => {
        seqs.push(seq);
      },
    });
    expect(events.map((event) => event.rkey)).toEqual([
      '3kone',
      '3ktwo',
      '3kthree',
    ]);
    expect(events[1].seq).toBe(8);
    expect(seqs).toEqual([7, 8]);
  });

  it('notices an event the handler threw on and keeps reading', async () => {
    server = new WebSocketServer({ port: 0 });
    const port = /** @type {import('node:net').AddressInfo} */ (
      server.address()
    ).port;
    server.on('connection', (socket) => {
      socket.send(JSON.stringify(frame({ rkey: '3kbad' })));
      socket.send(JSON.stringify(frame({ rkey: '3kgood', seq: 1001 })));
    });

    const controller = new AbortController();
    /** @type {string[]} */
    const notices = [];
    /** @type {string[]} */
    const seen = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      signal: controller.signal,
      onNotice: (message) => notices.push(message),
      onEvent: (event) => {
        const write = /** @type {any} */ (event);
        if (write.rkey === '3kbad') throw new Error('index is full');
        seen.push(write.rkey);
      },
      onSeq: (seq) => {
        if (seq === 1001) controller.abort();
      },
    });
    expect(seen).toEqual(['3kgood']);
    expect(notices).toContain('event dropped: index is full');
  });

  it('gives up a refused cursor and tails live when no resume is offered', async () => {
    const sockets = new WebSocketServer({ port: 0 });
    server = sockets;
    const port = /** @type {import('node:net').AddressInfo} */ (
      sockets.address()
    ).port;
    let connections = 0;
    /** @type {Array<string|null>} */
    const cursors = [];
    sockets.on('connection', (socket, request) => {
      connections += 1;
      cursors.push(
        new URL(request.url ?? '', 'ws://127.0.0.1').searchParams.get('cursor'),
      );
      if (connections <= 3) socket.close();
      else socket.send(JSON.stringify(frame({ seq: 200 })));
    });

    const controller = new AbortController();
    /** @type {string[]} */
    const notices = [];
    await subscribeJetstream({
      service: `ws://127.0.0.1:${port}`,
      collections: [COLLECTION],
      cursor: 42,
      signal: controller.signal,
      onNotice: (message) => notices.push(message),
      onEvent: () => {},
      onSeq: (seq) => {
        if (seq === 200) controller.abort();
      },
    });
    expect(cursors).toEqual(['42', '42', '42', null]);
    expect(notices).toContain('cursor 42 refused; tailing live');
  });
});