/**
 * Parked-session registry for `api-serve` (celilo#609).
 *
 * A command that parks on an unanswerable question must outlive the ssh session
 * that started it, so a *different* responder can answer later and the caller
 * can come back for the outcome. Two consequences shape this file:
 *
 *  1. **The registry is on disk, not in memory.** Each ssh connection is its own
 *     `api-serve` process, so the process that later handles `attach` is not the
 *     one holding the parked child. A directory under the data dir is the
 *     handoff: metadata in `session.json`, the command's output appended to
 *     `output.ndjson`. The attaching process replays the file and tails it.
 *  2. **The record is retained after the session ends.** A repeatedly-parking
 *     command is only detectable if the abandonments are still there to see —
 *     the same reason `module operations` keeps its rows.
 *
 * ponytail: files + polling rather than a socket/IPC server. Same process model,
 * a fraction of the machinery, and the buffered output has to be durable across
 * processes anyway. Move to a socket only if attach latency becomes a problem.
 */

import {
  appendFileSync,
  existsSync,
  mkdirSync,
  readFileSync,
  readdirSync,
  writeFileSync,
} from 'node:fs';
import { join } from 'node:path';
import type { ServerMessage } from '@celilo/core';
import { defineEvents, openBus } from '@celilo/event-bus';
import { z } from 'zod';
import { getDataDir } from '../config/paths';

const NO_SCHEMAS = defineEvents({});

/** How long a parked session may hold its child before the reaper abandons it. */
export const DEFAULT_SESSION_TTL_MS = 30 * 60 * 1000;

/** Lines of buffered output an attaching client is replayed at most. */
const REPLAY_LIMIT = 2000;

export const SessionStateSchema = z.enum(['running', 'parked', 'finished', 'abandoned']);
export type SessionState = z.infer<typeof SessionStateSchema>;

export const SessionRecordSchema = z.object({
  sessionId: z.string(),
  principal: z.string(),
  argv: z.array(z.string()),
  startedAt: z.number(),
  expiresAt: z.number(),
  state: SessionStateSchema,
  /** Bus event id of the question this session is parked on, when parked. */
  parkedEventId: z.string().nullable(),
  /** Bus event *type* of that question — what the reaper replies to. */
  parkedEventType: z.string().nullable(),
  question: z.string().nullable(),
  /** The parked question's `<scope>.<key>`. */
  questionKey: z.string().nullable(),
});
export type SessionRecord = z.infer<typeof SessionRecordSchema>;

export function getSessionsDir(): string {
  return join(getDataDir(), 'api-sessions');
}

function sessionDir(sessionId: string): string {
  return join(getSessionsDir(), sessionId);
}

function recordPath(sessionId: string): string {
  return join(sessionDir(sessionId), 'session.json');
}

function outputPath(sessionId: string): string {
  return join(sessionDir(sessionId), 'output.ndjson');
}

export function readSession(sessionId: string): SessionRecord | null {
  const path = recordPath(sessionId);
  if (!existsSync(path)) return null;
  // Untrusted only in the sense of "written by another process" — validate it
  // rather than trusting the shape (Rule 3.7).
  const parsed = SessionRecordSchema.safeParse(JSON.parse(readFileSync(path, 'utf-8')));
  return parsed.success ? parsed.data : null;
}

export function listSessions(): SessionRecord[] {
  const dir = getSessionsDir();
  if (!existsSync(dir)) return [];
  return readdirSync(dir)
    .map(readSession)
    .filter((s): s is SessionRecord => s !== null)
    .sort((a, b) => b.startedAt - a.startedAt);
}

/** The session parked on a given bus query, if one is. */
export function sessionParkedOn(eventId: string): SessionRecord | null {
  return listSessions().find((s) => s.state === 'parked' && s.parkedEventId === eventId) ?? null;
}

function write(record: SessionRecord): void {
  mkdirSync(sessionDir(record.sessionId), { recursive: true });
  writeFileSync(recordPath(record.sessionId), `${JSON.stringify(record, null, 2)}\n`);
}

/**
 * A live session's handle: the writer side of the registry. Held by the
 * `api-serve` process running the command.
 */
export class SessionWriter {
  private record: SessionRecord;

  private constructor(record: SessionRecord) {
    this.record = record;
    write(record);
  }

  static create(opts: {
    principal: string;
    argv: string[];
    ttlMs?: number;
    now?: number;
  }): SessionWriter {
    const now = opts.now ?? Date.now();
    return new SessionWriter({
      sessionId: crypto.randomUUID(),
      principal: opts.principal,
      argv: opts.argv,
      startedAt: now,
      expiresAt: now + (opts.ttlMs ?? DEFAULT_SESSION_TTL_MS),
      state: 'running',
      parkedEventId: null,
      parkedEventType: null,
      question: null,
      questionKey: null,
    });
  }

