import type { Bus } from './bus';
import { BUS_VERSION } from './bus';
import { retentionSweep } from './sweeper';
import { emitDueTimerTicks } from './timer';
import type { DispatcherOptions } from './types';

interface ClaimRow {
  event_id: number;
  subscriber_id: number;
  attempts: number;
  type: string;
  payload: string | null;
  subscriber_name: string;
  handler: string;
  max_attempts: number;
  timeout_ms: number;
}

const DEFAULT_DISPATCHER: Required<DispatcherOptions> = {
  pollIntervalMs: 1000,
  concurrency: 4,
  staleHandlerMultiplier: 2,
  haltOnRecovery: false,
};

/**
 * Drain all currently-claimable pending deliveries until the queue is
 * empty. Useful as a one-shot from cron. Does NOT write a heartbeat —
 * heartbeats are a runDispatcher concern.
 */
export async function drainOnce(
  bus: Bus,
  opts?: { concurrency?: number },
): Promise<{ processed: number }> {
  const concurrency = opts?.concurrency ?? DEFAULT_DISPATCHER.concurrency;
  let processed = 0;
  while (true) {
    const claims = claimPending(bus, concurrency);
    if (claims.length === 0) return { processed };
    await Promise.all(claims.map((c) => runOne(bus, c)));
    processed += claims.length;
  }
}

/**
 * Long-running dispatcher loop. Writes a heartbeat each tick, claims
 * up to `concurrency` pending deliveries, processes them concurrently,
 * sleeps `pollIntervalMs`. Returns a stop function.
 *
 * On startup, runs the crash-recovery sweep (see crashRecovery below).
 */
export function runDispatcher(bus: Bus, opts?: DispatcherOptions): { stop: () => Promise<void> } {
  // Coalesce per-field rather than spread: a caller that passes an explicit
  // `undefined` (e.g. `celilo events run` with no --concurrency flag) would
  // otherwise OVERWRITE the default with undefined, and `claimPending(bus,
  // undefined)` binds undefined as the SQL LIMIT → SQLITE_MISMATCH crash.
  // drainOnce already used `??`; this brings runDispatcher in line. See
  // v2/issues/ISS-0010.
  const cfg: Required<DispatcherOptions> = {
    pollIntervalMs: opts?.pollIntervalMs ?? DEFAULT_DISPATCHER.pollIntervalMs,
    concurrency: opts?.concurrency ?? DEFAULT_DISPATCHER.concurrency,
    staleHandlerMultiplier:
      opts?.staleHandlerMultiplier ?? DEFAULT_DISPATCHER.staleHandlerMultiplier,
    haltOnRecovery: opts?.haltOnRecovery ?? DEFAULT_DISPATCHER.haltOnRecovery,
  };

  assertSoleDispatcher(bus);

  const recovery = crashRecovery(bus, cfg);
  // AFTER recovery, never before: crash recovery reads the dead dispatcher's
  // heartbeat to decide whether to sweep stuck deliveries. Pruning first would
  // make every crash look like a fresh start and strand the stuck rows.
  pruneDeadHeartbeats(bus);
  if (recovery.recovered && cfg.haltOnRecovery) {
    throw new Error(
      `event-bus: dispatcher recovered from crash (${recovery.stuckCount} stuck deliveries reset). haltOnRecovery is true — refusing to start. Run \`event-bus resume\` to acknowledge.`,
    );
  }

  let stopping = false;
  let activeLoop: Promise<void> | null = null;
  // Captured once per process start. The heartbeat reports THIS process's start
  // time (not the first-ever row), so the doctor's started_at-vs-installed-code
  // currency check resets on restart instead of being permanently stuck (ISS-0121).
  const dispatcherStartedAt = Date.now();
  bus.dispatcherRunning = true;
  bus.dispatcherStop = () => {
    stopping = true;
  };

  // Run retention infrequently — every Nth idle tick — so it doesn't
  // dominate the loop on a busy bus. 1 sweep per ~5 minutes by default.
  const RETENTION_TICK_INTERVAL = Math.max(1, Math.floor((5 * 60 * 1000) / cfg.pollIntervalMs));
  let tick = 0;

  const loop = (async () => {
    while (!stopping) {
      try {
        writeHeartbeat(bus, dispatcherStartedAt);
        // Scheduler surface (timer.ts): emit due timer.tick.* events
        // before claiming, so a tick's deliveries are claimable in the
        // same pass. Subscriber- and interval-gated; cheap when idle.
        try {
          emitDueTimerTicks(bus);
        } catch (err) {
          console.error('[event-bus dispatcher] timer emission failed:', err);
        }
        const claims = claimPending(bus, cfg.concurrency);
        if (claims.length > 0) {
          await Promise.all(claims.map((c) => runOne(bus, c)));
        } else {
          await sleep(cfg.pollIntervalMs);
        }
        sweepStaleRunning(bus, cfg.staleHandlerMultiplier);
        tick = (tick + 1) % RETENTION_TICK_INTERVAL;
        if (tick === 0) {
          try {
            retentionSweep(bus);
          } catch (err) {
            console.error('[event-bus dispatcher] retention sweep failed:', err);
          }
        }
      } catch (err) {
        console.error('[event-bus dispatcher] loop error:', err);
        await sleep(cfg.pollIntervalMs);
      }
    }
    bus.dispatcherRunning = false;
  })();
  activeLoop = loop;

  return {
    async stop() {
      stopping = true;
      if (activeLoop) await activeLoop;
    },
  };
}

