import { zstdDecompressSync } from 'node:zlib';
import { cborEncodeDagCbor } from '@pdsjs/core/repo';
import { describe, expect, it } from 'vitest';
import {
  decodeBlock,
  decodeBlockColumns,
  decodeSegment,
  getBlock,
  planSnapshot,
  replayArchive,
  rowToEvent,
} from '../src/archive.js';
import { blockFrame, encodeBlockColumns } from './helpers/archive-block.js';

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

/**
 * @param {Uint8Array} bytes
 * @returns {Promise<Uint8Array>}
 */
const decompress = async (bytes) =>
  new Uint8Array(zstdDecompressSync(Buffer.from(bytes)));

const RECORD = {
  ref: 'refs/heads/fix',
  subject: { uri: 'at://did:plc:owner/dev.pdsjs.git.repo/proj' },
};

describe('decodeBlockColumns', () => {
  it('round-trips the rows the encoder wrote', () => {
    const rows = [
      {
        seq: 1000,
        kind: 1,
        did: DID,
        collection: COLLECTION,
        rkey: '3kpull',
        rev: '3krev',
        payload: new Uint8Array([1, 2]),
      },
      {
        seq: 1001,
        kind: 3,
        did: DID,
        collection: COLLECTION,
        rkey: '3kdel',
        rev: '',
        payload: new Uint8Array(),
      },
    ];
    const decoded = decodeBlockColumns(encodeBlockColumns(rows));
    expect(decoded).toHaveLength(2);
    expect(decoded[0]).toMatchObject({
      seq: 1000,
      kind: 1,
      did: DID,
      collection: COLLECTION,
      rkey: '3kpull',
      rev: '3krev',
    });
    expect([...decoded[0].payload]).toEqual([1, 2]);
    expect(decoded[1].kind).toBe(3);
    expect(decoded[1].rkey).toBe('3kdel');
  });

  it('decodes a zstd frame through the injected decompressor', async () => {
    const rows = [
      {
        seq: 5,
        kind: 1,
        did: DID,
        collection: COLLECTION,
        rkey: '3kpull',
        rev: '3krev',
        payload: new Uint8Array(),
      },
    ];
    const decoded = await decodeBlock(blockFrame(rows), decompress);
    expect(decoded[0].seq).toBe(5);
  });
});

describe('decodeSegment', () => {
  it('picks the block frames out of a sealed segment file', async () => {
    const rows = [
      {
        seq: 7,
        kind: 1,
        did: DID,
        collection: COLLECTION,
        rkey: '3kpull',
        rev: '3krev',
        payload: new Uint8Array(),
      },
    ];
    const frame = blockFrame(rows);
    const header = 256;
    const footer = header + 8 + frame.length;
    const file = new Uint8Array(footer + 52);
    const view = new DataView(file.buffer);
    file.set(new TextEncoder().encode('jss0'), 0);
    view.setUint32(14, 1, true);
    view.setBigUint64(58, BigInt(footer), true);
    view.setUint32(header, frame.length, true);
    file.set(frame, header + 8);
    view.setBigUint64(footer, BigInt(header), true);
    view.setUint32(footer + 8, frame.length, true);

    const blocks = await decodeSegment(file, decompress);
    expect(blocks).toHaveLength(1);
    expect(blocks[0][0].seq).toBe(7);
  });
});

describe('rowToEvent', () => {
  it('maps a create onto a record write whose value is the CBOR record', () => {
    const row = {
      seq: 9,
      kind: 1,
      did: DID,
      collection: COLLECTION,
      rkey: '3kpull',
      rev: '3krev',
      payload: cborEncodeDagCbor(RECORD),
    };
    expect(rowToEvent(row)).toEqual({
      did: DID,
      collection: COLLECTION,
      rkey: '3kpull',
      cid: null,
      uri: `at://${DID}/${COLLECTION}/3kpull`,
      seq: 9,
      value: RECORD,
    });
  });

  it('maps a resync create onto a record write', () => {
    const row = {
      seq: 10,
      kind: 7,
      did: DID,
      collection: COLLECTION,
      rkey: '3kresync',
      rev: '3krev',
      payload: cborEncodeDagCbor(RECORD),
    };
    expect(rowToEvent(row)).toMatchObject({
      uri: expect.stringContaining('3kresync'),
    });
  });

  it('maps a delete onto a marker', () => {
    const row = {
      seq: 11,
      kind: 3,
      did: DID,
      collection: COLLECTION,
      rkey: '3kdel',
      rev: '',
      payload: new Uint8Array(),
    };
    expect(rowToEvent(row)).toEqual({
      did: DID,
      collection: COLLECTION,
      rkey: '3kdel',
      uri: `at://${DID}/${COLLECTION}/3kdel`,
      seq: 11,
      deleted: true,
    });
  });

  it('reads the handle from an identity payload and skips one without', () => {
    expect(
      rowToEvent({
        seq: 12,
        kind: 4,
        did: DID,
        collection: '',
        rkey: '',
        rev: '',
        payload: cborEncodeDagCbor({ handle: 'stranger.example.com' }),
      }),
    ).toEqual({
      kind: 'identity',
      did: DID,
      handle: 'stranger.example.com',
      seq: 12,
    });
    expect(
      rowToEvent({
        seq: 13,
        kind: 4,
        did: DID,
        collection: '',
        rkey: '',
        rev: '',
        payload: new Uint8Array(),
      }),
    ).toBe(null);
  });

  it('reads the status from an account payload and skips one without', () => {
    expect(
      rowToEvent({
        seq: 14,
        kind: 5,
        did: DID,
        collection: '',
        rkey: '',
        rev: '',
        payload: cborEncodeDagCbor({ active: false, status: 'deleted' }),
      }),
    ).toEqual({ kind: 'account', did: DID, status: 'deleted', seq: 14 });
    expect(
      rowToEvent({
        seq: 15,
        kind: 5,
        did: DID,
        collection: '',
        rkey: '',
        rev: '',
        payload: new Uint8Array(),
      }),
    ).toBe(null);
  });
});

