import { lstat, readdir } from "node:fs/promises";
import { basename, isAbsolute, join, normalize } from "node:path";

import type { RunRecordDiagnostic } from "./run-store.js";
import type {
  ProcessIdentity,
  ProcessIdentityInspector,
  ProcessInspection,
  WorkerSessionLeaseDiagnostic,
  WorkflowLeaseOwnerV1,
} from "./workflow-lease.js";
import type { RunProjectionV2, WorkerAttemptProjection, WorkerProjection } from "./workflow-types.js";

export interface WorkflowDoctorCommand {
  readonly kind: "exec" | "pi";
  readonly argv: readonly string[];
  readonly purpose: string;
}

export interface WorkflowDoctorFinding {
  readonly severity: "error" | "warning";
  readonly code:
    | "corrupt-run-record"
    | "corrupt-session-lease"
    | "orphan-process"
    | "ledger-process-missing"
    | "stale-session-lease"
    | "abandoned-session-lease"
    | "dangling-session";
  readonly message: string;
  readonly paths: readonly string[];
  readonly commands: readonly WorkflowDoctorCommand[];
}

export interface WorkflowDoctorReport {
  readonly schema: 1;
  readonly checkedAt: string;
  readonly status: "ok" | "attention";
  readonly summary: {
    readonly healthyRuns: number;
    readonly corruptRuns: number;
    readonly workerProcessesChecked: number;
    readonly orphanProcesses: number;
    readonly staleOrAbandonedLeases: number;
    readonly danglingSessions: number;
  };
  readonly diskUsage: {
    readonly totalBytes: number;
    readonly byArea: readonly { readonly name: string; readonly path: string; readonly bytes: number }[];
  };
  readonly findings: readonly WorkflowDoctorFinding[];
  readonly suggestedCommands: readonly WorkflowDoctorCommand[];
}

interface RunDiagnosticCatalog {
  diagnoseRunRecords(): Promise<readonly RunRecordDiagnostic[]>;
}

interface LeaseDiagnosticCatalog {
  diagnoseWorkerSessionLeases(): Promise<readonly WorkerSessionLeaseDiagnostic[]>;
}

export interface WorkflowDoctorOptions {
  readonly namespaceDirectory: string;
  readonly runs: RunDiagnosticCatalog;
  readonly leases: LeaseDiagnosticCatalog;
  readonly inspector: ProcessIdentityInspector;
  readonly now?: () => Date;
}

interface AttemptReference {
  readonly run: RunProjectionV2;
  readonly worker: WorkerProjection;
  readonly attempt: WorkerAttemptProjection;
}

export class WorkflowDoctor {
  readonly #namespaceDirectory: string;
  readonly #runs: RunDiagnosticCatalog;
  readonly #leases: LeaseDiagnosticCatalog;
  readonly #inspector: ProcessIdentityInspector;
  readonly #now: () => Date;

  constructor(options: WorkflowDoctorOptions) {
    if (!isAbsolute(options.namespaceDirectory)) {
      throw new TypeError("Workflow doctor namespace must be absolute");
    }
    this.#namespaceDirectory = normalize(options.namespaceDirectory);
    this.#runs = options.runs;
    this.#leases = options.leases;
    this.#inspector = options.inspector;
    this.#now = options.now ?? (() => new Date());
  }

