/**
 * Module pause / unpause — taking a module out of celilo's control plane
 * without uninstalling it, so a capability provider its consumers depend on can
 * be removed and replaced.
 *
 * See openspec/changes/module-pause-lifecycle/. The shape of this file follows
 * Rule 10.4: `planPause`/`planUnpause` are pure and produce an explicit ordered
 * plan; `executePause`/`executeUnpause` perform the side effects. `--dry-run`
 * renders the plan, which for a fleet-wide operation is the most valuable part
 * of the feature.
 *
 * What makes pause cheap is that capability consumption is DEPLOY-time: every
 * consumer of `firewall` / `dhcp_server` calls it from `on_install`, and
 * nothing calls it while merely serving traffic. So a consumer only has to stop
 * participating in the control plane — not stop running — for its provider to
 * be replaced underneath it. That is why the data plane is left alone by
 * default (design D2).
 */

import { eq, inArray } from 'drizzle-orm';
import type { DbClient } from '../db/client';
import { IN_FLIGHT_STATES, type ModuleState, PAUSABLE_STATES, modules } from '../db/schema';
import type { ModuleManifest } from '../manifest/schema';
import {
  DependencyCycleError,
  type ModuleGraph,
  buildModuleGraph,
  topologicalOrder,
  transitiveConsumers,
} from './update/dep-graph';

export type PauseAction = 'pause' | 'unpause';

/** The read-only view of a module the planner needs. No DB handle, no I/O. */
export interface ModuleSnapshot {
  id: string;
  state: ModuleState;
  pausedAt: Date | null;
  pauseReason: string | null;
  manifest: ModuleManifest;
}

/**
 * `act` — the module is not yet in the target condition and will be changed.
 *
 * `skip_already` — it is already there. Crucially NOT an error and NOT a reason
 * to halt: a cascade must walk THROUGH members already in the target condition
 * to reach the ones beyond them, which is what makes a half-finished cascade
 * resumable (design D5, task 4.7).
 *
 * `skip_undeployed` — swept in by `--cascade` but never deployed, so there is
 * nothing to quiesce and nothing bound to the provider. Only ever applies to a
 * module the operator did NOT name; naming an undeployed module directly is
 * still refused, per the spec scenario "An undeployed module cannot be paused".
 */
export type StepDisposition = 'act' | 'skip_already' | 'skip_undeployed';

export interface PlanStep {
  moduleId: string;
  disposition: StepDisposition;
  /** Present on `skip_already`, so the report explains itself. */
  note?: string;
}

export interface PausePlan {
  action: PauseAction;
  /** The module the operator named. */
  requested: string;
  cascade: boolean;
  /** Whether the module's infrastructure should also be stopped (pause only). */
  stopInfra: boolean;
  /** Ordered: cascade pause is consumers-first, cascade unpause providers-first. */
  steps: PlanStep[];
}

/** Modules a plan would actually change — what confirmation must name. */
export function actedOn(plan: PausePlan): string[] {
  return plan.steps.filter((s) => s.disposition === 'act').map((s) => s.moduleId);
}

export class PauseRefusedError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'PauseRefusedError';
  }
}

export interface PlanRequest {
  moduleId: string;
  /** Every module celilo knows about — the planner filters, the caller does not. */
  fleet: ModuleSnapshot[];
  cascade: boolean;
  stopInfra?: boolean;
  /**
   * moduleId → operator-readable description of an operation currently in
   * flight for it. Pause is refused for those (design, closed question 3).
   */
  inFlight?: ReadonlyMap<string, string>;
}

/**
 * Has this module ever reached a deployed state? Anything settled-or-in-flight
 * past CONFIGURED has something on a machine; the earlier states do not.
 */
function isDeployed(state: ModuleState): boolean {
  return !(['IMPORTED', 'VALIDATED', 'CONFIGURED'] as readonly ModuleState[]).includes(state);
}

function snapshotById(fleet: ModuleSnapshot[]): Map<string, ModuleSnapshot> {
  return new Map(fleet.map((m) => [m.id, m]));
}

