/**
 * Module-operations tracking — used by `checkInFlight()` to refuse a
 * backup or restore while another module operation (deploy, uninstall,
 * backup, restore) is active.
 *
 * Lifecycle:
 *   const opId = startOperation('homebridge', 'deploy');
 *   try {
 *     ...work...
 *     completeOperation(opId);
 *   } catch (err) {
 *     failOperation(opId, err);
 *     throw err;
 *   }
 *
 * A row with status='in_progress' stops holding the lock once it looks
 * abandoned, which is three different things:
 *
 *   - the process is GONE      — it crashed before writing completion
 *   - the process is STOPPED   — suspended (Ctrl-Z) or a zombie, so it
 *                                will never reach the completion write
 *   - the row is OLD           — past `OPERATION_TTL_MS`
 *
 * Only the first was originally handled, and the other two are not
 * hypothetical: a `module deploy` Ctrl-Z'd on a lost terminal held the
 * lock for 20 days and blocked every backup on the fleet.
 *
 * Stale rows are ignored rather than deleted; `celilo module operations`
 * lists them and `... clear` sweeps them — on an hourly bus tick, not
 * only when a human remembers (see `ensureOperationsSweepSubscriber`).
 */

import { spawnSync } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import { eq } from 'drizzle-orm';
import { log } from '../cli/prompts';
import { getDb } from '../db/client';
import { type ModuleOperation, type ModuleOperationKind, moduleOperations } from '../db/schema';

/**
 * Insert an in-progress row for an operation. Returns the operation id
 * that must be passed to completeOperation/failOperation.
 */
export function startOperation(moduleId: string, operation: ModuleOperationKind): string {
  const db = getDb();
  const id = randomUUID();
  db.insert(moduleOperations)
    .values({
      id,
      moduleId,
      operation,
      status: 'in_progress',
      pid: process.pid,
    })
    .run();
  return id;
}

/**
 * Write an operation's outcome without ever being able to break the flow that
 * is reporting it (celilo#737).
 *
 * Recording an outcome is BOOKKEEPING; the caller's error is the information.
 * On celilo-mgr a `SQLITE_BUSY` inside `failOperation` propagated out of the
 * catch block that called it and REPLACED the deploy's own error, so the
 * operator was shown a database-locking problem and never learned what the
 * deploy actually did wrong. That error was destroyed and is unrecoverable —
 * the whole cost of the bug. It also skipped the `emitDeployFailed` that
 * follows the call, so the event bus never learned the deploy had failed at
 * all, and the module was left `INSTALLED` while in fact verified.
 *
 * `getDb()` is inside the try on purpose: opening the database is one of the
 * things that can throw here.
 *
 * ⚠️ `startOperation` deliberately does NOT get this treatment. Its row IS the
 * in-flight lock `checkInFlight` reads to refuse a backup during a deploy, so a
 * silently-missing row would let the two run together against the same module.
 * Failing before any work happens is honest; failing after it is what #737 is
 * about.
 */
function recordOutcome(outcome: 'completed' | 'failed', operationId: string, write: () => void) {
  try {
    write();
  } catch (persistError) {
    // Rule 6.2: never a bare catch. Secondary to whatever the caller is already
    // reporting, so it is a warning rather than the headline — the caller's own
    // error is what the operator needs to read.
    const reason = persistError instanceof Error ? persistError.message : String(persistError);
    log.warn(`Could not record operation ${operationId} as ${outcome}: ${reason}`);
  }
}

export function completeOperation(operationId: string): void {
  recordOutcome('completed', operationId, () => {
    getDb()
      .update(moduleOperations)
      .set({ status: 'completed', completedAt: new Date() })
      .where(eq(moduleOperations.id, operationId))
      .run();
  });
}

export function failOperation(operationId: string, error: unknown): void {
  const message = error instanceof Error ? error.message : String(error);
  recordOutcome('failed', operationId, () => {
    getDb()
      .update(moduleOperations)
      .set({ status: 'failed', completedAt: new Date(), errorMessage: message })
      .where(eq(moduleOperations.id, operationId))
      .run();
  });
}

/**
 * How long an in_progress row may hold the lock before it is treated as
 * abandoned regardless of what its process appears to be doing.
 *
 * This is not belt-and-braces on the liveness check — it is the only
 * check that survives pid reuse. A pid is a recycled number, not a
 * stable identity: a busy host wraps the whole pid space in days, after
 * which an old row's pid names an unrelated live process and the
 * liveness check happily reports "still running" forever. Ageing the row
 * out is the only thing that ends that.
 *
 * Two hours is longer than any real deploy and short enough that a wedge
 * is an inconvenience rather than an outage.
 */
export const OPERATION_TTL_MS = 2 * 60 * 60 * 1000;

/**
 * True if the process can still make progress on its operation.
 *
 * `kill(pid, 0)` answers only "does this pid exist". A STOPPED process —
 * SIGTSTP from a Ctrl-Z, or a lost controlling terminal — passes that
 * test while being permanently unable to finish, which is exactly how
 * the 20-day wedge happened. `ps -o state=` reports the state itself and
 * is spelled the same on Linux and macOS.
 */
