/**
 * The operator drain sentinel: one JSON file under the state root that says a
 * human asked this fleet to stop taking new work and finish what it has.
 *
 * It is a file rather than a store row because the things that read it are not
 * all the daemon — `omp-conductor drain`, the status render and `upgrade`'s
 * pre-flight all answer without opening the database, and an upgrade that has
 * just replaced the schema must still be able to read the drain it set. That
 * also fixes the boundary: everything here is pure file I/O plus a verdict, no
 * `Deps`, so it is the leaf the dispatch pass and the status projection both
 * read rather than a pass either of them has to import.
 *
 * Not to be confused with `DrainSignal` in `deps.ts`: that one is the in-process
 * "SIGTERM has landed" flag a tick re-checks mid-flight. This is the durable
 * operator intent that outlives the process.
 */
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { stateDir } from "../config.ts";

// (#484 slice 1) A project drain is a durable, self-expiring admission fence:
// the same boundary as a pause — settlement above it, nothing claimed below —
// but recorded with an absolute deadline, so an orchestrator crash can never
// strand admission. The record is a JSON file under the state directory (like
// the pause sentinel), scoped to exactly one configured project, and replaced
// atomically on create. Human CLI wording and release-verb coupling are later
// #484 children.

/** The persisted shape of one project drain. Every field is validated when a
 *  record is read — a record that cannot be trusted is never a fence. */
export interface DrainRecord {
  /** The configured project this drain fences; must match the reader. */
  project: string;
  /** ISO instant the drain intent was recorded. */
  createdAt: string;
  /** Absolute ISO deadline: admission resumes automatically at or after it. */
  expiresAt: string;
  /** ISO instant a dispatch pass first observed an empty active set under
   *  this drain — the moment the release window opened (#1078). Written by
   *  the tick's drain pass, never by the operator; absent until then. */
  drainedAt?: string;
  /** Purpose recorded at creation (release window, maintenance…). */
  reason?: string;
}

/** Create-time options for {@link createDrain}. */
export interface CreateDrainOptions {
  /** Absolute epoch-ms deadline — a drain must always expire on its own. */
  expiresAt: number;
  /** Purpose, persisted on the record and shown in structured status. */
  reason?: string;
}

/** Why a drain record could not be trusted, named deterministically. */
export type DrainProblem =
  | "unparseable-json"
  | "invalid-record"
  | "invalid-project"
  | "invalid-created-at"
  | "invalid-expires-at"
  | "invalid-drained-at"
  | "expiry-not-future"
  | "invalid-reason";

/** The verdict of one drain read. `error` is the malformed-record case: the
 *  caller may fail its pass closed, and the dispatch-side
 *  {@link consumeDrain} removes the record in the same transition, so it can
 *  never become a permanent drain. */
export type DrainVerdict =
  | { kind: "active"; drain: DrainRecord }
  | { kind: "inactive" }
  | { kind: "error"; problem: DrainProblem };

/** Where this project's drain record lives. Project-scoped by construction:
 *  a record present at one project's path never fences another project. */
export function drainPath(project: string): string {
  return join(stateDir(), `drain-${project}.json`);
}

/**
 * Records a bounded drain intent for `project`, replacing any prior drain of
 * the same project atomically (tmp + rename, exactly like the admission ack).
 * The record is a file, so it survives orchestrator and daemon loss; the
 * absolute `expiresAt` is what stops it from ever stranding admission.
 */
export function createDrain(
  project: string,
  opts: CreateDrainOptions,
  now = Date.now(),
): void {
  if (project === "") {
    throw new Error("drain project must not be empty");
  }
  if (!Number.isFinite(opts.expiresAt) || opts.expiresAt <= now) {
    throw new Error("drain expiresAt must be a finite epoch-ms timestamp in the future");
  }
  if (opts.reason !== undefined && typeof opts.reason !== "string") {
    throw new Error("drain reason must be a string");
  }
  const record: DrainRecord = {
    project,
    createdAt: new Date(now).toISOString(),
    expiresAt: new Date(opts.expiresAt).toISOString(),
    ...(opts.reason === undefined ? {} : { reason: opts.reason }),
  };
  const path = drainPath(project);
  mkdirSync(dirname(path), { recursive: true });
  const tmp = `${path}.${process.pid.toString(36)}.${Date.now().toString(36)}.tmp`;
  writeFileSync(tmp, `${JSON.stringify(record, null, 2)}\n`);
  renameSync(tmp, path);
}

