/// <reference types="node" />
import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import { test } from 'node:test';
import { IndexModel } from '../src/index/model.ts';
import { ingest, type IngestDeps, type WebSocketLike } from '../src/index/ingest.ts';
import { sqliteSql } from './sqlite-sql.ts';
import { upstream, type Upstream } from '../src/upstream.ts';

type FakeFrame = { type: 'message'; data: unknown } | { type: 'close' } | { type: 'error' };

/** A fake live tail that plays back a fixed script of frames, one per microtask, and records
 * the URL it was opened with. After the script runs out it stays open and silent. */
function fakeSocket(frames: FakeFrame[], urls: string[]): (url: string) => Promise<WebSocketLike> {
  return async (url) => {
    urls.push(url);
    const listeners: Record<string, ((event: { data: unknown }) => void)[]> = {};
    let index = 0;
    const playNext = () => {
      if (index >= frames.length) return;
      const frame = frames[index++]!;
      const data = frame.type === 'message' ? frame.data : undefined;
      for (const listener of listeners[frame.type] ?? []) listener({ data });
      if (frame.type !== 'close' && frame.type !== 'error') queueMicrotask(playNext);
    };
    setTimeout(playNext, 0);
    return {
      addEventListener(type: string, listener: (event: { data: unknown }) => void) {
        (listeners[type] ??= []).push(listener as never);
      },
      close() {
        for (const listener of listeners.close ?? []) listener({ data: undefined });
      },
    };
  };
}

const CAPTURED = readFileSync(`${import.meta.dirname}/fixtures/jetstream/project-create.json`, 'utf8');

/** The captured project-create frame (see jetstream.test.ts), re-timed and re-keyed. */
function commitFrame(seq: number, timeIso: string): FakeFrame {
  const frame = JSON.parse(CAPTURED);
  Object.assign(frame.payload, { seq, time: timeIso, witnessedAt: timeIso, rkey: `3la${seq}`, cid: `bafy${seq}`, did: 'did:plc:alice' });
  frame.payload.record.createdAt = timeIso;
  return { type: 'message', data: JSON.stringify(frame) };
}

const NOW = Date.parse('2026-11-06T00:00:10.000Z');
const us = (iso: string) => Date.parse(iso) * 1000;

const noRepos = upstream((async () => Response.json({ repos: [] })) as typeof fetch);

function baseDeps(overrides: Partial<IngestDeps> = {}): { model: IndexModel; urls: string[]; deps: IngestDeps } {
  const model = new IndexModel(sqliteSql());
  const urls: string[] = [];
  const deps: IngestDeps = {
    jetstreamUrl: 'wss://jetstream.us-east.bsky.network',
    connect: fakeSocket([], urls),
    index: model,
    backfill: {
      relayUrl: 'https://relay1.us-east.bsky.network',
      fetch: noRepos,
      resolvePds: async () => {
        throw new Error('not reached: no repos to resolve');
      },
    },
    now: () => NOW,
    drainBudgetMs: 20,
    caughtUpSkewMs: 5000,
    ...overrides,
  };
  return { model, urls, deps };
}

function cursorParam(url: string | undefined): string | null {
  return url ? new URL(url).searchParams.get('cursor') : null;
}

test('drains the live tail until an event lands within the caught-up skew, then closes', async () => {
  const urls: string[] = [];
  const { model, deps } = baseDeps({
    connect: fakeSocket(
      [
        commitFrame(1, '2026-11-05T12:00:00.000Z'),
        commitFrame(2, '2026-11-06T00:00:08.000Z'),
        commitFrame(3, '2026-11-06T00:00:09.000Z'),
      ],
      urls,
    ),
  });
  model.setCursor(us('2026-11-05T11:00:00.000Z'));

  await ingest(deps);

  assert.deepStrictEqual(model.recent().projects.map((p) => p.uri), [
    'at://did:plc:alice/dev.nibling.project/3la2',
    'at://did:plc:alice/dev.nibling.project/3la1',
  ]);
  assert.equal(model.cursor(), us('2026-11-06T00:00:08.000Z'));
  assert.equal(cursorParam(urls[0]), String(us('2026-11-05T11:00:00.000Z')));
});

test('a backfill records its start as the cursor and replays the live tail from it', async () => {
  let clock = NOW;
  const urls: string[] = [];
  const { model, deps } = baseDeps({
    now: () => clock,
    backfill: {
      relayUrl: 'https://relay1.us-east.bsky.network',
      fetch: upstream((async () => {
        clock += 30_000;
        return Response.json({ repos: [] });
      }) as typeof fetch),
      resolvePds: async () => 'https://example.pds',
    },
    connect: fakeSocket([{ type: 'close' }], urls),
  });

  await ingest(deps);

  assert.equal(model.cursor(), NOW * 1000);
  assert.equal(cursorParam(urls[0]), String(NOW * 1000));
});

