/**
 * Remote API server — `celilo api-serve --principal=<id>` (Slices 1–3).
 *
 * Invoked as the sshd forced command for an enrolled key, so it starts already
 * authenticated as `principal`. Reads `command`/`answer`/`unanswerable`/
 * `attach`/`cancel` messages as NDJSON on stdin/stdout, authorizes each command
 * against the principal's grants (deny-by-default), and — if allowed — runs it
 * as a child `celilo` process, translating the child's protocol-mode output into
 * `progress`/`log` and a terminal `result`. While a command runs, a remote
 * responder bridges the event bus to the wire so mid-run `interview`s are
 * answered by the client. Every attempt is audited to stderr.
 *
 * When the client cannot decide a question, the command **parks**: the query is
 * left unanswered on the bus, the child stays alive past the end of the ssh
 * session, its output is buffered in the session registry, and the client is
 * told `blocked` with the session and query ids (celilo#609). Another responder
 * answers by event id; a later `attach` collects the outcome.
 */

import { createInterface } from 'node:readline';
import {
  API_PROTOCOL_VERSION,
  ClientMessageSchema,
  type ServerMessage,
  ServerMessageSchema,
  translateOutputLine,
} from '@celilo/core';
import { parseArguments } from '../cli/parser';
import { getEventBusPath } from '../config/paths';
import { getDb } from '../db/client';
import { isAuthorized } from '../services/api-access';
import { EVENT_TYPES } from '../services/bus-interview';
import { describePausedModule, listPausedModules } from '../services/module-pause';
import { type WireInterview, startRemoteResponder } from '../services/remote-responder';
import {
  SessionWriter,
  abandonSession,
  expiryReason,
  readSession,
  reapExpiredSessions,
  replayOutput,
  stillParked,
} from './sessions';

/** Exit code returned to the client when authz denies a command. */
const EXIT_PERMISSION_DENIED = 126;

/** How often an attached client polls a session's buffer for new output. */
const ATTACH_POLL_MS = 250;

function send(msg: ServerMessage): void {
  process.stdout.write(`${JSON.stringify(msg)}\n`);
}

/**
 * The fleet-level warnings stamped onto every terminal result (design D7).
 *
 * Deliberately unconditional and deliberately on UNRELATED commands: a pause
 * suppresses its module's alerting, so the paused-ness itself is the only
 * remaining signal, and a forgotten pause is found incidentally rather than by
 * someone choosing to look. One indexed query (`modules_state_idx`), so this
 * stays cheap enough to run on every call.
 *
 * Never throws: a broken or missing DB must not turn a working command into a
 * failure over a warning.
 */
function fleetWarnings(): string[] {
  try {
    const paused = listPausedModules(getDb());
    if (paused.length === 0) return [];
    return [
      `${paused.length} module(s) PAUSED: ${paused.map((m) => describePausedModule(m)).join(', ')}. A paused module is quiesced and its alerts are suppressed. Unpause with "celilo module unpause <id>".`,
    ];
  } catch {
    return [];
  }
}

/** Terminal result, with the fleet warnings attached. */
function resultMessage(success: boolean, exitCode: number): ServerMessage {
  const warnings = fleetWarnings();
  return warnings.length > 0
    ? { type: 'result', success, exitCode, warnings }
    : { type: 'result', success, exitCode };
}

function audit(principal: string, op: string, decision: string, exitCode?: number): void {
  const suffix = exitCode === undefined ? '' : ` exit=${exitCode}`;
  process.stderr.write(
    `[api-audit] principal=${principal} op=${op} decision=${decision}${suffix}\n`,
  );
}

/** Derive the `command`/`subcommand` an argv would run, via the real parser. */
function opOf(argv: string[]): { command: string; subcommand?: string; label: string } {
  const parsed = parseArguments(['bun', 'celilo', ...argv]);
  const label = parsed.subcommand ? `${parsed.command}:${parsed.subcommand}` : parsed.command;
  return { command: parsed.command, subcommand: parsed.subcommand, label };
}

/**
 * Emit a message to the attached client (if any) AND to the session buffer, so
 * output produced while nobody is attached is not lost.
 */
function emit(session: SessionWriter, msg: ServerMessage, attached: () => boolean): void {
  session.append(msg);
  if (attached()) send(msg);
}

/** Drain a child stream line-by-line, forwarding each line as a message. */
async function pumpLines(
  stream: ReadableStream<Uint8Array>,
  forward: (msg: ServerMessage) => void,
): Promise<void> {
  const decoder = new TextDecoder();
  let buffer = '';
  for await (const chunk of stream) {
    buffer += decoder.decode(chunk, { stream: true });
    let nl = buffer.indexOf('\n');
    while (nl >= 0) {
      const line = buffer.slice(0, nl);
      buffer = buffer.slice(nl + 1);
      forward(translateOutputLine(line));
      nl = buffer.indexOf('\n');
    }
  }
  if (buffer.length > 0) {
    forward(translateOutputLine(buffer));
  }
}

