/**
 * Does `busInterview` actually PARK?
 *
 * celilo#609's whole scope rests on one claim: a command blocked on an
 * unanswered interview stays alive indefinitely and resumes when the question
 * is answered later. If true, #609 needs no command-state serialization — only
 * session lifetime and re-attach. If the command instead dies quietly, #609 is
 * a much larger design.
 *
 * That claim was originally read off the code (`timeoutMs: 0`), which is not
 * the same as watching it happen. This file watches it happen: a real
 * `module update` sweep parks on a breaking-update confirm nobody answers,
 * stays parked, and then resumes and uses the answer when one arrives by event
 * id — the same reply `celilo events reply <id> <value>` emits.
 *
 * The responder here answers `responder.probe` but deliberately ignores the
 * interview, which is exactly the #609 situation: a responder exists (the
 * api-serve bridge), so the fail-fast guard passes, but nobody can decide.
 */

import { afterEach, beforeEach, expect, test } from 'bun:test';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { type Bus, type BusEvent, defineEvents, openBus } from '@celilo/event-bus';
import { handleModuleUpdate } from '../cli/commands/module-update';
import { getDb } from '../db/client';
import { modules } from '../db/schema';
import { resetTestDbPath } from '../test-utils/db-path';
import { RESPONDER_PROBE_EVENT } from './responder-probe';

const NO_SCHEMAS = defineEvents({});
const QUERY_TYPE = 'interview.required.module-upgrade:iptables.apply_breaking';

let dir: string;
let server: ReturnType<typeof Bun.serve>;
let registryUrl: string;

/** A responder that proves liveness but never answers the question. */
function probeOnlyResponder(busDbPath: string): { bus: Bus; close: () => void } {
  const bus = openBus({ dbPath: busDbPath, events: NO_SCHEMAS });
  const watch = bus.watch(RESPONDER_PROBE_EVENT, async (event) => {
    if (event.replyFor !== null) return;
    bus.emitRaw(
      `${event.type}.reply`,
      { kind: 'daemon', emittedBy: 'park-test' },
      { replyFor: event.id, emittedBy: 'park-test' },
    );
  });
  return {
    bus,
    close: () => {
      watch.close();
      bus.close();
    },
  };
}

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

beforeEach(() => {
  dir = mkdtempSync(join(tmpdir(), 'celilo-park-'));
  process.env.CELILO_DB_PATH = join(dir, 'test.db');
  process.env.CELILO_ORIGINAL_CWD = dir;
  process.env.EVENT_BUS_DB = join(dir, 'events.db');

  getDb()
    .insert(modules)
    .values({
      id: 'iptables',
      name: 'iptables',
      sourcePath: join(dir, 'installed'),
      version: '1.0.2+9',
      manifestData: { celilo_contract: '1.0', id: 'iptables', name: 'iptables', version: '1.0.2' },
    })
    .run();

  server = Bun.serve({
    port: 0,
    fetch(req) {
      if (new URL(req.url).pathname === '/index/ip/ta/iptables') {
        return new Response(
          `${JSON.stringify({ name: 'iptables', vers: '2.0.0+1', deps: [], cksum: 'x' })}\n`,
        );
      }
      return new Response('not found', { status: 404 });
    },
  });
  registryUrl = `http://localhost:${server.port}`;
});

afterEach(() => {
  server.stop(true);
  rmSync(dir, { recursive: true, force: true });
  resetTestDbPath();
  delete process.env.CELILO_ORIGINAL_CWD;
  delete process.env.EVENT_BUS_DB;
});

test('a command parks on an unanswered interview, then resumes with an answer given later', async () => {
  const responder = probeOnlyResponder(join(dir, 'events.db'));
  const observer = openBus({ dbPath: join(dir, 'events.db'), events: NO_SCHEMAS });

  try {
    // Start the sweep but do NOT await it — it should block on the confirm.
    let settled = false;
    const sweep = handleModuleUpdate([], { registry: registryUrl }).then((r) => {
      settled = true;
      return r;
    });

    // 1. It parks. The sweep resolves in milliseconds if it does NOT park, so
    //    any interval well past the 250ms bus poll proves the point — no reason
    //    to hold a CI runner for seconds to say it.
    await sleep(600);
    expect(settled).toBe(false);

    // 2. The question is on the bus, unanswered, and identifiable by event id.
    const queries = observer.recentEvents({ type: QUERY_TYPE });
    expect(queries.length).toBe(1);
    const query = queries[0] as BusEvent;
    const replies = observer.recentEvents({ type: `${QUERY_TYPE}.reply` });
    expect(replies.length).toBe(0);

    // 3. Answer it out-of-band, exactly as `celilo events reply <id> false` does.
    observer.emitRaw(
      `${QUERY_TYPE}.reply`,
      { value: false },
      { replyFor: query.id, emittedBy: 'claude-config-responder' },
    );

    // 4. It resumes AND uses the answer: `false` is a genuine decline, so the
    //    summary must say declined — not "NOT declined", which is what we'd see
    //    if it had failed rather than parked.
    const result = await sweep;
    const report = result.success ? (result.message ?? '') : (result.error ?? '');
    expect(report).toContain('operator declined');
    expect(report).not.toContain('NOT declined');
    expect(result.success).toBe(true);
  } finally {
    observer.close();
    responder.close();
  }
}, 30_000);

/**
 * The instrument the original report reached for cannot see this question.
 *
 * `celilo events list-pending` is `bus.pendingDeliveries()` — it reads the
 * `deliveries` table (subscriber fan-out), while an unanswered interview is a
 * row in `events` awaiting a correlated reply. So "list-pending returned []"
 * was never evidence about the interview either way. This pins that down so
 * #609 builds its observability gate on something that can actually observe.
 */
test('events list-pending cannot see a parked interview — it reads a different table', async () => {
  const responder = probeOnlyResponder(join(dir, 'events.db'));
  const observer = openBus({ dbPath: join(dir, 'events.db'), events: NO_SCHEMAS });

  try {
    const sweep = handleModuleUpdate([], { registry: registryUrl });
    await sleep(600);

    // The question is genuinely there...
    const queries = observer.recentEvents({ type: QUERY_TYPE });
    expect(queries.length).toBe(1);

    // ...and list-pending shows nothing, because it is looking elsewhere.
    expect(observer.pendingDeliveries({ limit: 100 })).toHaveLength(0);

    observer.emitRaw(
      `${QUERY_TYPE}.reply`,
      { value: false },
      { replyFor: queries[0].id, emittedBy: 'park-test' },
    );
    await sweep;
  } finally {
    observer.close();
    responder.close();
  }
}, 30_000);