/**
 * Refuse to start when another dispatcher already holds this bus (#580).
 *
 * celilo-mgr ran two dispatchers for 40 days. The systemd unit could not have
 * prevented it: the elder process was not in the unit's cgroup at all — it was
 * started outside systemd and outlived a restart — so KillMode/Restart tuning
 * has no reach. Exclusion has to live here, where every launch path passes,
 * whether it came from systemd, launchd, cron, or a shell.
 *
 * Claiming a delivery is already safe under concurrency (see claimPending), so
 * the duplicate did not double-execute deliveries. What it did do was burn a
 * core polling, throw SQLITE_BUSY at every claim, and corrupt the single
 * heartbeat row both processes were writing.
 */
function assertSoleDispatcher(bus: Bus): void {
  const peers = bus
    .liveDispatchers()
    // Never mistake ourselves for a peer: a recycled pid could match a stale row.
    .filter((d) => d.dispatcherId !== bus.dispatcherId && d.pid !== process.pid);
  if (peers.length === 0) return;

  const list = peers
    .map((p) => `pid ${p.pid} (${p.dispatcherId}, up ${Math.round(p.startedAgoMs / 1000)}s)`)
    .join(', ');
  throw new Error(
    `event-bus: another dispatcher is already running on this bus: ${list}. Only one dispatcher may run per bus. Stop the other process (\`kill ${peers[0]?.pid}\`) or check \`celilo events status\`, then retry.`,
  );
}

/**
 * Drop heartbeat rows whose process is gone. Rows are per-process now, so
 * without this every restart leaks one and `dispatcherCount` counts corpses.
 */
function pruneDeadHeartbeats(bus: Bus): void {
  const liveIds = new Set(bus.liveDispatchers().map((d) => d.dispatcherId));
  liveIds.add(bus.dispatcherId);
  const stale = bus.db
    .query<{ dispatcher_id: string }, []>('SELECT dispatcher_id FROM dispatcher_heartbeat')
    .all()
    .filter((r) => !liveIds.has(r.dispatcher_id));
  for (const row of stale) {
    bus.db.run('DELETE FROM dispatcher_heartbeat WHERE dispatcher_id = ?', [row.dispatcher_id]);
  }
}

function writeHeartbeat(bus: Bus, startedAt: number): void {
  const now = Date.now();
  // Refresh started_at AND version on conflict — not just last_heartbeat/pid.
  // A restarted process must report its own start time and current code version;
  // otherwise the row stays frozen at the first-ever insert and the doctor's
  // "running current code" check is permanently stuck (a restart can never move
  // started_at, making its remediation futile). ISS-0121.
  bus.db.run(
    `INSERT INTO dispatcher_heartbeat (dispatcher_id, last_heartbeat, started_at, pid, version)
     VALUES (?, ?, ?, ?, ?)
     ON CONFLICT(dispatcher_id) DO UPDATE SET
       last_heartbeat = excluded.last_heartbeat,
       pid = excluded.pid,
       started_at = excluded.started_at,
       version = excluded.version`,
    [bus.dispatcherId, now, startedAt, process.pid, BUS_VERSION],
  );
}

/**
 * Claim up to `n` pending deliveries atomically. Uses BEGIN IMMEDIATE
 * + UPDATE ... RETURNING so a single dispatcher can pull a batch
 * without racing the next tick. (Multi-dispatcher would need per-row
 * leases; that's v2.)
 *
 * Returns rows joined with their subscriber + event so the caller has
 * everything needed to spawn the handler without follow-up queries.
 */