/**
 * The ordered set a cascade covers: the named module plus every transitive
 * consumer of it.
 *
 * Computed from the dependency GRAPH, never from which modules happen to be
 * paused — otherwise an already-unpaused module would truncate the set and the
 * cascade would stop at the first member that needed no work (task 4.7).
 */
function cascadeOrder(graph: ModuleGraph, moduleId: string, action: PauseAction): string[] {
  const affected = new Set([moduleId, ...transitiveConsumers(graph, moduleId)]);
  const providersFirst = topologicalOrder(graph).filter((id) => affected.has(id));

  // Unpausing redeploys, and a redeploy resolves capabilities, so a provider
  // must be live before any consumer redeploys. Pausing is the mirror: a
  // consumer must stop depending before its provider goes quiet.
  return action === 'unpause' ? providersFirst : providersFirst.slice().reverse();
}

function buildSteps(request: PlanRequest, action: PauseAction): PlanStep[] {
  const byId = snapshotById(request.fleet);
  const target = byId.get(request.moduleId);
  if (!target) {
    throw new PauseRefusedError(`Module not found: ${request.moduleId}`);
  }

  let ordered: string[];
  if (request.cascade) {
    try {
      ordered = cascadeOrder(
        buildModuleGraph(request.fleet.map((m) => m.manifest)),
        request.moduleId,
        action,
      );
    } catch (err) {
      // A cycle means there is no safe order, so acting on part of the set
      // would leave the fleet in a state no re-run can reason about. Refuse and
      // name the cycle (task 4.3) rather than pausing a partial set.
      if (err instanceof DependencyCycleError) {
        throw new PauseRefusedError(
          `Cannot ${action} with cascade: the affected modules declare a dependency cycle.\n  ${err.cycle.join(' → ')} → ${err.cycle[0]}\nFix the manifests, or ${action} each module individually.`,
        );
      }
      throw err;
    }
  } else {
    ordered = [request.moduleId];
  }

  return ordered.map((id) => {
    const snapshot = byId.get(id);
    // A graph node with no DB row cannot happen (the graph is built from the
    // fleet), but the map lookup is nullable and a silent skip would be worse
    // than a loud one.
    if (!snapshot) {
      throw new PauseRefusedError(`Module '${id}' is in the dependency graph but has no record`);
    }
    const isPaused = snapshot.state === 'PAUSED';
    const alreadyDone = action === 'pause' ? isPaused : !isPaused;
    if (alreadyDone) {
      return {
        moduleId: id,
        disposition: 'skip_already' as const,
        note: action === 'pause' ? 'already paused' : 'not paused',
      };
    }

    // A cascade sweeps in every transitive consumer from the manifest graph,
    // including ones that were imported but never deployed. Those are not bound
    // to the provider and have nothing to quiesce, so refusing the whole cascade
    // over them would wedge exactly the migration this feature exists to enable
    // — one stray imported module would block the swap. Skip them instead.
    //
    // The module the operator NAMED is not eligible for this: asking to pause an
    // undeployed module is a mistake worth reporting, and the spec says so.
    const named = id === request.moduleId;
    if (action === 'pause' && !named && !isDeployed(snapshot.state)) {
      return {
        moduleId: id,
        disposition: 'skip_undeployed' as const,
        note: `${snapshot.state} — never deployed, nothing to quiesce`,
      };
    }

    return { moduleId: id, disposition: 'act' as const };
  });
}

/**
 * Why a module cannot be paused right now, or null if it can. Split out so the
 * message names the specific case — "nothing deployed to quiesce" and "would
 * strand the transition" are different problems with different fixes.
 */
function pauseRefusal(
  snapshot: ModuleSnapshot,
  inFlight: ReadonlyMap<string, string>,
): string | null {
  const operation = inFlight.get(snapshot.id);
  if (operation) {
    return `'${snapshot.id}' has an operation in progress (${operation}). Wait for it to finish, or release it with "celilo module operations clear".`;
  }
  if ((PAUSABLE_STATES as readonly ModuleState[]).includes(snapshot.state)) return null;
  if ((IN_FLIGHT_STATES as readonly ModuleState[]).includes(snapshot.state)) {
    return `'${snapshot.id}' is ${snapshot.state} — pausing mid-transition would strand it. Wait for the transition to finish.`;
  }
  return `'${snapshot.id}' is ${snapshot.state} — it has never been deployed, so there is nothing to quiesce.`;
}

