/**
 * apps/worker — automation run pipeline (TASK-0113).
 *
 * Executes one `run-automation` job: PENDING → RUNNING →
 * SUCCEEDED/FAILED (+ retry / DEAD_LETTER). This is the single place where
 * automation side effects are observed — every run, including condition
 * skips and loop-guard skips, is recorded (exit criterion: "every run
 * logged").
 *
 * Per-run flow:
 *  1. load the run row (idempotent re-read — retries carry the same runId)
 *     + automation + actions (position order) + condition;
 *  2. PENDING → RUNNING (startedAt = now) — transition asserted; a
 *     SUCCEEDED/DEAD_LETTER row (duplicate job delivery) is dropped; a stale
 *     RUNNING row (crash redelivery) is tolerated;
 *  3. loop guard (shared `loopGuard`): self-trigger / depth ≥ 5 → SUCCEEDED
 *     with `output = { skipped: "self_trigger" | "depth_limit" }`, logged,
 *     and `automation.executed` emitted (depth 0 only);
 *  4. 50-action guard (shared `AUTOMATION_MAX_ACTIONS`) → FAILED
 *     RUN_ACTION_LIMIT (a corrupt row must not run 51 actions);
 *  5. condition gate (TASK-0111 `evaluateCondition` over a context built
 *     from the event + record): a condition-skip → SUCCEEDED
 *     `{ skipped: "conditions_not_met" }`; a corrupt/over-complex tree →
 *     FAILED CONDITION_INVALID; an over-budget eval → FAILED
 *     CONDITION_TIMEOUT;
 *  6. for each action: log step start (redacted config), execute via the
 *     executor registry (per-action timeout), log step end;
 *      - FATAL → run FAILED (`error { code, message, actionIndex, actionType,
 *        attempts }`), stop;
 *      - RETRYABLE → retry decision (§7 below).
 *
 * Retry policy (the worker manages it — NO BullMQ attempts on this job):
 *   retryCount = the run's attempts so far. A retryable error with
 *   retryCount < AUTO_RUN_MAX_RETRIES (8) records FAILED then hops to
 *   PENDING (retryCount + 1, backoff `runBackoffSeconds(retryCount+1)`
 *   delay) and re-enqueues the SAME runId; at the cap → DEAD_LETTER (§4).
 *   A FATAL error never retries (it is a validation/permission/quota/stub/
 *   timeout condition, which will not heal).
 *
 * Idempotency notes (spec §3): the retry re-executes from action 1.
 * UPDATE_RECORD / ASSIGN_USER / CHANGE_STATUS are idempotent by value-set;
 * SEND_EMAIL is deduped by the email job key `automation-email:{runId}:{i}`;
 * CREATE_PAGE is NOT value-keyed (a retry may create a second page — the run
 * log records both ids, documented per executor).
 */
import type { PrismaClient } from "@prisma/client";
import type { Redis } from "ioredis";
import { z } from "zod";
import {
  AUTOMATION_MAX_ACTIONS,
  loopGuard,
  runBackoffSeconds,
  truncateJson,
  type InterpolationVars
} from "@factory/shared";
import { asJson } from "../../lib/prisma.js";
import { getLogger } from "../../logger.js";
import type { WorkerEnv } from "../../config.js";
import { buildVars, runWithTimeout } from "./executors/helpers.js";
import type { AutomationRunContext } from "./executors/types.js";
import { classifyError, FatalError } from "./executors/classify.js";
import { executeAction, executorFor, validateAction, type ValidatedAction } from "./executors/registry.js";
import { buildConditionContext, ConditionError, evaluateCondition } from "./condition-evaluator.js";
import { assertTransition, isTerminal } from "./run-state.js";
import { runLog } from "./log.js";
import { deadLetterRun, emitAutomationExecuted } from "./dead-letter.js";

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