function claimPending(bus: Bus, n: number): ClaimRow[] {
  const now = Date.now();
  // IMMEDIATE, not the default DEFERRED. Deferred takes a read lock for the
  // SELECT and then tries to UPGRADE at the first UPDATE, which is the classic
  // SQLITE_BUSY_SNAPSHOT path — and busy_timeout does NOT retry an upgrade, so
  // the whole tick throws. celilo-mgr's journal was full of "database is locked"
  // from exactly this line while two dispatchers overlapped. IMMEDIATE takes the
  // write lock up front, where busy_timeout does apply. The comment here claimed
  // IMMEDIATE for a long time while the code did not (#580).
  const txn = bus.db.transaction(() => {
    const candidates = bus.db
      .query<{ event_id: number; subscriber_id: number }, [number, number]>(
        `SELECT event_id, subscriber_id
           FROM deliveries
          WHERE status = 'pending' AND next_attempt_at <= ?
          ORDER BY next_attempt_at ASC
          LIMIT ?`,
      )
      .all(now, n);
    if (candidates.length === 0) return [] as ClaimRow[];

    const claimed: ClaimRow[] = [];
    for (const c of candidates) {
      const updated = bus.db.run(
        `UPDATE deliveries
           SET status = 'running', started_at = ?, attempts = attempts + 1
         WHERE event_id = ? AND subscriber_id = ? AND status = 'pending'`,
        [now, c.event_id, c.subscriber_id],
      );
      if (updated.changes === 0) continue;
      const row = bus.db
        .query<ClaimRow, [number, number]>(
          `SELECT d.event_id, d.subscriber_id, d.attempts,
                  e.type, e.payload,
                  s.name AS subscriber_name, s.handler, s.max_attempts, s.timeout_ms
             FROM deliveries d
             JOIN events e ON e.id = d.event_id
             JOIN subscribers s ON s.id = d.subscriber_id
            WHERE d.event_id = ? AND d.subscriber_id = ?`,
        )
        .get(c.event_id, c.subscriber_id);
      if (row) claimed.push(row);
    }
    return claimed;
  });
  return txn.immediate();
}

async function runOne(bus: Bus, claim: ClaimRow): Promise<void> {
  const id = { eventId: claim.event_id, subscriberId: claim.subscriber_id };
  const log = (msg: string) =>
    console.log(`[event-bus delivery=${claim.event_id}/${claim.subscriber_id}] ${msg}`);

  // Prefer in-process onEvent callback if registered for this subscriber.
  const inProcess = bus.onEventHandlers.get(claim.subscriber_name);
  if (inProcess) {
    try {
      const event = bus.getEvent(claim.event_id);
      if (!event) throw new Error('event vanished before handler ran');
      await withTimeout(
        Promise.resolve().then(() => inProcess.callback(event)),
        claim.timeout_ms,
      );
      bus.markSucceeded(id);
      return;
    } catch (err) {
      handleFailure(bus, id, claim, err, log);
      return;
    }
  }

  // Subprocess path. Pass EVENT_ID and EVENT_TYPE via env.
  // EVENT_BUS_DB lets the linked handler (or CLI) reopen the same db.
  const proc = Bun.spawn({
    cmd: parseCommand(claim.handler).concat([String(claim.event_id)]),
    env: {
      ...process.env,
      EVENT_ID: String(claim.event_id),
      EVENT_TYPE: claim.type,
      SUBSCRIBER_ID: String(claim.subscriber_id),
      EVENT_BUS_DB: dbPathOf(bus),
    },
    stdout: 'pipe',
    stderr: 'pipe',
  });

  let timedOut = false;
  const killer = setTimeout(() => {
    timedOut = true;
    try {
      proc.kill();
    } catch {
      /* ignore */
    }
  }, claim.timeout_ms);

  let exitCode: number | null = null;
  let signalCode: string | null = null;
  let output = '';
  try {
    exitCode = await proc.exited;
    signalCode = proc.signalCode;
    // Fall back to stdout when stderr is empty: plenty of handlers report their
    // failure on stdout (the celilo CLI's own error rendering does), and an
    // empty tail turns a handler that fails identically every tick into one
    // that looks like it never ran at all — which is exactly how
    // celilo-mgmt.registry-poll hid for weeks (its "Module not found: 5517"
    // went to stdout). Diagnosis needs whichever stream spoke.
    const [err, out] = await Promise.all([
      new Response(proc.stderr).text(),
      new Response(proc.stdout).text(),
    ]);
    output = err.trim() ? err : out;
  } catch (err) {
    output = String(err);
  } finally {
    clearTimeout(killer);
  }

  // Re-read delivery state. If the handler wrote ack/fail explicitly
  // (linked via markSucceeded/markFailed, or unlinked via the CLI),
  // that's authoritative. Otherwise, use exit code.
  const post = bus.getDelivery(claim.event_id, claim.subscriber_id);
  if (post && post.status === 'succeeded') return;
  if (post && (post.status === 'failed' || post.status === 'abandoned')) {
    return;
  }

  if (timedOut) {
    handleFailure(
      bus,
      id,
      claim,
      new Error(`handler timed out after ${claim.timeout_ms}ms`),
      log,
      output,
    );
    return;
  }
  if (exitCode === 0) {
    bus.markSucceeded(id);
    return;
  }
  handleFailure(bus, id, claim, new Error(describeHandlerExit(exitCode, signalCode)), log, output);
}