describe('planSnapshot', () => {
  it('sends the filter and window and returns the plan', async () => {
    /** @type {Array<{url: string, init: RequestInit}>} */
    const calls = [];
    const fetch = /** @type {import('../src/archive.js').FetchLike} */ (
      async (input, init) => {
        calls.push({ url: String(input), init: init ?? {} });
        return new Response(
          JSON.stringify({
            sealedTipSeq: 2000,
            plannedThroughSeq: 2000,
            segments: [],
          }),
          { status: 200 },
        );
      }
    );
    const plan = await planSnapshot({
      service: 'https://jetstream.example',
      collections: [COLLECTION],
      afterSeq: 42,
      beforeSeq: 100,
      apiKey: 'key',
      fetch,
    });
    expect(plan.sealedTipSeq).toBe(2000);
    const body = JSON.parse(String(calls[0].init.body));
    expect(body).toMatchObject({
      collections: [COLLECTION],
      afterSeq: 42,
      beforeSeq: 100,
    });
    const headers = /** @type {Record<string, string>} */ (
      calls[0].init.headers
    );
    expect(headers.authorization).toBe('Bearer key');
  });

  it('backs off a 429 and retries after Retry-After', async () => {
    let tries = 0;
    const fetch = async () => {
      tries += 1;
      if (tries === 1) {
        return new Response('{"error":"byte limit exceeded"}', {
          status: 429,
          headers: { 'retry-after': '0' },
        });
      }
      return new Response(
        JSON.stringify({ sealedTipSeq: 1, plannedThroughSeq: 1, segments: [] }),
        { status: 200 },
      );
    };
    await planSnapshot({
      service: 'https://jetstream.example',
      fetch,
    });
    expect(tries).toBe(2);
  });
});

describe('getBlock', () => {
  it('downloads the frame for one block index', async () => {
    const frame = new Uint8Array([1, 2, 3]);
    const fetch = /** @type {import('../src/archive.js').FetchLike} */ (
      async (input) => {
        expect(String(input)).toContain('segment=seg_0000000000.jss');
        expect(String(input)).toContain('blockIndex=3');
        return new Response(/** @type {BodyInit} */ (frame), { status: 200 });
      }
    );
    expect(
      await getBlock({
        service: 'https://jetstream.example',
        segment: 'seg_0000000000.jss',
        blockIndex: 3,
        fetch,
      }),
    ).toEqual(frame);
  });
});

describe('replayArchive', () => {
  it('plans, downloads, folds and checkpoints to the sealed tip', async () => {
    const rows = [
      {
        seq: 10,
        kind: 1,
        did: DID,
        collection: COLLECTION,
        rkey: '3kpull',
        rev: '3krev',
        payload: cborEncodeDagCbor(RECORD),
      },
      {
        seq: 11,
        kind: 3,
        did: DID,
        collection: COLLECTION,
        rkey: '3kdel',
        rev: '',
        payload: new Uint8Array(),
      },
      {
        seq: 12,
        kind: 1,
        did: 'did:plc:other',
        collection: 'app.bsky.feed.post',
        rkey: '3kpost',
        rev: '',
        payload: cborEncodeDagCbor({}),
      },
    ];
    const frame = blockFrame(rows);
    const plans = [
      {
        sealedTipSeq: 2000,
        plannedThroughSeq: 1000,
        segments: [
          {
            name: 'seg_0000000000.jss',
            index: 0,
            mode: 'blocks',
            blocks: [{ first: 0, last: 0 }],
          },
        ],
      },
      {
        sealedTipSeq: 2000,
        plannedThroughSeq: 2000,
        segments: [],
      },
    ];
    let planCalls = 0;
    const fetch = /** @type {import('../src/archive.js').FetchLike} */ (
      async (input, _init) => {
        if (String(input).includes('planSnapshot')) {
          const plan = plans[planCalls];
          planCalls += 1;
          return new Response(JSON.stringify(plan), { status: 200 });
        }
        return new Response(/** @type {BodyInit} */ (frame), { status: 200 });
      }
    );

    /** @type {any[]} */
    const events = [];
    /** @type {number[]} */
    const checkpoints = [];
    const tip = await replayArchive({
      service: 'https://jetstream.example',
      collections: [COLLECTION],
      afterSeq: 0,
      fetch,
      decompress,
      onEvent: (event) => {
        events.push(event);
      },
      onSeq: (seq) => {
        checkpoints.push(seq);
      },
    });

    expect(tip).toBe(2000);
    expect(events.map((event) => event.uri)).toEqual([
      `at://${DID}/${COLLECTION}/3kpull`,
      `at://${DID}/${COLLECTION}/3kdel`,
    ]);
    expect(events[0].value).toEqual(RECORD);
    expect(events[1].deleted).toBe(true);
    expect(checkpoints).toEqual([1000, 2000]);
    expect(planCalls).toBe(2);
  });
});