/**
 * Does the reaper actually fire, and does the command say the right thing?
 *
 * A parked session holds a live child process and possibly a module-operation
 * lock. The TTL reaper is the only thing standing between that and the 20-day
 * lock outage `module operations` was built for — so a reaper nobody has watched
 * fire is decorative. This drives a real `module update` sweep into a park,
 * expires its session deliberately, and asserts on what comes back.
 *
 * What it asserts is the *value*, not liveness. Three outcomes are confusable
 * here and only one is correct:
 *
 *   declined   — someone said no.                    (`{value: false}`)
 *   unanswered — nobody was listening at all.        (responder-probe throw)
 *   abandoned  — the question stood; the deadline passed; nobody decided.
 *
 * Reporting an abandonment as "operator declined" is the original celilo#609
 * bug wearing a different hat, so that string is asserted absent.
 *
 * WATCHED RED: with `abandonSession` emitting `{value: false}` instead of
 * `{abandoned}` — i.e. a reaper that resolves the question the old way — this
 * file fails on `expect(report).not.toContain('operator declined')`. With the
 * reaper removed entirely it fails by timeout, the sweep never resuming.
 */

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 { RESPONDER_PROBE_EVENT } from '../services/responder-probe';
import { resetTestDbPath } from '../test-utils/db-path';
import { SessionWriter, listSessions, reapExpiredSessions } from './sessions';

const NO_SCHEMAS = defineEvents({});
const QUERY_TYPE = 'interview.required.module-upgrade:iptables.apply_breaking';
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

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

/** A responder that proves liveness but never answers the question. */
function probeOnlyResponder(dbPath: string): { bus: Bus; close: () => void } {
  const bus = openBus({ dbPath, 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: 'reaper-test' },
      { replyFor: event.id, emittedBy: 'reaper-test' },
    );
  });
  return {
    bus,
    close: () => {
      watch.close();
      bus.close();
    },
  };
}

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

  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_DATA_DIR;
  delete process.env.CELILO_ORIGINAL_CWD;
  delete process.env.EVENT_BUS_DB;
});

test('an expired session is reaped and the command reports abandoned, not declined', async () => {
  const responder = probeOnlyResponder(busDbPath);
  const observer = openBus({ dbPath: busDbPath, events: NO_SCHEMAS });

  try {
    let settled = false;
    const sweep = handleModuleUpdate([], { registry: registryUrl }).then((r) => {
      settled = true;
      return r;
    });

    await sleep(600);
    expect(settled).toBe(false);

    const query = observer.recentEvents({ type: QUERY_TYPE })[0] as BusEvent;
    expect(query).toBeDefined();

    // A session parked on that question, already past its deadline.
    const session = SessionWriter.create({
      principal: 'tester',
      argv: ['module', 'update'],
      ttlMs: -1,
    });
    session.park({
      eventId: String(query.id),
      eventType: QUERY_TYPE,
      question: 'Apply breaking update for iptables (1.0.2+9 → 2.0.0+1)?',
      questionKey: 'module-upgrade:iptables.apply_breaking',
    });

    const reaped = reapExpiredSessions({ busDbPath });
    expect(reaped.map((r) => r.sessionId)).toEqual([session.id]);

    // The command resumes — and says nobody decided, not that anyone declined.
    const result = await sweep;
    const report = result.success ? (result.message ?? '') : (result.error ?? '');
    expect(report).toContain('Abandoned:');
    expect(report).toContain('never decided');
    expect(report).not.toContain('operator declined');
    // ...and not the *unanswered* wording either: a responder was listening.
    expect(report).not.toContain('could not be answered');
    expect(report).toContain('iptables');
    // A breaking update that silently didn't land must not read as success.
    expect(result.success).toBe(false);
    // The module is untouched — nothing was applied on nobody's authority.
    expect(getDb().select().from(modules).all()[0].version).toBe('1.0.2+9');

    // The record is retained, so a repeatedly-parking command is detectable.
    const retained = listSessions();
    expect(retained).toHaveLength(1);
    expect(retained[0].state).toBe('abandoned');
    expect(retained[0].question).toContain('iptables');
  } finally {
    observer.close();
    responder.close();
  }
}, 30_000);

test('a question answered before the deadline is never overwritten by the reaper', async () => {
  const observer = openBus({ dbPath: busDbPath, events: NO_SCHEMAS });
  try {
    const query = observer.emitRaw(QUERY_TYPE, { scope: 'x', key: 'y' });
    observer.emitRaw(
      `${QUERY_TYPE}.reply`,
      { value: true },
      { replyFor: query.id, emittedBy: 'a-real-decider' },
    );

    const session = SessionWriter.create({
      principal: 'tester',
      argv: ['module', 'update'],
      ttlMs: -1,
    });
    session.park({
      eventId: String(query.id),
      eventType: QUERY_TYPE,
      question: 'Apply breaking update for iptables?',
    });

    reapExpiredSessions({ busDbPath });

    // Exactly one reply, and it is the decision that was actually made.
    const replies = observer.repliesFor(query.id);
    expect(replies).toHaveLength(1);
    expect((replies[0].payload as { value: unknown }).value).toBe(true);
  } finally {
    observer.close();
  }
});