/**
 * Plan a pause. Pure: no DB, no bus, no filesystem.
 *
 * Refuses the WHOLE plan when any module it would act on cannot be paused,
 * rather than pausing a partial set and leaving the operator to work out which
 * half happened.
 */
export function planPause(request: PlanRequest): PausePlan {
  const steps = buildSteps(request, 'pause');
  const byId = snapshotById(request.fleet);
  const inFlight = request.inFlight ?? new Map<string, string>();

  const refusals = steps
    .filter((s) => s.disposition === 'act')
    .map((s) => byId.get(s.moduleId))
    .filter((s): s is ModuleSnapshot => s !== undefined)
    .map((s) => pauseRefusal(s, inFlight))
    .filter((r): r is string => r !== null);

  if (refusals.length > 0) {
    throw new PauseRefusedError(`Cannot pause:\n${refusals.map((r) => `  • ${r}`).join('\n')}`);
  }

  return {
    action: 'pause',
    requested: request.moduleId,
    cascade: request.cascade,
    stopInfra: request.stopInfra ?? false,
    steps,
  };
}

/**
 * Plan an unpause. Pure.
 *
 * There is no source-state gate here: the only module an unpause acts on is one
 * in `PAUSED`, which is settled by construction.
 */
export function planUnpause(request: PlanRequest): PausePlan {
  return {
    action: 'unpause',
    requested: request.moduleId,
    cascade: request.cascade,
    stopInfra: false,
    steps: buildSteps(request, 'unpause'),
  };
}

// ---------------------------------------------------------------------------
// Reading paused state
// ---------------------------------------------------------------------------

export interface PausedModule {
  id: string;
  pausedAt: Date | null;
  pauseReason: string | null;
}

/**
 * Every currently paused module. One indexed query (`modules_state_idx`) — it
 * runs on every management-API response (design D7), so it has to stay cheap.
 */
export function listPausedModules(db: DbClient): PausedModule[] {
  return db
    .select({
      id: modules.id,
      pausedAt: modules.pausedAt,
      pauseReason: modules.pauseReason,
    })
    .from(modules)
    .where(eq(modules.state, 'PAUSED'))
    .all();
}

/** True if this module is paused. Used by the remove guard and hook dispatch. */
export function isModulePaused(db: DbClient, moduleId: string): boolean {
  const row = db
    .select({ state: modules.state })
    .from(modules)
    .where(eq(modules.id, moduleId))
    .get();
  return row?.state === 'PAUSED';
}

/** The subset of `moduleIds` that are paused. One query, for guard/plan use. */
export function pausedAmong(db: DbClient, moduleIds: string[]): Set<string> {
  if (moduleIds.length === 0) return new Set();
  const rows = db
    .select({ id: modules.id, state: modules.state })
    .from(modules)
    .where(inArray(modules.id, moduleIds))
    .all();
  return new Set(rows.filter((r) => r.state === 'PAUSED').map((r) => r.id));
}

/**
 * How long a module has been paused, human-readable. One implementation so no
 * call site formats the age by hand (task 1.3) — `module list`, `module
 * status`, `system doctor` and the management-API warning all render the same
 * string for the same pause.
 */
export function formatPausedDuration(pausedAt: Date | null, now: Date = new Date()): string {
  if (!pausedAt) return 'unknown';
  const ms = Math.max(0, now.getTime() - pausedAt.getTime());
  const minutes = Math.floor(ms / 60_000);
  if (minutes < 1) return 'just now';
  if (minutes < 60) return `${minutes}m`;
  const hours = Math.floor(minutes / 60);
  if (hours < 24) return `${hours}h`;
  return `${Math.floor(hours / 24)}d`;
}