/** Run `output` storage cap (spec §7: 128 KB, truncated flag preserved). */
export const RUN_OUTPUT_MAX_BYTES = 128 * 1024;

/**
 * §10 metrics. The worker has no prometheus infra, so these are lightweight
 * Redis counters (documented as the metric surface; dashboards/Phase-15 can
 * scrape or replace them): `automations:runs:{status}`,
 * `automations:actions:{type}:{status}`, `automations:matched:{type}`,
 * `automations:skipped:{reason}`. Best-effort — a Redis hiccup never fails a
 * run. `automations:run_duration_ms` (histogram) is logged, not stored
 * (a real histogram is Phase-15).
 */
async function bumpMetric(redis: Redis, key: string): Promise<void> {
  try {
    await redis.incr(key);
  } catch {
    /* best-effort */
  }
}

/** Run-level retry cap (spec §2). `AUTO_RUN_MAX_RETRIES` (dev-only, default 8). */
export function autoRunMaxRetries(env: WorkerEnv): number {
  const v = env.AUTO_RUN_MAX_RETRIES;
  return Number.isInteger(v) && v >= 1 ? v : 8;
}

/** Validate the job payload (defense: a corrupt job must not crash the worker). */
export const runAutomationJobSchema = z.object({
  runId: z.string().min(1),
  automationId: z.string().min(1),
  workspaceId: z.string().min(1),
  triggerEvent: z.record(z.string(), z.unknown()),
  eventId: z.string().min(1)
});

export interface RunPipelineDeps {
  prisma: PrismaClient;
  redis: Redis;
  env: WorkerEnv;
}

function depthOf(triggerEvent: Record<string, unknown>): number {
  const d = (triggerEvent.automationContext as { depth?: number } | undefined)?.depth;
  return typeof d === "number" ? d : 0;
}