export function isPidRunnable(pid: number): boolean {
  const result = spawnSync('ps', ['-o', 'state=', '-p', String(pid)], { encoding: 'utf-8' });

  // No usable `ps`. Fall back to bare existence: a stopped process will
  // still block, which is the old behavior, but we never wrongly release
  // a lock that a live operation is holding.
  if (result.error) {
    try {
      process.kill(pid, 0);
      return true;
    } catch {
      return false;
    }
  }

  if (result.status !== 0) return false; // no such process

  // Linux reports multi-character states ("Tl", "Ss"); the first
  // character is the state proper. T = stopped, Z = zombie.
  const state = result.stdout.trim()[0] ?? '';
  return state !== 'T' && state !== 'Z';
}

/**
 * The `errorMessage` written when a row is released as abandoned.
 *
 * Load-bearing, not cosmetic: releasing marks rows `failed` rather than
 * deleting them, and this exact string is what later distinguishes "the
 * operation reported a failure" from "the operation never reported
 * anything and the sweep reclaimed it". The abandoned-operations audit
 * counts rows by it (`services/audit/abandoned-operations.ts`).
 */
export const ABANDONED_RELEASE_MESSAGE = 'abandoned — released by "celilo module operations clear"';

/**
 * Reclaim abandoned rows on a schedule instead of when a human remembers.
 *
 * Registered as an ordinary bus subscriber whose handler is the existing
 * `celilo module operations clear`, exactly like the backup sweep
 * (`services/backup-sweep.ts`) — no new command and no new scheduler.
 * Hourly is far finer than the two-hour TTL, so a wedge never survives
 * long, and clearing is idempotent: a pass with nothing abandoned is a
 * single read.
 *
 * `clear` without `--all` only touches rows that `checkInFlight` already
 * ignores, so the sweep can never release a lock a live operation holds.
 */
export const OPERATIONS_SWEEP_SUBSCRIBER = 'celilo-operations-sweep';
export const OPERATIONS_SWEEP_PATTERN = 'timer.tick.1h';

export interface SubscriberRegistrar {
  subscribe(options: {
    name: string;
    pattern: string;
    handler: string;
    registeredBy?: string;
  }): unknown;
}

/** Idempotent: `bus.subscribe` upserts by name. */
export function ensureOperationsSweepSubscriber(bus: SubscriberRegistrar): void {
  bus.subscribe({
    name: OPERATIONS_SWEEP_SUBSCRIBER,
    pattern: OPERATIONS_SWEEP_PATTERN,
    handler: 'celilo module operations clear',
    registeredBy: 'celilo-module-operations',
  });
}

export interface InFlightConflict {
  operation: ModuleOperation;
  /** A short, operator-readable description: "deploy of homebridge (pid 12345)". */
  describe: string;
}

/**
 * Returns rows that genuinely look in-flight: status='in_progress', the
 * row is younger than `OPERATION_TTL_MS`, AND the originating process is
 * still able to make progress. Everything else is abandoned and excluded,
 * so a crashed, suspended, or forgotten operation cannot wedge the fleet.
 *
 * @param excludeOperationId - operation id to exclude from the check
 *   (so an operation doesn't see itself as a conflict).
 */
export function checkInFlight(excludeOperationId?: string): InFlightConflict[] {
  const db = getDb();
  const rows = db
    .select()
    .from(moduleOperations)
    .where(eq(moduleOperations.status, 'in_progress'))
    .all();

  const now = Date.now();
  const conflicts: InFlightConflict[] = [];
  for (const row of rows) {
    if (excludeOperationId && row.id === excludeOperationId) continue;
    // Age first: it costs nothing, where the liveness probe spawns `ps`.
    if (now - row.startedAt.getTime() > OPERATION_TTL_MS) continue;
    if (!isPidRunnable(row.pid)) continue;
    conflicts.push({
      operation: row,
      describe: `${row.operation} of ${row.moduleId} (pid ${row.pid})`,
    });
  }
  return conflicts;
}

/**
 * Throws an InFlightError when conflicts exist. Returned error message
 * is operator-readable: it names the conflicting operation(s) and the
 * suggested retry.
 */
export class InFlightError extends Error {
  constructor(public readonly conflicts: InFlightConflict[]) {
    const list = conflicts.map((c) => `  • ${c.describe}`).join('\n');
    const hint =
      'If it is not really running: "celilo module operations" to inspect, "celilo module operations clear" to release.';
    super(
      `Cannot start: another module operation is in progress.\n${list}\nWait for it to complete (or fail) and re-run.\n${hint}`,
    );
    this.name = 'InFlightError';
  }
}

/**
 * Throws InFlightError if a conflicting operation is in flight.
 * Call this BEFORE startOperation() at the entry of backup/restore
 * service functions.
 */
export function refuseIfInFlight(excludeOperationId?: string): void {
  const conflicts = checkInFlight(excludeOperationId);
  if (conflicts.length > 0) {
    throw new InFlightError(conflicts);
  }
}

/**
 * Restricted variant: refuse if any operation matching the predicate is
 * in-flight. Used by callers that want to allow some concurrent ops
 * (e.g. multiple backups across modules) but not others. Today's spec
 * doesn't need this — refuseIfInFlight() suffices — but the hook is
 * here for the future.
 */
export function refuseIfInFlightMatching(
  predicate: (op: ModuleOperation) => boolean,
  excludeOperationId?: string,
): void {
  const conflicts = checkInFlight(excludeOperationId).filter((c) => predicate(c.operation));
  if (conflicts.length > 0) {
    throw new InFlightError(conflicts);
  }
}