/**
 * What `--stop-infra` reports for a module hosted on a MACHINE (machine-pool
 * sense): not applicable, by design (design D2, revised).
 *
 * Extracted so the semantics are testable rather than buried in a string. The
 * wording is load-bearing: `--stop-infra` acts only on infrastructure celilo
 * PROVISIONED, and a machine is operator-pre-provisioned — it may predate
 * celilo and may run work celilo has never been told about. Saying celilo
 * "cannot determine the service unit" would imply a capability gap where the
 * truth is that this host is not celilo's to stop.
 */
export function describeMachineStopInfra(hostname: string, moduleId: string): string {
  return `${hostname}: machine-hosted — not applicable. --stop-infra acts only on infrastructure celilo provisioned; this machine is operator-managed and may run more than '${moduleId}'.`;
}

/** `caddy (3d, "swapping the edge router")` — the shared one-line rendering. */
export function describePausedModule(module: PausedModule, now: Date = new Date()): string {
  const age = formatPausedDuration(module.pausedAt, now);
  return module.pauseReason
    ? `${module.id} (${age}, "${module.pauseReason}")`
    : `${module.id} (${age})`;
}

// ---------------------------------------------------------------------------
// Execution
// ---------------------------------------------------------------------------

/**
 * The side-effecting collaborators, injected so the executor is testable
 * without a Proxmox, a bus, or a real deploy (Rule 2.3).
 */
export interface PauseDeps {
  db: DbClient;
  /** Quiesce: drop the module's bus subscriptions so nothing is delivered. */
  unsubscribe(moduleId: string): void;
  /** Re-arm them after a successful unpause redeploy. */
  resubscribe(moduleId: string): void;
  /** Unpause's rebinding mechanism (design D4). Resolves false on failure. */
  redeploy(moduleId: string): Promise<{ success: boolean; error?: string }>;
  /** `--stop-infra`. Returns what it did, for the report. */
  stopInfrastructure(moduleId: string): Promise<InfraStopOutcome>;
  /** Progress + resumability substrate (`module_operations`). */
  startOperation(moduleId: string, operation: 'pause' | 'unpause'): string;
  completeOperation(operationId: string): void;
  failOperation(operationId: string, error: unknown): void;
  now(): Date;
  log(message: string): void;
}

export interface InfraStopOutcome {
  stopped: boolean;
  /** Operator-readable: "stopped container vmid 231", "no infrastructure to stop". */
  detail: string;
}

export interface StepOutcome {
  moduleId: string;
  /** `acted` changed it; `skipped` was already there; `failed` did not happen. */
  result: 'acted' | 'skipped' | 'failed';
  detail?: string;
}

export interface ExecutionReport {
  action: PauseAction;
  outcomes: StepOutcome[];
  /** False if any step failed — the caller renders a non-zero result. */
  success: boolean;
}

function markPaused(deps: PauseDeps, moduleId: string, reason: string | null): void {
  deps.db
    .update(modules)
    .set({ state: 'PAUSED', pausedAt: deps.now(), pauseReason: reason, updatedAt: deps.now() })
    .where(eq(modules.id, moduleId))
    .run();
}

/**
 * Pause every module the plan acts on, in the plan's order.
 *
 * A step that fails does NOT abort the rest: the remaining modules are
 * independent, the half-paused state is durable and safe (paused modules are
 * quiesced, not broken), and re-running the cascade completes the outstanding
 * work. Aborting would leave a smaller done-set for no benefit.
 */
export async function executePause(
  plan: PausePlan,
  deps: PauseDeps,
  reason: string | null,
): Promise<ExecutionReport> {
  const outcomes: StepOutcome[] = [];

  for (const step of plan.steps) {
    if (step.disposition !== 'act') {
      // Deliberately not a halt (task 4.7) and deliberately not a re-write:
      // re-pausing preserves the ORIGINAL pausedAt/pauseReason, so the age
      // keeps measuring the real outage rather than resetting on every retry.
      outcomes.push({ moduleId: step.moduleId, result: 'skipped', detail: step.note });
      continue;
    }

    const opId = deps.startOperation(step.moduleId, 'pause');
    try {
      deps.unsubscribe(step.moduleId);
      markPaused(deps, step.moduleId, reason);

      let detail = 'paused';
      if (plan.stopInfra) {
        const outcome = await deps.stopInfrastructure(step.moduleId);
        detail = `paused; ${outcome.detail}`;
      }
      deps.completeOperation(opId);
      deps.log(`${step.moduleId}: ${detail}`);
      outcomes.push({ moduleId: step.moduleId, result: 'acted', detail });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      deps.failOperation(opId, err);
      deps.log(`${step.moduleId}: pause failed — ${message}`);
      outcomes.push({ moduleId: step.moduleId, result: 'failed', detail: message });
    }
  }

  return { action: 'pause', outcomes, success: outcomes.every((o) => o.result !== 'failed') };
}