/**
 * Exit codes a handler killed by a signal comes back with.
 *
 * Only the ones that actually reach an operator. A signal outside this map
 * still gets named by number, which is strictly better than nothing and better
 * than an incomplete table pretending to be complete.
 */
const SIGNAL_EXITS: Record<number, string> = {
  130: 'SIGINT',
  134: 'SIGABRT',
  137: 'SIGKILL',
  139: 'SIGSEGV',
  143: 'SIGTERM',
};

/**
 * Extra context for signals whose usual cause is not guessable from the name.
 *
 * SIGKILL is the one that matters. It cannot be caught or handled, so a handler
 * receiving it writes nothing about its own death — the record it leaves behind
 * is whatever the observer says, and for a year the observer said "exited with
 * code 137". That is a true statement carrying no information: an operator has
 * to already know that 137 is 128+9 and that 9 is SIGKILL and that the usual
 * sender on a Linux box under memory pressure is the OOM killer.
 *
 * Nothing else here sends SIGKILL. The dispatcher's own timeout kill is SIGTERM
 * and is reported by the `timedOut` branch before this code is reached, so a
 * SIGKILL arriving here came from outside celilo.
 *
 * Not hypothetical: celilo#685 was 20+ consecutive forgejo backups killed at
 * ~1.9 GB on a 3784 MB management server, and the anonymous 137 is why it read
 * as a mystery for a day rather than as "you ran out of memory".
 */
const SIGNAL_HINTS: Record<string, string> = {
  SIGKILL:
    'the process was killed outright and could not report anything itself; on a Linux host the usual sender is the OOM killer, so check free memory against what this handler needs',
  SIGSEGV: 'the process crashed',
};

/**
 * Say what actually happened to a handler process, not just its exit number.
 *
 * Bun reports a signal death two ways depending on how the process was
 * spawned and what sat between: `signalCode` set with a null `exitCode`, or an
 * `exitCode` of 128+signal with no `signalCode`. Both are handled — celilo#685
 * was observed in the second shape, and trusting only the first would have
 * translated nothing.
 */
export function describeHandlerExit(exitCode: number | null, signalCode: string | null): string {
  const signal = signalCode ?? (exitCode === null ? null : SIGNAL_EXITS[exitCode]);
  if (!signal) return `handler exited with code ${exitCode}`;

  const code = exitCode === null ? '' : ` (exit code ${exitCode})`;
  const hint = SIGNAL_HINTS[signal];
  return `handler was killed by ${signal}${code}${hint ? ` — ${hint}` : ''}`;
}

function handleFailure(
  bus: Bus,
  id: { eventId: number; subscriberId: number },
  claim: ClaimRow,
  err: unknown,
  log: (msg: string) => void,
  outputTail?: string,
): void {
  const message = outputTail
    ? `${(err as Error).message}\n--- handler output (last 1000 chars) ---\n${outputTail.slice(-1000)}`
    : (err as Error).message;
  const errorPayload = { message, stack: (err as Error).stack };

  if (claim.attempts >= claim.max_attempts) {
    log(`abandoned after ${claim.attempts} attempts: ${message}`);
    bus.markFailed(id, errorPayload, { abandoned: true });
    return;
  }
  const backoff = backoffMs(claim.attempts);
  const retryAfter = Date.now() + backoff;
  log(`failed (attempt ${claim.attempts}/${claim.max_attempts}); retrying in ${backoff}ms`);
  bus.markFailed(id, errorPayload, { retryAfter });
}

/**
 * Exponential backoff with a cap. attempt=1 → 1s, attempt=2 → 2s,
 * attempt=3 → 4s, capped at 5 minutes.
 */
function backoffMs(attempt: number): number {
  const base = 1000 * 2 ** Math.max(0, attempt - 1);
  return Math.min(base, 5 * 60 * 1000);
}

/**
 * Reset deliveries that have been in `running` longer than
 * `timeout_ms × multiplier`. Their dispatcher presumably crashed.
 *
 * This runs every tick of runDispatcher; it's also the recovery
 * mechanism on startup after a previous-process crash.
 */