test('a backfill prunes rows the repo no longer lists', async () => {
  const { model, deps } = baseDeps({
    backfill: {
      relayUrl: 'https://relay1.us-east.bsky.network',
      fetch: upstream((async (input: string | URL | Request) => {
        const url = new URL(typeof input === 'string' ? input : input.toString());
        if (url.pathname === '/xrpc/com.atproto.sync.listReposByCollection') {
          return Response.json({ repos: url.searchParams.get('collection') === 'dev.nibling.project' ? [{ did: 'did:plc:alice' }] : [] });
        }
        return Response.json({
          records: [
            {
              uri: 'at://did:plc:alice/dev.nibling.project/kept',
              cid: 'bafy-kept',
              value: { $type: 'dev.nibling.project', name: 'Kept', files: [], createdAt: '2026-11-01T00:00:00.000Z' },
            },
          ],
        });
      }) as typeof fetch),
      resolvePds: async () => 'https://alice.pds.example',
    },
  });
  for (const rkey of ['kept', 'deleted-while-down']) {
    model.apply({
      action: 'put',
      collection: 'dev.nibling.project',
      uri: `at://did:plc:alice/dev.nibling.project/${rkey}`,
      cid: 'bafy-old',
      did: 'did:plc:alice',
      record: { name: rkey, createdAt: '2026-10-01T00:00:00.000Z' },
      indexedAt: '2026-10-01T00:00:00.000Z',
    });
  }
  model.apply({
    action: 'put',
    collection: 'dev.nibling.project',
    uri: 'at://did:plc:bob/dev.nibling.project/other-repo',
    cid: 'bafy-bob',
    did: 'did:plc:bob',
    record: { name: 'Bob', createdAt: '2026-10-01T00:00:00.000Z' },
    indexedAt: '2026-10-01T00:00:00.000Z',
  });

  await ingest(deps);

  assert.deepStrictEqual(model.recent().projects.map((p) => p.uri).sort(), [
    'at://did:plc:alice/dev.nibling.project/kept',
    'at://did:plc:bob/dev.nibling.project/other-repo',
  ]);
});

test('a quiet drain moves the cursor up to when the socket opened', async () => {
  let clock = NOW;
  const { model, deps } = baseDeps({ now: () => clock });
  model.setCursor((NOW - 60_000) * 1000);
  const run = ingest(deps);
  await new Promise((r) => setTimeout(r, 5));
  clock += 6000;
  await run;
  assert.equal(model.cursor(), (NOW - 5000) * 1000);
});

test('a quiet drain moves the cursor at most maxQuietAdvanceMs past where it started', async () => {
  let clock = NOW;
  const { model, deps } = baseDeps({ now: () => clock, maxQuietAdvanceMs: 10_000 });
  const start = (NOW - 3_600_000) * 1000;
  model.setCursor(start);
  const run = ingest(deps);
  await new Promise((r) => setTimeout(r, 5));
  clock += 6000;
  await run;
  assert.equal(model.cursor(), start + 10_000 * 1000);
});

test('a drain that errors does not move the cursor', async () => {
  const { model, deps } = baseDeps({ connect: fakeSocket([{ type: 'error' }], []) });
  model.setCursor((NOW - 60_000) * 1000);
  await ingest(deps);
  assert.equal(model.cursor(), (NOW - 60_000) * 1000);
});

test('skips backfill when the stored cursor is still inside the replay window', async () => {
  let backfillRan = false;
  const { model, deps } = baseDeps({
    backfill: {
      relayUrl: 'https://relay1.us-east.bsky.network',
      fetch: upstream((async () => {
        backfillRan = true;
        return Response.json({ repos: [] });
      }) as typeof fetch),
      resolvePds: async () => 'https://example.pds',
    },
    connect: fakeSocket([{ type: 'close' }], []),
  });
  model.setCursor(us('2026-11-06T00:00:00.000Z'));
  await ingest(deps);
  assert.equal(backfillRan, false);
});

test('backfills when the stored cursor is older than the replay window', async () => {
  let backfillRan = false;
  const { model, deps } = baseDeps({
    backfill: {
      relayUrl: 'https://relay1.us-east.bsky.network',
      fetch: upstream((async () => {
        backfillRan = true;
        return Response.json({ repos: [] });
      }) as typeof fetch),
      resolvePds: async () => 'https://example.pds',
    },
    connect: fakeSocket([{ type: 'close' }], []),
  });
  model.setCursor(us('2026-11-04T00:00:00.000Z'));
  await ingest(deps);
  assert.equal(backfillRan, true);
  assert.equal(model.cursor(), NOW * 1000);
});