// Two-layer cleanup, audit-cleaned per round-2 findings.
//
//   1. Boot sweep: docker rm any tale.sandbox=1 container left over from a
//      previous spawner process, AND host-dir sweep over old session dirs
//      whose mtime is past the watchdog cutoff. The dead "volume sweep"
//      that the original code shipped is gone — workspaces are host bind
//      mounts (no volume), and the cache volumes carry a different label
//      and MUST NOT be reaped.
//   2. Periodic sweep: every 5 min, kill any tale-sbx-* container whose
//      `tale.started=<ms>` label is older than 2× max_timeout AND whose
//      session id isn't in the live in-flight set. Same host-dir sweep
//      for orphan session dirs.
//   3. SIGTERM handler (in server.ts after refactor): stop accepting new
//      requests, wait for in-flight count to drop, then exit.

import {
  mkdir,
  readFile,
  readdir,
  rm,
  stat,
  utimes,
  writeFile,
} from 'node:fs/promises';
import { hostname } from 'node:os';
import { join } from 'node:path';

import type { ExecutionBackend } from './backend/types.ts';
import { runDocker, dockerRm } from './spawn-util.ts';
import { cancelExecution, inFlightIds, isInFlight } from './spawn.ts';
import type { SpawnerConfig } from './types.ts';

const PERIODIC_INTERVAL_MS = 5 * 60_000;
const SPAWNER_LOCK_FILE = '.spawner.lock';
// If an existing lock file is fresher than this, treat the previous spawner
// as still alive and refuse to start. Otherwise we assume the previous
// process crashed without cleanup and take over the lock.
const SPAWNER_LOCK_FRESH_MS = 60_000;
// Refresh the lock's mtime at 1/3 of the freshness window so a peer
// looking for a "fresh" lock always sees one as long as we're alive.
// Without this the lock starts looking stale once we cross the
// freshness threshold and a second spawner would happily reclaim it,
// defeating the lock's only purpose (audit follow-up F15).
const SPAWNER_LOCK_REFRESH_MS = Math.floor(SPAWNER_LOCK_FRESH_MS / 3);
let lockRefreshHandle: ReturnType<typeof setInterval> | undefined;

interface SpawnerLockPayload {
  pid: number;
  hostname: string;
  bootEpoch: number;
}

/**
 * Best-effort cross-process lock for the host session root. Prevents two
 * spawners pointed at the same `/var/lib/tale-sandbox/sessions/` from
 * stomping on each other — specifically, prevents bootSweep's host-dir
 * sweep from deleting another live spawner's in-flight workspace
 * (audit finding R2-B5).
 *
 * Lock contract: if a fresh lock (mtime within SPAWNER_LOCK_FRESH_MS)
 * exists, refuse to start. Otherwise overwrite. On graceful shutdown the
 * server.ts caller deletes the lock; an ungraceful exit leaves the lock
 * stale and the next start can reclaim it after the freshness window.
 */
export async function acquireSpawnerLock(cfg: SpawnerConfig): Promise<void> {
  await mkdir(cfg.hostSessionRoot, { recursive: true });
  const lockPath = join(cfg.hostSessionRoot, SPAWNER_LOCK_FILE);
  try {
    const st = await stat(lockPath);
    // Clamp to [0, ∞) to defend against backward wall-clock skew (NTP
    // step, VM snapshot resume). A negative `age` would otherwise read
    // as "fresh forever" via the `<` comparison even though the lock
    // hasn't been touched in minutes (audit follow-up F15).
    const age = Math.max(0, Date.now() - st.mtimeMs);
    if (age < SPAWNER_LOCK_FRESH_MS) {
      let existing = '<unreadable>';
      try {
        existing = await readFile(lockPath, 'utf8');
      } catch (err) {
        console.warn(`[sandbox.lock] reading existing lock failed:`, err);
      }
      throw new Error(
        `Another spawner appears to be running at ${cfg.hostSessionRoot} ` +
          `(lock fresh, age=${age}ms): ${existing.trim()}`,
      );
    }
    // Stale lock; fall through to overwrite.
    console.warn(
      `[sandbox.lock] reclaiming stale lock at ${lockPath} (age=${age}ms)`,
    );
  } catch (err) {
    // `code` is a non-standard property only present on NodeJS fs errors; the
    // `instanceof Error` + `'code' in err` guards above prove it exists at
    // runtime, but TS can't narrow to the typed shape, so we read it through a
    // minimal interface.
    const code =
      err instanceof Error && 'code' in err
        ? // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
          (err as { code?: string }).code
        : undefined;
    if (code !== 'ENOENT') {
      // Either the lock-fresh refusal above (rethrow) OR an unexpected error.
      if (err instanceof Error && err.message.startsWith('Another spawner')) {
        throw err;
      }
      console.warn(`[sandbox.lock] stat ${lockPath} failed:`, err);
    }
  }
  const payload: SpawnerLockPayload = {
    pid: process.pid,
    hostname: hostname(),
    bootEpoch: Date.now(),
  };
  await writeFile(lockPath, JSON.stringify(payload));
  // Keep the lock visibly "alive" via mtime refresh while the process
  // runs. Stops a long-running spawner from accidentally looking stale
  // to a peer that started later than SPAWNER_LOCK_FRESH_MS after our
  // initial write.
  if (lockRefreshHandle !== undefined) clearInterval(lockRefreshHandle);
  lockRefreshHandle = setInterval(() => {
    const now = Date.now() / 1000;
    utimes(lockPath, now, now).catch((err) => {
      console.warn(`[sandbox.lock] refresh ${lockPath} failed:`, err);
    });
  }, SPAWNER_LOCK_REFRESH_MS);
  // Don't keep the event loop alive solely to refresh the lock — the
  // shutdown handler will clear this. .unref() avoids a hung-process
  // case if every other timer is cleared.
  lockRefreshHandle.unref?.();
}