  async inspect(): Promise<WorkflowDoctorReport> {
    const records = await this.#runs.diagnoseRunRecords();
    const healthy = records.filter((record) => record.kind === "healthy");
    const findings: WorkflowDoctorFinding[] = [];
    const orphanProcessIds = new Set<number>();
    for (const record of records) {
      if (record.kind !== "corrupt") continue;
      findings.push({
        severity: "error",
        code: "corrupt-run-record",
        message: record.diagnostic,
        paths: [record.directory],
        commands: [quarantineCommand(this.#namespaceDirectory, "run-record", record.directory)],
      });
    }

    const attempts = healthy.flatMap((record) => attemptsIn(record.projection));
    let workerProcessesChecked = 0;
    for (const reference of attempts) {
      const pid = reference.attempt.pidStartEvidence.pid;
      if (pid === null) continue;
      workerProcessesChecked += 1;
      const inspection = await this.#inspector.inspect(pid);
      const matches = inspection.kind === "live"
        && processIdentityMatchesAttempt(inspection.identity, reference.attempt);
      if (matches && reference.attempt.pidStartEvidence.observedExitAt !== null) {
        orphanProcessIds.add(pid);
        findings.push(orphanProcessFinding(
          pid,
          reference.attempt.session.path,
          `Process ${pid} still matches terminal ledger attempt ${reference.attempt.attemptId}`,
        ));
      } else if (!matches && reference.attempt.pidStartEvidence.observedExitAt === null) {
        findings.push({
          severity: "warning",
          code: "ledger-process-missing",
          message: processMismatchDiagnostic(pid, inspection, reference.attempt),
          paths: [reference.attempt.session.path],
          commands: [inspectProcessCommand(pid)],
        });
      }
    }

    let currentIdentity: ProcessIdentity | undefined;
    try {
      currentIdentity = await this.#inspector.current();
    } catch {
      // Individual records stay visible as abandoned/unverifiable below.
    }
    const leaseRecords = await this.#leases.diagnoseWorkerSessionLeases();
    for (const record of leaseRecords) {
      if (record.projection.kind === "empty") continue;
      if (record.projection.kind === "corrupt") {
        findings.push({
          severity: "error",
          code: "corrupt-session-lease",
          message: record.projection.diagnostic,
          paths: [record.path],
          commands: [quarantineCommand(this.#namespaceDirectory, "session-lease", record.path)],
        });
        continue;
      }
      const owner = record.projection.owner;
      const reference = attempts.find((candidate) => leaseOwnerReferences(owner, candidate));
      const inspection = await inspectLeaseOwner(owner, currentIdentity, this.#inspector);
      const live = inspection.kind === "live" && processIdentityMatchesOwner(inspection.identity, owner);
      if (live && (reference === undefined || !leaseOwnerConsistentWithLedger(owner, reference))) {
        if (!orphanProcessIds.has(owner.pid)) {
          orphanProcessIds.add(owner.pid);
          findings.push(orphanProcessFinding(
            owner.pid,
            record.path,
            reference === undefined
              ? `Live lease process ${owner.pid} has no matching readable ledger attempt`
              : `Live lease process ${owner.pid} conflicts with ledger attempt ${reference.attempt.attemptId}`,
          ));
        }
      } else if (!live && inspection.kind !== "unknown") {
        findings.push({
          severity: "warning",
          code: "stale-session-lease",
          message: leaseMismatchDiagnostic(owner, inspection),
          paths: [record.path, ...(owner.sessionPath ? [owner.sessionPath] : [])],
          commands: [quarantineCommand(this.#namespaceDirectory, "session-lease", record.path)],
        });
      } else if (inspection.kind === "unknown") {
        findings.push({
          severity: "warning",
          code: "abandoned-session-lease",
          message: inspection.diagnostic,
          paths: [record.path, ...(owner.sessionPath ? [owner.sessionPath] : [])],
          commands: [inspectProcessCommand(owner.pid)],
        });
      }
    }

    const referencedSessions = new Set(attempts.map((reference) => normalize(reference.attempt.session.path)));
    for (const path of await sessionDirectories(this.#namespaceDirectory)) {
      if (referencedSessions.has(path)) continue;
      findings.push({
        severity: "warning",
        code: "dangling-session",
        message: records.some((record) => record.kind === "corrupt")
          ? "Session is not referenced by a readable run; corrupt run records make ownership uncertain"
          : "Session is not referenced by any workflow run",
        paths: [path],
        commands: [quarantineCommand(this.#namespaceDirectory, "session", path)],
      });
    }

    findings.sort((left, right) =>
      left.code.localeCompare(right.code) || left.paths[0]!.localeCompare(right.paths[0]!));
    const byArea = await topLevelDiskUsage(this.#namespaceDirectory);
    return {
      schema: 1,
      checkedAt: this.#now().toISOString(),
      status: findings.length === 0 ? "ok" : "attention",
      summary: {
        healthyRuns: healthy.length,
        corruptRuns: records.length - healthy.length,
        workerProcessesChecked,
        orphanProcesses: orphanProcessIds.size,
        staleOrAbandonedLeases: findings.filter((finding) =>
          finding.code === "stale-session-lease"
          || finding.code === "abandoned-session-lease"
          || finding.code === "corrupt-session-lease").length,
        danglingSessions: findings.filter((finding) => finding.code === "dangling-session").length,
      },
      diskUsage: {
        totalBytes: byArea.reduce((sum, area) => sum + area.bytes, 0),
        byArea,
      },
      findings,
      suggestedCommands: [{
        kind: "pi",
        argv: ["/pi-amq-cleanup", "--older-than", "7d"],
        purpose: "Preview explicit cleanup of old terminal workflow state",
      }],
    };
  }
}

function attemptsIn(run: RunProjectionV2): AttemptReference[] {
  return Object.values(run.workers).flatMap((worker) =>
    worker.attempts.map((attempt) => ({ run, worker, attempt })));
}

function leaseOwnerReferences(owner: WorkflowLeaseOwnerV1, reference: AttemptReference): boolean {
  return owner.runId === reference.run.runId
    && owner.workerId === reference.worker.workerId
    && owner.attemptId === reference.attempt.attemptId
    && owner.workerAttemptGeneration === reference.attempt.workerAttemptGeneration
    && owner.sessionPath === reference.attempt.session.path;
}

function leaseOwnerConsistentWithLedger(
  owner: WorkflowLeaseOwnerV1,
  reference: AttemptReference,
): boolean {
  if (owner.stage === "reserved") {
    return reference.attempt.state === "pending"
      && reference.attempt.pidStartEvidence.pid === null
      && reference.attempt.session.leaseGeneration === null;
  }
  return reference.attempt.pidStartEvidence.observedExitAt === null
    && reference.attempt.session.leaseGeneration === owner.generation
    && processIdentityMatchesAttempt({
      pid: owner.pid,
      processStartIdentity: owner.processStartIdentity,
      bootIdentity: owner.bootIdentity,
      executable: owner.executable,
      hostname: owner.hostname,
    }, reference.attempt);
}

async function inspectLeaseOwner(
  owner: WorkflowLeaseOwnerV1,
  current: ProcessIdentity | undefined,
  inspector: ProcessIdentityInspector,
): Promise<ProcessInspection> {
  if (!current) return { kind: "unknown", diagnostic: "Current process identity is unavailable" };
  if (owner.hostname !== current.hostname) {
    return { kind: "unknown", diagnostic: `Lease belongs to remote host ${owner.hostname}` };
  }
  return inspector.inspect(owner.pid);
}

function processIdentityMatchesAttempt(
  identity: ProcessIdentity,
  attempt: WorkerAttemptProjection,
): boolean {
  const recorded = attempt.pidStartEvidence;
  return identity.pid === recorded.pid
    && identity.processStartIdentity === recorded.processStartIdentity
    && identity.bootIdentity === recorded.bootIdentity
    && identity.executable === recorded.executable
    && identity.hostname === recorded.hostname;
}

function processIdentityMatchesOwner(identity: ProcessIdentity, owner: WorkflowLeaseOwnerV1): boolean {
  return identity.pid === owner.pid
    && identity.processStartIdentity === owner.processStartIdentity
    && identity.bootIdentity === owner.bootIdentity
    && identity.executable === owner.executable
    && identity.hostname === owner.hostname;
}

function processMismatchDiagnostic(
  pid: number,
  inspection: ProcessInspection,
  attempt: WorkerAttemptProjection,
): string {
  if (inspection.kind === "absent") return `Ledger attempt ${attempt.attemptId} has no process ${pid}`;
  if (inspection.kind === "unknown") {
    return `Ledger attempt ${attempt.attemptId} process ${pid} is unverifiable: ${inspection.diagnostic}`;
  }
  return `Ledger attempt ${attempt.attemptId} PID ${pid} now has different process identity`;
}

function leaseMismatchDiagnostic(owner: WorkflowLeaseOwnerV1, inspection: ProcessInspection): string {
  return inspection.kind === "absent"
    ? `Lease owner process ${owner.pid} is absent`
    : `Lease owner PID ${owner.pid} now has different process identity`;
}

function orphanProcessFinding(
  pid: number,
  path: string,
  message: string,
): WorkflowDoctorFinding {
  return {
    severity: "error",
    code: "orphan-process",
    message,
    paths: [path],
    commands: [inspectProcessCommand(pid)],
  };
}

function inspectProcessCommand(pid: number): WorkflowDoctorCommand {
  return {
    kind: "exec",
    argv: ["ps", "-p", String(pid), "-o", "pid=,lstart=,comm="],
    purpose: "Re-inspect PID identity before any operator action",
  };
}

function quarantineCommand(
  namespaceDirectory: string,
  prefix: string,
  path: string,
): WorkflowDoctorCommand {
  return {
    kind: "exec",
    argv: ["mv", "--", path, join(namespaceDirectory, "quarantine", `doctor-${prefix}-${basename(path)}`)],
    purpose: "Quarantine the record for manual recovery instead of deleting it",
  };
}

async function sessionDirectories(namespaceDirectory: string): Promise<readonly string[]> {
  const parent = join(namespaceDirectory, "worker-sessions");
  try {
    return (await readdir(parent, { withFileTypes: true }))
      .filter((entry) => !entry.name.startsWith("."))
      .map((entry) => normalize(join(parent, entry.name)))
      .sort();
  } catch (error) {
    if (isNodeError(error) && error.code === "ENOENT") return [];
    throw error;
  }
}

async function topLevelDiskUsage(
  namespaceDirectory: string,
): Promise<readonly { readonly name: string; readonly path: string; readonly bytes: number }[]> {
  let entries: string[];
  try {
    entries = await readdir(namespaceDirectory);
  } catch (error) {
    if (isNodeError(error) && error.code === "ENOENT") return [];
    throw error;
  }
  const areas = await Promise.all(entries.filter((entry) => !entry.startsWith(".")).sort().map(async (name) => {
    const path = join(namespaceDirectory, name);
    return { name, path, bytes: await pathBytes(path) };
  }));
  return areas;
}

async function pathBytes(path: string): Promise<number> {
  let info;
  try {
    info = await lstat(path);
  } catch (error) {
    if (isNodeError(error) && error.code === "ENOENT") return 0;
    throw error;
  }
  if (!info.isDirectory()) return info.size;
  let total = info.size;
  for (const entry of await readdir(path)) total += await pathBytes(join(path, entry));
  return total;
}

function isNodeError(error: unknown): error is NodeJS.ErrnoException {
  return error instanceof Error && "code" in error;
}