/** Run an authorized command as a child; returns its exit code. */
async function runCommand(argv: string[], forward: (msg: ServerMessage) => void): Promise<number> {
  // Re-invoke this same CLI as a child. The child is non-TTY (piped stdout) so
  // ProgressDisplay resolves to protocol mode and emits `[progress:*]` markers.
  const child = Bun.spawn([process.execPath, Bun.main, ...argv], {
    stdin: 'ignore',
    stdout: 'pipe',
    stderr: 'pipe',
  });

  // Each stream is pumped separately and the log message names its origin,
  // so a client can keep stderr distinguishable from the command's result
  // (celilo#1362 — the two used to merge into one undifferentiated channel).
  const markStream =
    (stream: 'stdout' | 'stderr') =>
    (msg: ServerMessage): void => {
      forward(msg.type === 'log' ? { ...msg, stream } : msg);
    };

  await Promise.all([
    pumpLines(child.stdout, markStream('stdout')),
    pumpLines(child.stderr, markStream('stderr')),
  ]);
  const exitCode = await child.exited;
  forward(resultMessage(exitCode === 0, exitCode));
  return exitCode;
}

/**
 * Replay a parked session's buffered output to the attaching client, then tail
 * it until the session goes terminal. Deny-by-default on the owning principal.
 */
async function handleAttach(principal: string, sessionId: string): Promise<void> {
  const record = readSession(sessionId);
  if (!record) {
    send({ type: 'error', error: `no such session: ${sessionId}` });
    send(resultMessage(false, 1));
    return;
  }
  if (record.principal !== principal) {
    send({
      type: 'error',
      error: `permission denied: session ${sessionId} belongs to another principal`,
    });
    send(resultMessage(false, EXIT_PERMISSION_DENIED));
    audit(principal, `attach:${sessionId}`, 'deny');
    return;
  }
  audit(principal, `attach:${sessionId}`, 'allow');

  let cursor = 0;
  for (;;) {
    const { messages, next } = replayOutput(sessionId, cursor);
    cursor = next;
    for (const line of messages) {
      const parsed = ServerMessageSchema.safeParse(JSON.parse(line));
      if (!parsed.success) continue;
      send(parsed.data);
      if (parsed.data.type === 'result') return;
    }
    const current = readSession(sessionId);
    if (!current) return;
    // Still waiting on a decision → say so rather than tailing forever. Asked of
    // the bus, not of the record: the answer arrives from a third party the
    // owning process never hears from, so its `state` lags.
    if (current.parkedEventId && stillParked(current, getEventBusPath())) {
      send({
        type: 'blocked',
        sessionId,
        eventId: current.parkedEventId,
        question: current.question ?? '',
        key: current.questionKey ?? undefined,
      });
      return;
    }
    await new Promise((r) => setTimeout(r, ATTACH_POLL_MS));
  }
}

/** Abandon a parked session on the operator's say-so, before its TTL. */
function handleCancel(principal: string, sessionId: string): void {
  const record = readSession(sessionId);
  if (!record) {
    send({ type: 'error', error: `no such session: ${sessionId}` });
    send(resultMessage(false, 1));
    return;
  }
  if (record.principal !== principal) {
    send({
      type: 'error',
      error: `permission denied: session ${sessionId} belongs to another principal`,
    });
    send(resultMessage(false, EXIT_PERMISSION_DENIED));
    audit(principal, `cancel:${sessionId}`, 'deny');
    return;
  }
  abandonSession(record, {
    busDbPath: getEventBusPath(),
    reason: `Session ${sessionId} was cancelled by ${principal} with "${record.question ?? 'the question'}" still unanswered.`,
    emittedBy: `api:${principal}`,
  });
  audit(principal, `cancel:${sessionId}`, 'allow');
  send(resultMessage(true, 0));
}