/**
 * Drop the lock on graceful shutdown so a fast restart doesn't need to wait
 * out the freshness window. Called by DockerBackend.shutdown().
 */
export async function releaseSpawnerLock(cfg: SpawnerConfig): Promise<void> {
  if (lockRefreshHandle !== undefined) {
    clearInterval(lockRefreshHandle);
    lockRefreshHandle = undefined;
  }
  const lockPath = join(cfg.hostSessionRoot, SPAWNER_LOCK_FILE);
  try {
    await rm(lockPath, { force: true });
  } catch (err) {
    console.warn(`[sandbox.lock] release ${lockPath} failed:`, err);
  }
}

async function listLabeledContainers(label: string): Promise<string[]> {
  const result = await runDocker(['ps', '-aq', '-f', `label=${label}`]);
  if (result.exitCode !== 0) return [];
  return result.stdout
    .split('\n')
    .map((s) => s.trim())
    .filter((s) => s.length > 0);
}

async function sweepHostSessionDirs(
  cfg: SpawnerConfig,
  staleThreshold: number,
): Promise<number> {
  let entries;
  try {
    entries = await readdir(cfg.hostSessionRoot, { withFileTypes: true });
  } catch (err) {
    // Root not yet created (first boot) — fine.
    if (err instanceof Error && 'code' in err && err.code === 'ENOENT') {
      return 0;
    }
    console.warn(
      `[sandbox.cleanup] failed to read host session root ${cfg.hostSessionRoot}:`,
      err,
    );
    return 0;
  }
  let removed = 0;
  for (const e of entries) {
    if (!e.isDirectory()) continue;
    if (isInFlight(e.name)) continue;
    const abs = join(cfg.hostSessionRoot, e.name);
    let st;
    try {
      st = await stat(abs);
    } catch (err) {
      console.warn(`[sandbox.cleanup] stat ${abs} failed:`, err);
      continue;
    }
    if (st.mtimeMs >= staleThreshold) continue;
    try {
      await rm(abs, { recursive: true, force: true });
      removed += 1;
    } catch (err) {
      console.warn(`[sandbox.cleanup] rm ${abs} failed:`, err);
    }
  }
  return removed;
}

export async function bootSweep(cfg?: SpawnerConfig): Promise<void> {
  const containers = await listLabeledContainers('tale.sandbox=1');
  for (const c of containers) {
    try {
      await dockerRm(c);
    } catch (err) {
      console.warn(`[sandbox.bootSweep] dockerRm ${c} failed:`, err);
    }
  }
  const stagingContainers = await listLabeledContainers(
    'tale.sandbox-staging=1',
  );
  for (const c of stagingContainers) {
    try {
      await dockerRm(c);
    } catch (err) {
      console.warn(`[sandbox.bootSweep] dockerRm staging ${c} failed:`, err);
    }
  }
  let dirsRemoved = 0;
  if (cfg) {
    // Belt-and-braces: even with the acquireSpawnerLock guarantee above
    // that no other live spawner shares this hostSessionRoot, use the
    // same `2 × maxTimeoutMs` staleness cutoff as the periodic sweep.
    // Dirs younger than that may belong to a recently-killed previous
    // spawner whose in-flight workspace was reaped along with its
    // container; nothing references them anymore so they're safe to
    // delete, but the conservative cutoff matches the rest of the code
    // path's contract and is robust under any future change where the
    // lock acquire is loosened (audit finding R2-B5).
    dirsRemoved = await sweepHostSessionDirs(
      cfg,
      Date.now() - 2 * cfg.maxTimeoutMs,
    );
  }
  if (containers.length > 0 || dirsRemoved > 0) {
    console.log(
      `[sandbox] boot sweep removed ${containers.length} container(s) and ${dirsRemoved} session dir(s)`,
    );
  }
}

/**
 * Docker-specific orphan reap: kill any `tale-sbx-*` container whose
 * `tale.started` label predates `staleThreshold` and whose session id is no
 * longer live, then sweep orphaned host session dirs. Called by
 * `DockerBackend.sweepOrphans` (boot + periodic). Returns the count removed
 * (containers + dirs); errors are logged, never thrown, so the periodic
 * scheduler keeps running.
 */
