/**
 * apps/worker — automation run watchdog (TASK-0113 §9).
 *
 * A repeatable maintenance job (every 5 min) that recovers runs stuck in a
 * non-terminal state after a worker crash (a SIGKILL mid-run leaves the row
 * RUNNING; an enqueued-but-never-processed job leaves it PENDING).
 *
 * Reclaim rules:
 *  - RUNNING rows with `startedAt` older than 15 min are STALE (the BullMQ
 *    lockDuration of 60 s / stalledInterval of 30 s already reprocess
 *    in-process stalls; this is the cross-restart backstop):
 *      - retryCount < cap → hop to PENDING (retryCount + 1) and re-enqueue
 *        the same runId (the run resumes from action 1 — idempotency notes
 *        in the pipeline apply);
 *      - retryCount >= cap → FAILED `WORKER_CRASHED` (terminal) + the same
 *        notification path as dead-letter (admin notification + audit +
 *        counter) via `deadLetterRun` (which records FAILED→DEAD_LETTER-safe
 *        bookkeeping).
 *  - PENDING rows with `createdAt` older than 15 min are STALE (enqueued but
 *    never picked up — e.g. the enqueue succeeded but the worker was down):
 *    re-enqueued the same way (no retryCount bump — they never started).
 *
 * Idempotent: a run already reclaimed by another watchdog pass (its status
 * moved on) is skipped by the transition assert. The query uses the
 * `automation_runs(status)` index — fine at MVP scale (a partial index is a
 * Phase-15 optimization, per the spec).
 */
import type { PrismaClient } from "@prisma/client";
import type { Redis } from "ioredis";
import { asJson } from "../../lib/prisma.js";
import { getLogger } from "../../logger.js";
import type { WorkerEnv } from "../../config.js";
import { runLog } from "./log.js";
import { enqueueRunJob } from "./matcher.js";
import { autoRunMaxRetries } from "./pipeline.js";

const log = getLogger("automation-watchdog");

/** Stale threshold (spec §9: 15 min). */
export const STALE_MS = 15 * 60_000;

export interface WatchdogResult {
  reclaimed: number;
  deadLettered: number;
  checked: number;
}

/** Run one watchdog pass. Returns the counts for observability. Never throws. */
export async function runWatchdog(
  prisma: PrismaClient,
  redis: Redis,
  env: WorkerEnv
): Promise<WatchdogResult> {
  const now = Date.now();
  const staleBefore = new Date(now - STALE_MS);
  const maxRetries = autoRunMaxRetries(env);
  const result: WatchdogResult = { reclaimed: 0, deadLettered: 0, checked: 0 };

  // RUNNING + startedAt < now-15m.
  const runningStale = await prisma.automationRun.findMany({
    where: { status: "RUNNING", startedAt: { lt: staleBefore } },
    select: { id: true, automationId: true, workspaceId: true, triggerEvent: true, retryCount: true, createdAt: true, status: true }
  });
  // PENDING + createdAt < now-15m (enqueued but never started).
  const pendingStale = await prisma.automationRun.findMany({
    where: { status: "PENDING", createdAt: { lt: staleBefore } },
    select: { id: true, automationId: true, workspaceId: true, triggerEvent: true, retryCount: true, createdAt: true, status: true }
  });

  for (const run of [...runningStale, ...pendingStale]) {
    result.checked++;
    try {
      await reclaimRun(prisma, redis, env, run, maxRetries, run.status === "RUNNING");
      if (run.status === "RUNNING" && run.retryCount >= maxRetries) result.deadLettered++;
      else result.reclaimed++;
    } catch (err) {
      log.warn({ err: (err as Error).message, runId: run.id, status: run.status }, "watchdog reclaim failed");
    }
  }

  if (result.checked > 0) {
    log.info({ ...result }, "watchdog pass complete");
  }
  return result;
}

interface StaleRun {
  id: string;
  automationId: string;
  workspaceId: string;
  triggerEvent: unknown;
  retryCount: number;
  createdAt: Date;
  status: string;
}

