import type { FastifyInstance } from 'fastify';
import type { Pool } from 'pg';
import { afterAll, beforeEach, describe, expect, it } from 'vitest';
import { processEvents } from '../src/pipeline/process.ts';
import { makeApp } from './helpers/app.ts';
import { closeTestPool, getTestPool, resetDb } from './helpers/db.ts';
import { ALICE, BOB, ev } from './helpers/events.ts';

let pool: Pool;
let app: FastifyInstance;
let base: string;

beforeEach(async () => {
  await app?.close();
  pool = await getTestPool();
  await resetDb(pool);
  ({ app } = await makeApp(pool));
  await app.listen({ port: 0, host: '127.0.0.1' });
  const addr = app.server.address();
  base = `http://127.0.0.1:${typeof addr === 'object' && addr ? addr.port : 0}`;
  await app.liveHub.ready();
});
afterAll(async () => {
  await app?.close();
  await closeTestPool();
});

interface Sse {
  id: number;
  event: string;
  data: Record<string, unknown>;
}

/** Opens /api/live and collects parsed SSE messages until `until` returns true. */
async function collect(until: (msgs: Sse[]) => boolean, headers: Record<string, string> = {}, trigger?: () => Promise<void>) {
  const ctrl = new AbortController();
  const res = await fetch(`${base}/api/live`, { headers, signal: ctrl.signal });
  expect(res.headers.get('content-type')).toContain('text/event-stream');
  const reader = res.body!.getReader();
  const msgs: Sse[] = [];
  let buf = '';
  const timeout = setTimeout(() => ctrl.abort(), 8000);
  let triggered = false;
  try {
    for (;;) {
      if (!triggered && trigger) {
        triggered = true;
        // wait for the stream to be registered before triggering
        const first = await reader.read();
        buf += new TextDecoder().decode(first.value);
        await trigger();
      }
      let idx: number;
      while ((idx = buf.indexOf('\n\n')) >= 0) {
        const block = buf.slice(0, idx);
        buf = buf.slice(idx + 2);
        const fields = Object.fromEntries(
          block
            .split('\n')
            .filter((l) => !l.startsWith(':'))
            .map((l) => [l.slice(0, l.indexOf(':')), l.slice(l.indexOf(':') + 1).trim()]),
        );
        if (fields.event) msgs.push({ id: Number(fields.id), event: fields.event, data: JSON.parse(fields.data ?? '{}') });
      }
      if (until(msgs)) return msgs;
      const { value, done } = await reader.read();
      if (done) return msgs;
      buf += new TextDecoder().decode(value);
    }
  } finally {
    clearTimeout(timeout);
    ctrl.abort();
  }
}

const death = (who: typeof ALICE) =>
  ev('player_death', { ...who, dimension: 'minecraft:overworld', cause: 'minecraft:fall', killerType: null, killerName: null });

describe('GET /api/live', () => {
  it('delivers death event after ingest', async () => {
    const msgs = await collect(
      (m) => m.some((x) => x.event === 'death'),
      {},
      async () => {
        await processEvents(pool, [death(ALICE)], 'push', 'atm10');
      },
    );
    const d = msgs.find((m) => m.event === 'death')!;
    expect(d.data).toMatchObject({ uuid: ALICE.uuid, name: ALICE.name, cause: 'minecraft:fall' });
  });

  it('resumes from Last-Event-ID', async () => {
    await processEvents(pool, [ev('player_join', { ...ALICE })], 'push', 'atm10');
    await processEvents(pool, [death(ALICE)], 'push', 'atm10');
    await new Promise((r) => setTimeout(r, 300));
    const all = app.liveHub.since(0);
    expect(all.map((e) => e.kind)).toEqual(['online', 'death']);
    const msgs = await collect((m) => m.length >= 1, { 'last-event-id': String(all[0]!.id) });
    expect(msgs.map((m) => m.event)).toEqual(['death']);
  });

  it('hidden player not broadcast', async () => {
    await processEvents(pool, [ev('player_join', { ...BOB })], 'push', 'atm10');
    await pool.query('UPDATE players SET hidden = true WHERE uuid = $1', [BOB.uuid]);
    const msgs = await collect(
      (m) => m.some((x) => x.event === 'death'),
      { 'last-event-id': String(app.liveHub.lastId()) },
      async () => {
        await processEvents(pool, [death(BOB)], 'push', 'atm10');
        await processEvents(pool, [death(ALICE)], 'push', 'atm10');
      },
    );
    const deaths = msgs.filter((m) => m.event === 'death');
    expect(deaths.map((d) => d.data.uuid)).toEqual([ALICE.uuid]);
  });
});