/**
 * One drained state read with no side effects — this never touches the record
 * on disk. That purity is what lets observational surfaces live off it: a
 * status/dashboard/health read must not consume a malformed record before a
 * dispatch pass fails closed on it, or the pass would read absent and admit.
 * The dispatch side performs the actual cleanup transition through
 * {@link consumeDrain}. Callers that only need to *see* the state (including
 * the claim path, which refuses but must not unbind its own pass) use this.
 */
export function readDrain(project: string, now = Date.now()): DrainVerdict {
  const path = drainPath(project);
  if (!existsSync(path)) return { kind: "inactive" };
  let raw: unknown;
  try {
    raw = JSON.parse(readFileSync(path, "utf8"));
  } catch {
    return { kind: "error", problem: "unparseable-json" };
  }
  if (raw === null || typeof raw !== "object" || Array.isArray(raw)) {
    return { kind: "error", problem: "invalid-record" };
  }
  const rec = raw as Record<string, unknown>;
  if (typeof rec["project"] !== "string" || rec["project"] === "") {
    return { kind: "error", problem: "invalid-project" };
  }
  if (typeof rec["createdAt"] !== "string" || Number.isNaN(Date.parse(rec["createdAt"]))) {
    return { kind: "error", problem: "invalid-created-at" };
  }
  if (typeof rec["expiresAt"] !== "string" || Number.isNaN(Date.parse(rec["expiresAt"]))) {
    return { kind: "error", problem: "invalid-expires-at" };
  }
  if (Date.parse(rec["expiresAt"] as string) <= Date.parse(rec["createdAt"] as string)) {
    return { kind: "error", problem: "expiry-not-future" };
  }
  const reason = rec["reason"];
  if (reason !== undefined && typeof reason !== "string") {
    return { kind: "error", problem: "invalid-reason" };
  }
  const drainedAt = rec["drainedAt"];
  if (
    drainedAt !== undefined &&
    (typeof drainedAt !== "string" || Number.isNaN(Date.parse(drainedAt)))
  ) {
    return { kind: "error", problem: "invalid-drained-at" };
  }
  if (rec["project"] !== project) {
    // A record persisted at this project's path but naming another project is
    // either a copy or a rename mishap; it fences nobody (that project's drain
    // lives at its own path).
    return { kind: "inactive" };
  }
  if (Date.parse(rec["expiresAt"] as string) <= now) {
    return { kind: "inactive" };
  }
  const drain: DrainRecord = {
    project: rec["project"],
    createdAt: rec["createdAt"],
    expiresAt: rec["expiresAt"],
    ...(reason === undefined ? {} : { reason }),
    ...(drainedAt === undefined ? {} : { drainedAt }),
  };
  return { kind: "active", drain };
}

/**
 * The dispatch-side drain read: the same verdict as {@link readDrain}, and the
 * one place a stale or untrustworthy record is cleared as a side effect —
 * expired and wrong-project records are removed (bounded stale-state cleanup
 * on the next pass), and a malformed record is removed in the very transition
 * that fails the pass closed, so it can never become an unbounded permanent
 * drain. Only dispatch callers use this: an observational read here would let
 * a status/health reader consume the malformed marker before the pass that
 * must fail closed on it ever ran.
 */
export function consumeDrain(project: string, now = Date.now()): DrainVerdict {
  const verdict = readDrain(project, now);
  if (verdict.kind !== "active") {
    rmSync(drainPath(project), { force: true });
  }
  return verdict;
}

/**
 * Marks an active drain with the instant its active set first reached zero —
 * the moment the release window opened (#1078). A no-op for anything that is
 * not a fresh, unmarked fence: absent, expired, malformed and already-marked
 * records are left exactly as they were, so only the opening transition ever
 * writes. The rewrite is atomic (tmp + rename, exactly like
 * {@link createDrain}), so a crash mid-write can never leave a half-record.
 * The annotation is what makes the window's opening durable and keeps the
 * tick's log line a once-per-drain event instead of once-per-pass noise.
 */
export function markDrained(project: string, now = Date.now()): void {
  const verdict = readDrain(project, now);
  if (verdict.kind !== "active" || verdict.drain.drainedAt !== undefined) return;
  const marked: DrainRecord = { ...verdict.drain, drainedAt: new Date(now).toISOString() };
  const path = drainPath(project);
  const tmp = `${path}.${process.pid.toString(36)}.${Date.now().toString(36)}.tmp`;
  writeFileSync(tmp, `${JSON.stringify(marked, null, 2)}\n`);
  renameSync(tmp, path);
}

/** Removes this project's drain, idempotently — a second cancel is a no-op. */
export function cancelDrain(project: string): void {
  rmSync(drainPath(project), { force: true });
}