async function reclaimRun(
  prisma: PrismaClient,
  redis: Redis,
  env: WorkerEnv,
  run: StaleRun,
  maxRetries: number,
  wasRunning: boolean
): Promise<void> {
  const te = (run.triggerEvent as Record<string, unknown> | null) ?? {};
  const triggerType = typeof te.type === "string" ? te.type : "unknown";
  const depth = (te.automationContext as { depth?: number } | undefined)?.depth ?? 0;

  const automation = await prisma.automation.findUnique({
    where: { id: run.automationId },
    select: { id: true, name: true, workspaceId: true }
  });
  const automationName = automation?.name ?? "unknown automation";

  const reenqueue = async (): Promise<void> => {
    await enqueueRunJob(redis, {
      runId: run.id,
      automationId: run.automationId,
      workspaceId: run.workspaceId,
      triggerEvent: te,
      eventId: (te.eventId as string) ?? run.id
    });
  };

  if (wasRunning) {
    if (run.retryCount < maxRetries) {
      await prisma.automationRun.update({
        where: { id: run.id },
        data: { status: "PENDING", retryCount: run.retryCount + 1 }
      });
      await runLog(prisma, run.id, "warn", "watchdog reclaimed stale RUNNING run", {
        retryCount: run.retryCount + 1,
        startedAgoMs: STALE_MS
      });
      await reenqueue();
      return;
    }
    // At the cap → FAILED WORKER_CRASHED (spec §9: "marks FAILED
    // WORKER_CRASHED at the cap (log + same notification path as
    // dead-letter)" — FAILED, not DEAD_LETTER; the retry endpoint can
    // still revive it).
    try {
      await prisma.automationRun.update({
        where: { id: run.id },
        data: {
          status: "FAILED",
          finishedAt: new Date(),
          error: asJson({ code: "WORKER_CRASHED", message: "run stuck RUNNING past the watchdog threshold at the retry cap", attempts: run.retryCount })
        }
      });
      await runLog(prisma, run.id, "error", "run FAILED WORKER_CRASHED (watchdog, retry cap reached)", {
        attempts: run.retryCount,
        triggerType
      });
      const { emitAutomationExecuted } = await import("./dead-letter.js");
      await emitAutomationExecuted({
        redis,
        workspaceId: run.workspaceId,
        automationId: run.automationId,
        runId: run.id,
        status: "FAILED",
        triggerType,
        durationMs: STALE_MS,
        error: { code: "WORKER_CRASHED" },
        depth
      });
      const { notifyWorkspaceAdmins } = await import("../../lib/notify.js");
      await notifyWorkspaceAdmins(
        prisma,
        redis,
        env.SMTP_HOST,
        run.workspaceId,
        `Automation "${automationName}" failed (worker crash)`,
        "The run was stuck RUNNING past the watchdog threshold and exceeded the retry cap (WORKER_CRASHED).",
        { type: "automation", id: run.automationId },
        "AUTOMATION"
      );
      await prisma.auditLog.create({
        data: {
          actorType: "SYSTEM",
          actorId: `automation:${run.automationId}`,
          workspaceId: run.workspaceId,
          resourceType: "automation_run",
          resourceId: run.id,
          action: "automation.worker_crashed",
          metadata: asJson({ automationId: run.automationId, runId: run.id, code: "WORKER_CRASHED", attempts: run.retryCount })
        }
      });
      await redis.incr(`automations:dead_letter:${run.workspaceId}`).catch(() => undefined);
      log.error({ runId: run.id, automationId: run.automationId }, "stale RUNNING run → FAILED WORKER_CRASHED (cap)");
    } catch (err) {
      log.error({ err: (err as Error).message, runId: run.id }, "watchdog crash-cap bookkeeping failed");
    }
    return;
  }

  // PENDING (never started) → re-enqueue, no retryCount bump.
  await reenqueue();
  await runLog(prisma, run.id, "warn", "watchdog re-enqueued stale PENDING run", {
    pendingForMs: STALE_MS
  });
}