export async function apiServeMode(principal: string): Promise<void> {
  send({ type: 'ready', protocolVersion: API_PROTOCOL_VERSION });

  const busDbPath = getEventBusPath();
  // A session whose owning process died (reboot, OOM) would otherwise stay
  // parked forever, holding whatever its command holds.
  reapExpiredSessions({ busDbPath });

  // When the ssh client goes away, sshd hangs up its forced command. Ignoring
  // SIGHUP is what lets a parked command outlive the session that started it —
  // the TTL reaper, not the transport, is what bounds it now.
  process.on('SIGHUP', () => {});

  const pendingAnswers = new Map<
    string,
    { resolve: (value: unknown) => void; reject: (error: Error) => void; interview: WireInterview }
  >();

  /** Set while a command is running; the parked child outlives the transport. */
  const live: { session: SessionWriter | null; clientAttached: boolean } = {
    session: null,
    clientAttached: true,
  };
  let commandRunning: Promise<void> | null = null;

  const ask = (interview: WireInterview): Promise<unknown> =>
    new Promise((resolve, reject) => {
      pendingAnswers.set(interview.id, { resolve, reject, interview });
      if (!live.clientAttached) {
        // Nobody to ask. Re-park on THIS question: a detached command that was
        // answered and moved on is now waiting on a *different* event, and a
        // record still naming the previous one is actively wrong — it hides the
        // live question from `events list-unanswered`, and it aims the TTL
        // reaper at a question that was already decided, so the reaper retires
        // the session while the child is still alive and blocked.
        live.session?.park({
          eventId: interview.id,
          eventType: EVENT_TYPES.interviewRequired(interview.scope, interview.key),
          question: interview.message,
          questionKey: `${interview.scope}.${interview.key}`,
        });
        return;
      }
      send({
        type: 'interview',
        id: interview.id,
        scope: interview.scope,
        key: interview.key,
        kind: interview.kind,
        message: interview.message,
        description: interview.description,
        defaultValue: interview.defaultValue,
        placeholder: interview.placeholder,
        options: interview.options,
        required: interview.required,
      });
    });

  const handleCommand = async (argv: string[]): Promise<void> => {
    const { command, subcommand, label } = opOf(argv);

    if (!(await isAuthorized(principal, command, subcommand))) {
      send({ type: 'error', error: `permission denied: "${principal}" is not granted "${label}"` });
      send(resultMessage(false, EXIT_PERMISSION_DENIED));
      audit(principal, label, 'deny');
      return;
    }

    const writer = SessionWriter.create({ principal, argv });
    live.session = writer;
    const reaper = setTimeout(
      () => {
        if (writer.current.state !== 'parked') return;
        abandonSession(writer.current, {
          busDbPath,
          reason: expiryReason(writer.current),
          emittedBy: `api:${principal}`,
        });
        writer.finish('abandoned');
      },
      Math.max(0, writer.current.expiresAt - Date.now()),
    );

    // Bridge the bus to the wire so mid-run interviews reach the client.
    const responder = startRemoteResponder({ busDbPath, ask, emittedBy: `api:${principal}` });
    try {
      const exitCode = await runCommand(argv, (msg) =>
        emit(writer, msg, () => live.clientAttached),
      );
      writer.finish('finished');
      audit(principal, label, 'allow', exitCode);
    } finally {
      clearTimeout(reaper);
      responder.close();
      live.session = null;
    }
  };

  const rl = createInterface({ input: process.stdin, terminal: false });

  for await (const line of rl) {
    if (!line.trim()) continue;

    let msg: ReturnType<typeof ClientMessageSchema.parse>;
    try {
      msg = ClientMessageSchema.parse(JSON.parse(line));
    } catch (error) {
      send({ type: 'error', error: error instanceof Error ? error.message : String(error) });
      continue;
    }

    if (msg.type === 'answer') {
      const pending = pendingAnswers.get(msg.id);
      if (pending) {
        pendingAnswers.delete(msg.id);
        live.session?.unpark();
        pending.resolve(msg.value);
      }
      continue;
    }

    if (msg.type === 'unanswerable') {
      // The client cannot decide. Do NOT resolve the question — park on it and
      // tell the client where it stands. The responder, seeing `ask` reject,
      // emits nothing on the bus, so the query is still there to be answered.
      const pending = pendingAnswers.get(msg.id);
      if (!pending) continue;
      pendingAnswers.delete(msg.id);
      const { interview } = pending;
      pending.reject(new Error(msg.reason));
      if (live.session) {
        live.session.park({
          eventId: msg.id,
          eventType: EVENT_TYPES.interviewRequired(interview.scope, interview.key),
          question: interview.message,
          questionKey: `${interview.scope}.${interview.key}`,
        });
        send({
          type: 'blocked',
          sessionId: live.session.id,
          eventId: msg.id,
          question: interview.message,
          key: `${interview.scope}.${interview.key}`,
        });
      }
      continue;
    }

    if (msg.type === 'attach') {
      await handleAttach(principal, msg.sessionId);
      continue;
    }

    if (msg.type === 'cancel') {
      handleCancel(principal, msg.sessionId);
      continue;
    }

    if (msg.type === 'command') {
      // Fire-and-forget so the read loop keeps consuming `answer` messages while
      // the command runs — mid-run interviews are answered in-flight.
      commandRunning = handleCommand(msg.argv);
      void commandRunning;
    }
  }

  // stdin closed: the client is gone. A parked command must NOT die with it —
  // that is the bug. Stay alive (output goes to the session buffer) until the
  // command finishes or the reaper abandons it.
  live.clientAttached = false;
  if (commandRunning) await commandRunning;
  process.exit(0);
}