export async function dockerSweepOrphans(
  cfg: SpawnerConfig,
  staleThreshold: number,
  isLive: (executionId: string) => boolean,
): Promise<number> {
  let removed = 0;
  // Match the prior startPeriodicSweep semantics: a failed/throwing `docker
  // ps` short-circuits the whole tick (neither the container loop NOR the
  // host-dir sweep runs), so we don't reap host session dirs while the daemon
  // is unreachable.
  let containerProbeOk = false;
  try {
    const result = await runDocker([
      'ps',
      '-a',
      '--filter',
      'label=tale.sandbox=1',
      '--format',
      '{{.Names}}\t{{.Labels}}',
    ]);
    if (result.exitCode === 0) {
      containerProbeOk = true;
      for (const line of result.stdout.split('\n')) {
        const [name, labels] = line.split('\t');
        if (!name) continue;
        const m = labels?.match(/tale\.started=(\d+)/);
        if (!m) continue;
        const started = Number.parseInt(m[1] ?? '0', 10);
        if (Number.isNaN(started) || started >= staleThreshold) continue;
        // session id is the second component of the name (tale-sbx-<id>).
        const sessionId = name.replace(/^tale-sbx-/, '');
        if (isLive(sessionId)) continue;
        try {
          await dockerRm(name);
        } catch (err) {
          console.warn(
            `[sandbox.periodic] dockerRm stale ${name} failed:`,
            err,
          );
          continue;
        }
        removed += 1;
        console.log(
          `[sandbox] periodic sweep removed stale container ${name} (started ${new Date(started).toISOString()})`,
        );
      }
    }
  } catch (err) {
    console.warn(`[sandbox.periodic] container sweep error:`, err);
  }
  // Host-dir sweep: per-execution session dirs that lived past the stale
  // threshold without an active in-flight entry are orphaned. Replaces the
  // old volume-sweep block that targeted volumes nobody creates (audit
  // finding R2-3 C5). Gated on the container probe so a wedged daemon defers
  // dir reaping to the next cycle (matches the prior short-circuit).
  if (containerProbeOk) {
    removed += await sweepHostSessionDirs(cfg, staleThreshold);
  }
  return removed;
}

/**
 * Generic periodic-sweep scheduler. Backend-agnostic: every 5 min it asks the
 * active backend to reap orphans (DockerBackend → `dockerSweepOrphans`; a
 * future KubernetesBackend → a label-selector Pod delete).
 */
export function startPeriodicSweep(
  backend: ExecutionBackend,
  cfg: SpawnerConfig,
): () => void {
  const interval = setInterval(() => {
    void backend
      .sweepOrphans({
        staleBeforeMs: Date.now() - 2 * cfg.maxTimeoutMs,
        isLive: isInFlight,
      })
      .catch((err) => {
        console.warn(`[sandbox.periodic] sweep error:`, err);
      });
  }, PERIODIC_INTERVAL_MS);
  return () => clearInterval(interval);
}

/**
 * Graceful shutdown handler.
 *
 * The original code called `process.exit(0)` immediately after issuing
 * `docker kill` for every in-flight id — but `executeRequest`'s finally
 * block (which rm -rfs the host session dir) was racing with the exit,
 * so SIGTERM mid-execution leaked the host workspace. The new flow:
 *
 *   1. Mark "draining" so the HTTP layer stops accepting new work
 *      (callers pass the stop callback in).
 *   2. Issue `cancelExecution` for every in-flight id; this aborts the
 *      runDocker subprocess via AbortSignal and lets each
 *      `executeRequest` proceed to its finally block.
 *   3. Wait (with a 20s ceiling) for the in-flight Map to drain.
 *   4. exit().
 */
export function installSignalHandlers(
  stopAccepting: () => void,
  backend: ExecutionBackend,
): void {
  let shuttingDown = false;
  const onTerm = async (sig: string) => {
    if (shuttingDown) {
      console.warn(`[sandbox] received second ${sig}; forcing exit`);
      process.exit(1);
    }
    shuttingDown = true;
    console.log(`[sandbox] received ${sig}; draining in-flight executions`);
    try {
      stopAccepting();
    } catch (err) {
      console.warn(`[sandbox.shutdown] stopAccepting failed:`, err);
    }
    // Abort-only: each execute() observes the abort and performs its own
    // teardown (kill/remove container, delete Pod+Secret); the drain loop
    // below waits on those finally blocks via the in-flight registry.
    for (const id of inFlightIds()) {
      cancelExecution(id);
    }
    const deadline = Date.now() + 20_000;
    while (inFlightIds().length > 0 && Date.now() < deadline) {
      await new Promise<void>((resolve) => setTimeout(resolve, 200));
    }
    const remaining = inFlightIds();
    if (remaining.length > 0) {
      console.warn(
        `[sandbox] shutdown deadline; ${remaining.length} execution(s) still in-flight (${remaining.join(', ')})`,
      );
    }
    await backend.shutdown();
    process.exit(0);
  };
  process.on('SIGTERM', () => void onTerm('SIGTERM'));
  process.on('SIGINT', () => void onTerm('SIGINT'));
}