function sweepStaleRunning(bus: Bus, multiplier: number): number {
  const now = Date.now();
  const result = bus.db.run(
    `UPDATE deliveries
       SET status = 'pending', next_attempt_at = ?, started_at = NULL,
           last_error = COALESCE(last_error, 'sweep-reset (likely dispatcher crash)')
     WHERE status = 'running'
       AND started_at IS NOT NULL
       AND started_at < (
         SELECT (? - (s.timeout_ms * ?))
           FROM subscribers s WHERE s.id = deliveries.subscriber_id
       )`,
    [now, now, multiplier],
  );
  return result.changes;
}

export interface RecoveryResult {
  recovered: boolean;
  stuckCount: number;
  lastHeartbeatAgeMs: number | null;
}

/**
 * Run the crash-recovery sweep without starting the dispatcher loop.
 * Useful from tests, from `event-bus repair`, and from celilo's
 * emergency / startup paths. runDispatcher calls this internally on
 * startup.
 */
export function recoverFromCrash(
  bus: Bus,
  opts?: { staleHandlerMultiplier?: number; pollIntervalMs?: number },
): RecoveryResult {
  return crashRecovery(bus, {
    ...DEFAULT_DISPATCHER,
    staleHandlerMultiplier:
      opts?.staleHandlerMultiplier ?? DEFAULT_DISPATCHER.staleHandlerMultiplier,
    pollIntervalMs: opts?.pollIntervalMs ?? DEFAULT_DISPATCHER.pollIntervalMs,
  });
}

function crashRecovery(bus: Bus, cfg: Required<DispatcherOptions>): RecoveryResult {
  const now = Date.now();
  const hb = bus.db
    .query<{ last_heartbeat: number; pid: number }, []>(
      'SELECT last_heartbeat, pid FROM dispatcher_heartbeat ORDER BY last_heartbeat DESC LIMIT 1',
    )
    .get();

  // No prior dispatcher → fresh start, nothing to recover.
  if (!hb) return { recovered: false, stuckCount: 0, lastHeartbeatAgeMs: null };

  const age = now - hb.last_heartbeat;
  // Heartbeat too fresh to be a crash. If pid differs, log a warning —
  // could mean another dispatcher is running. (Single-dispatcher v1
  // doesn't try to negotiate; we trust the operator.)
  if (age < cfg.pollIntervalMs * 3) {
    if (hb.pid !== process.pid) {
      console.warn(
        `[event-bus] another dispatcher (pid ${hb.pid}) appears active. Continuing — ensure only one dispatcher per bus.`,
      );
    }
    return { recovered: false, stuckCount: 0, lastHeartbeatAgeMs: age };
  }

  // Stale heartbeat. Reset stuck running rows.
  const stuckCount = sweepStaleRunning(bus, cfg.staleHandlerMultiplier);
  if (stuckCount > 0) {
    console.warn(
      `[event-bus] crash recovery: heartbeat ${age}ms old, reset ${stuckCount} stuck running deliveries.`,
    );
  } else {
    console.log(
      `[event-bus] heartbeat ${age}ms old (likely a clean shutdown), no stuck deliveries.`,
    );
  }
  // Best-effort emit of recovery notice. If schema doesn't have this
  // event registered, fall through silently — internal events bypass
  // schema validation.
  try {
    bus.emitRaw('_internal.dispatcher.recovered_from_crash', {
      stuck_count: stuckCount,
      last_heartbeat_age_ms: age,
      prior_pid: hb.pid,
    });
  } catch {
    /* no-op */
  }
  return { recovered: true, stuckCount, lastHeartbeatAgeMs: age };
}

function parseCommand(handler: string): string[] {
  // Naive shell-like split. Good enough for v1 — handlers are
  // typically `bun /path/to/handler.ts` with no embedded quoting.
  // If a handler legitimately needs quoted args, document a
  // structured `handler` shape later (e.g. JSON array).
  return handler.trim().split(/\s+/);
}

function dbPathOf(bus: Bus): string {
  // bun:sqlite Database has a .filename property; cast through
  // because the type isn't fully exposed.
  return (bus.db as unknown as { filename: string }).filename;
}

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}

async function withTimeout<T>(p: Promise<T>, ms: number): Promise<T> {
  return new Promise<T>((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error(`timed out after ${ms}ms`)), ms);
    p.then(
      (v) => {
        clearTimeout(timer);
        resolve(v);
      },
      (err) => {
        clearTimeout(timer);
        reject(err);
      },
    );
  });
}