  get id(): string {
    return this.record.sessionId;
  }

  get current(): SessionRecord {
    return this.record;
  }

  /** Append a message to the buffer so a client attaching later sees it. */
  append(msg: ServerMessage): void {
    appendFileSync(outputPath(this.record.sessionId), `${JSON.stringify(msg)}\n`);
  }

  park(opts: { eventId: string; eventType: string; question: string; questionKey?: string }): void {
    this.record = {
      ...this.record,
      state: 'parked',
      parkedEventId: opts.eventId,
      parkedEventType: opts.eventType,
      question: opts.question,
      questionKey: opts.questionKey ?? null,
    };
    write(this.record);
  }

  /** The question was answered (by anyone) — the command is running again. */
  unpark(): void {
    if (this.record.state !== 'parked') return;
    this.record = {
      ...this.record,
      state: 'running',
      parkedEventId: null,
      parkedEventType: null,
      question: null,
      questionKey: null,
    };
    write(this.record);
  }

  /**
   * `abandoned` outranks `finished`: the command exits *because* it was
   * abandoned, and that exit must not overwrite why. Retained either way.
   */
  finish(state: 'finished' | 'abandoned'): void {
    if (this.record.state === 'abandoned') return;
    this.record = { ...this.record, state };
    write(this.record);
  }
}

/**
 * Answer a parked session's outstanding query as `abandoned` and retain the
 * record as history.
 *
 * The reply is what releases everything the parked command holds: the child
 * resumes, throws `InterviewAbandonedError`, and unwinds — releasing its own
 * module-operation lock on the way out, exactly as any other failure does. That
 * is the whole point of answering rather than killing.
 *
 * ponytail: no forced kill if the child ignores its own unwinding. Add one only
 * once a real command is seen to survive an abandoned answer.
 */
export function abandonSession(
  record: SessionRecord,
  opts: { busDbPath: string; reason: string; emittedBy?: string },
): boolean {
  if (record.parkedEventId && record.parkedEventType) {
    const bus = openBus({ dbPath: opts.busDbPath, events: NO_SCHEMAS });
    try {
      // Someone decided it while we were on our way to reap. Abandoning now
      // would overwrite a real answer with "nobody decided" — the exact
      // fabrication this whole change exists to remove, only in reverse.
      if (bus.repliesFor(Number(record.parkedEventId)).length > 0) return false;
      bus.emitRaw(
        `${record.parkedEventType}.reply`,
        { abandoned: { reason: opts.reason } },
        {
          replyFor: Number(record.parkedEventId),
          emittedBy: opts.emittedBy ?? 'api-session-reaper',
        },
      );
    } finally {
      bus.close();
    }
  }
  write({ ...record, state: 'abandoned' });
  return true;
}

/**
 * Is this session still waiting on a decision *according to the bus*?
 *
 * The record's `state` is the owning process's view and can lag: the answer
 * arrives on the bus from a third party, which the owner never sees. The bus is
 * the source of truth for whether a question stands — ask it, don't infer.
 */
export function stillParked(record: SessionRecord, busDbPath: string): boolean {
  if (record.state !== 'parked' || !record.parkedEventId) return false;
  const bus = openBus({ dbPath: busDbPath, events: NO_SCHEMAS });
  try {
    return bus.repliesFor(Number(record.parkedEventId)).length === 0;
  } finally {
    bus.close();
  }
}

/** Why a reaped session's question was never decided. */
export function expiryReason(record: SessionRecord): string {
  const what = record.question ?? 'the question';
  return `Session ${record.sessionId} expired with "${what}" still unanswered — nobody decided it.`;
}

/**
 * Abandon every parked session past its TTL. Run by the process that owns a
 * parked child on its own timer, and again at `api-serve` startup so a session
 * whose owner died (reboot, OOM) doesn't sit parked forever — the failure mode
 * that left a `module deploy` holding every backup lock on the fleet for 20 days.
 */
export function reapExpiredSessions(opts: { busDbPath: string; now?: number }): SessionRecord[] {
  const now = opts.now ?? Date.now();
  const expired = listSessions().filter((s) => s.state === 'parked' && s.expiresAt <= now);
  for (const record of expired) {
    abandonSession(record, { busDbPath: opts.busDbPath, reason: expiryReason(record) });
  }
  return expired;
}

/** Buffered messages an attaching client should be replayed. */
export function replayOutput(
  sessionId: string,
  fromLine = 0,
): { messages: string[]; next: number } {
  const path = outputPath(sessionId);
  if (!existsSync(path)) return { messages: [], next: fromLine };
  const lines = readFileSync(path, 'utf-8').split('\n').filter(Boolean);
  const start = Math.max(fromLine, lines.length - REPLAY_LIMIT);
  return { messages: lines.slice(start), next: lines.length };
}