/**
 * Unpause every paused module the plan covers, providers first.
 *
 * A failure DOES stop the cascade here, unlike pause: the plan is ordered so
 * providers come first, so continuing past a failed provider would redeploy its
 * consumers against a provider that is not there — precisely the mis-binding
 * the whole design exists to prevent. The unreached modules stay paused, which
 * is safe and visible, and re-running completes the remainder.
 */
export async function executeUnpause(plan: PausePlan, deps: PauseDeps): Promise<ExecutionReport> {
  const outcomes: StepOutcome[] = [];

  for (const step of plan.steps) {
    if (step.disposition !== 'act') {
      // An already-unpaused member is skipped WITHOUT redeploying (task 4.8) —
      // safe because a failed unpause leaves its module PAUSED, so there is no
      // "unpaused but never redeployed" state to repair.
      outcomes.push({ moduleId: step.moduleId, result: 'skipped', detail: step.note });
      continue;
    }

    // Read the pause metadata BEFORE the redeploy overwrites `state`, so a
    // failure can restore the module to the pause it was actually in rather
    // than stamping a fresh one.
    const previous = deps.db
      .select({ pausedAt: modules.pausedAt, pauseReason: modules.pauseReason })
      .from(modules)
      .where(eq(modules.id, step.moduleId))
      .get();

    const opId = deps.startOperation(step.moduleId, 'unpause');
    let deployResult: { success: boolean; error?: string };
    try {
      deployResult = await deps.redeploy(step.moduleId);
    } catch (err) {
      deployResult = { success: false, error: err instanceof Error ? err.message : String(err) };
    }

    if (!deployResult.success) {
      // The redeploy has already moved `state` off PAUSED (and possibly to
      // ERROR). Put it back, keeping the original timestamp, and re-drop the
      // subscriptions in case the deploy re-registered them: a module that is
      // reported as paused must actually BE quiesced.
      deps.db
        .update(modules)
        .set({
          state: 'PAUSED',
          pausedAt: previous?.pausedAt ?? deps.now(),
          pauseReason: previous?.pauseReason ?? null,
          updatedAt: deps.now(),
        })
        .where(eq(modules.id, step.moduleId))
        .run();
      deps.unsubscribe(step.moduleId);
      deps.failOperation(opId, deployResult.error ?? 'redeploy failed');

      const message = deployResult.error ?? 'redeploy failed';
      deps.log(`${step.moduleId}: unpause failed — ${message}; left paused`);
      outcomes.push({ moduleId: step.moduleId, result: 'failed', detail: message });

      // Everything after this depends on it. Stop rather than mis-bind.
      const unreached = plan.steps
        .slice(plan.steps.indexOf(step) + 1)
        .filter((s) => s.disposition === 'act');
      for (const s of unreached) {
        outcomes.push({
          moduleId: s.moduleId,
          result: 'skipped',
          detail: 'left paused — a provider it depends on failed to unpause',
        });
      }
      return { action: 'unpause', outcomes, success: false };
    }

    // The deploy set the live state; all that is left is to clear the pause.
    deps.db
      .update(modules)
      .set({ pausedAt: null, pauseReason: null, updatedAt: deps.now() })
      .where(eq(modules.id, step.moduleId))
      .run();
    deps.resubscribe(step.moduleId);
    deps.completeOperation(opId);
    deps.log(`${step.moduleId}: unpaused and redeployed`);
    outcomes.push({ moduleId: step.moduleId, result: 'acted', detail: 'unpaused and redeployed' });
  }

  return { action: 'unpause', outcomes, success: true };
}