/** The BullMQ worker handler. Returns the terminal status for observability. */
export async function handleRunAutomation(
  deps: RunPipelineDeps,
  job: { data: unknown }
): Promise<string> {
  const parsed = runAutomationJobSchema.safeParse(job.data);
  if (!parsed.success) {
    log.error({ err: parsed.error.message }, "malformed run-automation job (dropped)");
    return "FAILED";
  }
  const { runId, automationId, workspaceId, triggerEvent } = parsed.data;
  const startedAt = Date.now();

  // 1. Load the run row (idempotent re-read — a retry carries the same runId).
  const run = await deps.prisma.automationRun.findUnique({ where: { id: runId } });
  if (!run) {
    log.error({ runId }, "run row missing (dropped — enqueuer should have created it)");
    return "FAILED";
  }
  const automation = await deps.prisma.automation.findUnique({
    where: { id: automationId },
    include: {
      actions: { orderBy: { position: "asc" } },
      conditions: { orderBy: { position: "asc" } }
    }
  });
  if (!automation) {
    await runLog(deps.prisma, runId, "error", "automation deleted before run (skipped)");
    await markFailed(deps, runId, run.status, { code: "AUTOMATION_DELETED", message: "automation no longer exists" }, { durationMs: 0, retries: 0 });
    return "FAILED";
  }

  const depth = depthOf(triggerEvent);
  const triggerType = typeof triggerEvent.type === "string" ? triggerEvent.type : "unknown";
  const retryCount = run.retryCount;
  const maxRetries = autoRunMaxRetries(deps.env);
  const duration = () => Date.now() - startedAt;

  // 2. State entry. Terminal rows (SUCCEEDED/DEAD_LETTER) are duplicate job
  //    deliveries → drop. A stale RUNNING row (crash redelivery) is
  //    tolerated (executors re-run; idempotency notes above).
  if (isTerminal(run.status)) {
    log.warn({ runId, status: run.status }, "duplicate job for a terminal run (dropped)");
    return run.status;
  }
  if (run.status !== "RUNNING") {
    assertTransition(run.status, "RUNNING");
    await deps.prisma.automationRun.update({ where: { id: runId }, data: { status: "RUNNING", startedAt: new Date() } });
    await runLog(deps.prisma, runId, "info", "run started", { triggerType, retryCount, depth });
  }

  const ctx = await buildRunContext(deps, runId, automation, triggerEvent);

  // 3. Loop guard (self-trigger / depth limit) — SUCCEEDED skip.
  const guard = loopGuard(
    triggerEvent.automationContext as { automationId?: string; depth?: number } | undefined,
    automationId
  );
  if (guard.blocked) {
    const output = { skipped: guard.reason };
    await deps.prisma.automationRun.update({
      where: { id: runId },
      data: { status: "SUCCEEDED", finishedAt: new Date(), output: asJson(output) }
    });
    await runLog(deps.prisma, runId, "warn", `run skipped: ${guard.reason}`, {
      depth: guard.depth,
      reason: guard.reason === "self_trigger" ? "own event re-triggered this automation" : "chain depth limit reached"
    });
    await emitAutomationExecuted({
      redis: deps.redis,
      workspaceId,
      automationId,
      runId,
      status: "SUCCEEDED",
      triggerType,
      durationMs: duration(),
      depth
    });
    await recordTerminal(deps, { runId, automationId, workspaceId, status: "SUCCEEDED", triggerType, durationMs: duration(), retries: retryCount, depth });
    return "SUCCEEDED";
  }

  // 4. 50-action guard (spec: corrupt 60-action automation → FAILED RUN_ACTION_LIMIT).
  const actions = automation.actions;
  if (actions.length > AUTOMATION_MAX_ACTIONS) {
    const code = "RUN_ACTION_LIMIT";
    const msg = `automation has ${actions.length} actions (max ${AUTOMATION_MAX_ACTIONS})`;
    await runLog(deps.prisma, runId, "error", msg, { count: actions.length });
    await markFailed(deps, runId, "RUNNING", { code, message: msg, attempts: retryCount }, { durationMs: duration(), retries: retryCount });
    return "FAILED";
  }

  // 5. Condition gate.
  const conditionTree = automation.conditions.length > 0 ? (automation.conditions[0]?.expression ?? null) : null;
  if (conditionTree) {
    try {
      const recordId = typeof triggerEvent.recordId === "string" ? triggerEvent.recordId : null;
      const cctx = await buildConditionContext(deps.prisma, triggerEvent, recordId);
      const decision = evaluateCondition(conditionTree as never, cctx);
      if (!decision.result) {
        await deps.prisma.automationRun.update({
          where: { id: runId },
          data: { status: "SUCCEEDED", finishedAt: new Date(), output: asJson({ skipped: "conditions_not_met" }) }
        });
        await runLog(deps.prisma, runId, "info", "run skipped: conditions not met", {
          evaluatedNodes: decision.evaluatedNodes,
          ms: decision.ms,
          warnings: decision.warnings.slice(0, 10)
        });
        await emitAutomationExecuted({
          redis: deps.redis,
          workspaceId,
          automationId,
          runId,
          status: "SUCCEEDED",
          triggerType,
          durationMs: duration(),
          depth
        });
        await recordTerminal(deps, { runId, automationId, workspaceId, status: "SUCCEEDED", triggerType, durationMs: duration(), retries: retryCount, depth });
        return "SUCCEEDED";
      }
    } catch (err) {
      if (err instanceof ConditionError) {
        const code = err.code;
        await runLog(deps.prisma, runId, "error", `condition evaluation failed: ${code}`, { message: err.message });
        await markFailed(deps, runId, "RUNNING", { code, message: err.message, attempts: retryCount }, { durationMs: duration(), retries: retryCount });
        return "FAILED";
      }
      throw err;
    }
  }

  // 6. Execute actions in position order.
  const output = { actions: [] as Record<string, unknown>[], totalMs: 0 };
  for (let i = 0; i < actions.length; i++) {
    const action = actions[i]!;
    const actionType = String(action.type);
    await runLog(deps.prisma, runId, "info", `action ${i + 1}/${actions.length} ${actionType} started`, {
      actionId: action.id,
      config: (action.config as Record<string, unknown>) ?? {}
    });
    const actionStarted = Date.now();
    try {
      const validated = normalizeStoredAction(action);
      const result = await executeActionWithTimeout(deps, validated, ctx, i);
      const ms = Date.now() - actionStarted;
      output.actions.push({ index: i, type: actionType, output: result.output, ms, status: "succeeded" });
      await bumpMetric(deps.redis, `automations:actions:${actionType.toLowerCase()}:succeeded`);
      await runLog(deps.prisma, runId, "info", `action ${i + 1}/${actions.length} ${actionType} done`, {
        actionId: action.id,
        ms,
        output: result.output,
        emittedEvents: result.emittedEvents
      });
      if (result.audit) {
        await auditAction(deps, ctx, result.audit, actionType, action.id);
      }
    } catch (err) {
      const cls = classifyError(err);
      const errorObj = {
        code: cls.code,
        message: (err as Error)?.message ?? String(err),
        actionIndex: i,
        actionType,
        attempts: retryCount
      };
      await runLog(deps.prisma, runId, "error", `action ${i + 1}/${actions.length} ${actionType} failed`, {
        actionId: action.id,
        code: cls.code,
        retryable: cls.retryable,
        message: errorObj.message
      });
      await bumpMetric(deps.redis, `automations:actions:${actionType.toLowerCase()}:failed`);
      if (cls.retryable && retryCount < maxRetries) {
        await scheduleRetry(deps, runId, retryCount, errorObj);
        return "RETRYING";
      }
      if (cls.retryable && retryCount >= maxRetries) {
        await deadLetterRun(deps.prisma, deps.redis, deps.env, {
          runId,
          workspaceId,
          automationId,
          automationName: automation.name,
          fromStatus: "RUNNING",
          code: cls.code,
          message: errorObj.message,
          actionIndex: i,
          actionType,
          attempts: retryCount,
          triggerType,
          durationMs: duration(),
          depth
        });
        return "DEAD_LETTER";
      }
      await markFailed(deps, runId, "RUNNING", errorObj, { durationMs: duration(), retries: retryCount });
      return "FAILED";
    }
  }

  // 7. All actions completed (or there were none) → SUCCEEDED.
  output.totalMs = duration();
  const { value: outputVal, truncated: outputTruncated } = truncateJson(output, RUN_OUTPUT_MAX_BYTES);
  await deps.prisma.automationRun.update({
    where: { id: runId },
    data: { status: "SUCCEEDED", finishedAt: new Date(), output: asJson(outputTruncated ? { ...(outputVal as Record<string, unknown>), truncated: true } : outputVal) }
  });
  await runLog(deps.prisma, runId, "info", "run completed", { actions: actions.length, ms: duration() });
  await emitAutomationExecuted({
    redis: deps.redis,
    workspaceId,
    automationId,
    runId,
    status: "SUCCEEDED",
    triggerType,
    durationMs: duration(),
    depth
  });
  await recordTerminal(deps, { runId, automationId, workspaceId, status: "SUCCEEDED", triggerType, durationMs: duration(), retries: retryCount, depth });
  return "SUCCEEDED";
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/**
 * §10 per-run structured log line + status counter. Called once per terminal
 * (or skip) outcome. Best-effort.
 */
async function recordTerminal(
  deps: RunPipelineDeps,
  args: {
    runId: string;
    automationId: string;
    workspaceId: string;
    status: string;
    triggerType: string;
    durationMs: number;
    retries: number;
    depth: number;
  }
): Promise<void> {
  log.info(
    {
      runId: args.runId,
      automationId: args.automationId,
      status: args.status,
      durationMs: args.durationMs,
      retries: args.retries,
      triggerType: args.triggerType
    },
    "automation.run"
  );
  await bumpMetric(deps.redis, `automations:runs:${args.status.toLowerCase()}`);
  // TASK-0133 audit catalog: EVERY terminal run is an audit row (the
  // catalog's `automation.executed`; required metadata `automationId`,
  // `runId`, `status`). This is the single terminal chokepoint (SUCCEEDED +
  // every markFailed path), so the audit cannot be missed. Best-effort (an
  // audit failure must not lose the run's bookkeeping). The DOMAIN event is
  // depth-0 only (event-storm guard) — the audit records all depths (a
  // complete trail, bounded by the depth guard).
  try {
    await deps.prisma.auditLog.create({
      data: {
        actorType: "SYSTEM",
        actorId: `worker:automation`,
        workspaceId: args.workspaceId || null,
        resourceType: "automation_run",
        resourceId: args.runId,
        action: "automation.executed",
        metadata: asJson({
          automationId: args.automationId,
          runId: args.runId,
          status: args.status,
          triggerType: args.triggerType,
          durationMs: args.durationMs,
          retries: args.retries,
          depth: args.depth
        })
      }
    });
  } catch (err) {
    log.warn({ err: (err as Error).message, runId: args.runId }, "automation.executed audit write failed");
  }
}

async function buildRunContext(
  deps: RunPipelineDeps,
  runId: string,
  automation: { id: string; name: string; workspaceId: string; databaseId: string | null; createdById: string },
  triggerEvent: Record<string, unknown>
): Promise<AutomationRunContext> {
  const recordId = typeof triggerEvent.recordId === "string" ? triggerEvent.recordId : null;
  const depth = depthOf(triggerEvent);
  const base = {
    runId,
    automationId: automation.id,
    automationName: automation.name,
    workspaceId: automation.workspaceId,
    databaseId: automation.databaseId,
    recordId,
    triggerEvent,
    depth,
    input: (triggerEvent.input as Record<string, unknown> | undefined) ?? null,
    actorUserId: automation.createdById,
    vars: {}
  };
  const vars: InterpolationVars = await buildVars(deps.prisma, base);
  return { ...base, vars };
}

/**
 * A stored action row may be corrupt (drift, a bad manual edit). Re-validate
 * through the shared schema; an invalid row → a FATAL INVALID_CONFIG (no
 * retry) rather than a cryptic executor crash.
 */
function normalizeStoredAction(action: { type: string; config: unknown; retries: number; timeoutMs: number }): ValidatedAction {
  return validateAction({ type: action.type, config: action.config, retries: action.retries, timeoutMs: action.timeoutMs });
}

async function executeActionWithTimeout(
  deps: RunPipelineDeps,
  validated: ValidatedAction,
  ctx: AutomationRunContext,
  index: number
): Promise<Awaited<ReturnType<typeof executeAction>>> {
  const executor = executorFor(validated.type);
  if (!executor) throw new FatalError("INVALID_CONFIG", `no executor for ${validated.type}`);
  return runWithTimeout(validated.timeoutMs, () => executor.execute(validated.config, ctx, deps, index));
}

async function auditAction(
  deps: RunPipelineDeps,
  ctx: AutomationRunContext,
  entry: { action: string; resourceType: string; resourceId: string; metadata?: Record<string, unknown> },
  actionType: string,
  actionId: string
): Promise<void> {
  try {
    await deps.prisma.auditLog.create({
      data: {
        actorType: "SYSTEM",
        actorId: `automation:${ctx.automationId}`,
        workspaceId: ctx.workspaceId,
        resourceType: entry.resourceType,
        resourceId: entry.resourceId,
        action: entry.action,
        metadata: asJson({
          ...(entry.metadata ?? {}),
          onBehalfOf: "automation",
          automationId: ctx.automationId,
          runId: ctx.runId,
          actionType,
          actionId
        })
      }
    });
  } catch (err) {
    log.warn({ err: (err as Error).message, runId: ctx.runId }, "action audit write failed");
  }
}

/**
 * Retryable error below the cap: record the FAILED hop (the error is visible
 * immediately) then hop to PENDING with retryCount + 1 and re-enqueue the
 * SAME runId with the backoff delay. State path asserted: RUNNING → FAILED
 * → PENDING (the §2 state machine); a single atomic write lands on PENDING.
 */
async function scheduleRetry(
  deps: RunPipelineDeps,
  runId: string,
  retryCount: number,
  errorObj: Record<string, unknown>
): Promise<void> {
  const nextRetry = retryCount + 1;
  const backoffSec = runBackoffSeconds(nextRetry);
  assertTransition("RUNNING", "FAILED");
  assertTransition("FAILED", "PENDING");
  await deps.prisma.automationRun.update({
    where: { id: runId },
    data: {
      status: "PENDING",
      error: asJson(errorObj),
      retryCount: nextRetry
    }
  });
  await runLog(deps.prisma, runId, "warn", `run retrying (attempt ${nextRetry})`, {
    code: errorObj.code as string,
    backoffSeconds: backoffSec
  });

  // Re-enqueue with the backoff delay (no BullMQ attempts — the pipeline is
  // the retry manager; this is a fresh job carrying the same runId).
  const { Queue } = await import("bullmq");
  const q = new Queue("automations", { connection: deps.redis });
  try {
    const row = await deps.prisma.automationRun.findUnique({
      where: { id: runId },
      select: { automationId: true, workspaceId: true, triggerEvent: true }
    });
    const te = (row?.triggerEvent as Record<string, unknown> | null) ?? {};
    await q.add(
      "run-automation",
      {
        runId,
        automationId: row?.automationId ?? "",
        workspaceId: row?.workspaceId ?? "",
        triggerEvent: te,
        eventId: (te.eventId as string) ?? runId
      },
      { delay: backoffSec * 1000, removeOnComplete: 1000, removeOnFail: 500 }
    );
  } finally {
    await q.close();
  }
}

/**
 * Mark a run FAILED (terminal; the API retry endpoint can revive it).
 * Emits `automation.executed` at depth 0. Never throws.
 */
async function markFailed(
  deps: RunPipelineDeps,
  runId: string,
  fromStatus: string,
  errorObj: Record<string, unknown>,
  opts: { durationMs: number; retries: number }
): Promise<void> {
  try {
    if (fromStatus === "SUCCEEDED" || fromStatus === "DEAD_LETTER") return; // terminal
    assertTransition(fromStatus as never, "FAILED");
    const run = await deps.prisma.automationRun.findUnique({
      where: { id: runId },
      select: { triggerEvent: true, workspaceId: true, automationId: true }
    });
    await deps.prisma.automationRun.update({
      where: { id: runId },
      data: { status: "FAILED", finishedAt: new Date(), error: asJson(errorObj) }
    });
    const te = (run?.triggerEvent as Record<string, unknown> | null) ?? {};
    const triggerType = typeof te.type === "string" ? te.type : "unknown";
    await emitAutomationExecuted({
      redis: deps.redis,
      workspaceId: run?.workspaceId ?? "",
      automationId: run?.automationId ?? "",
      runId,
      status: "FAILED",
      triggerType,
      durationMs: opts.durationMs,
      error: { code: (errorObj.code as string) ?? "UNKNOWN" },
      depth: depthOf(te)
    });
    await recordTerminal(deps, {
      runId,
      automationId: run?.automationId ?? "",
      workspaceId: run?.workspaceId ?? "",
      status: "FAILED",
      triggerType,
      durationMs: opts.durationMs,
      retries: opts.retries,
      depth: depthOf(te)
    });
  } catch (err) {
    log.warn({ err: (err as Error).message, runId }, "markFailed bookkeeping failed");
  }
}