/**
 * AI job runner — single entry point for cron + queue consumers.
 *
 * runJob() NEVER throws. Returns RunResult so callers map to their
 * retry mechanism (queue.message.retry() or next cron tick).
 *
 * Outcome semantics:
 *   'ok'    — job processed or skipped (no further action needed).
 *   'retry' — transient error; retryLlmJob already called (bumps retry_count,
 *             resets to PENDING). Caller requeues or lets next cron pick up.
 *   'fatal' — terminal. onRetryExhausted + markJobFailed already called. Caller acks.
 */
import { NeonDbError } from '@neondatabase/serverless';
import type { TxDrizzleClient } from '@/server/db/client.js';
import { resolveQueueForJobType, LlmQueueConfigError } from '@/server/ai/credentials.js';
import type { AiCredentials } from '@/server/ai/credentials.js';
import { retryLlmJob } from '@/server/admin/resources/llm-jobs/actions.js';
import { JOB_KIND_REGISTRY } from '@/server/ai/kinds/registry.js';
import {
  ProviderRateLimitError,
  ProviderTransientError,
  ProviderQuotaExhaustedError,
  ProviderAuthError,
  ProviderFatalError,
  ProviderNotImplementedError,
} from '@/server/ai/providers/types.js';
import {
  loadJobRow,
  transitionToRunning,
  markJobFailed,
  markJobCompleted,
} from '@/server/ai/runner-helpers.js';
import type { RunResult } from '@/server/ai/kinds/types.js';
import { captureCaught } from '@/server/observability/capture.server.js';

export interface RunJobEnv {
  DATABASE_URL: string;
  PII_KEY: string;
  R2_BUCKET?: R2Bucket;
}

export type { RunResult };

const ERROR_CAUSE_PROPS = [
  'code',
  'severity',
  'detail',
  'hint',
  'routine',
  'position',
  'table',
  'column',
  'constraint',
  'sourceError',
] as const;

function serializeErrorCauseChain(err: unknown, maxDepth = 5): unknown[] {
  const chain: unknown[] = [];
  const seen = new WeakSet<object>();
  let current: unknown = err;

  for (let depth = 0; depth < maxDepth && current != null; depth++) {
    if (typeof current === 'object') {
      if (seen.has(current)) {
        chain.push({ circular: true });
        break;
      }
      seen.add(current);
    }

    const entry: Record<string, unknown> = {};

    if (current instanceof Error) {
      entry.name = current.name;
      entry.message = current.message;
      if (current.stack) {
        entry.stack = current.stack.split('\n').slice(0, 3).join('\n');
      }

      const record = current as unknown as Record<string, unknown>;
      for (const prop of ERROR_CAUSE_PROPS) {
        const value = record[prop];
        if (value === undefined) continue;
        if (prop === 'sourceError') {
          entry.sourceError =
            value instanceof Error
              ? value.message
              : typeof value === 'object' && value !== null && 'message' in value
                ? String((value as { message: unknown }).message)
                : String(value);
        } else {
          entry[prop] = value;
        }
      }

      current = 'cause' in current ? (current as Error & { cause?: unknown }).cause : undefined;
    } else if (typeof current === 'object') {
      const record = current as Record<string, unknown>;
      if ('name' in record) entry.name = String(record.name);
      if ('message' in record) entry.message = String(record.message);
      entry.value = String(current);
      current = record.cause;
    } else {
      entry.value = String(current);
      current = undefined;
    }

    chain.push(entry);
  }

  return chain;
}

function isTransientDbError(err: unknown): boolean {
  const seen = new WeakSet<object>();
  let current: unknown = err;

  for (let depth = 0; depth < 5 && current != null; depth++) {
    if (current instanceof NeonDbError) return true;

    if (typeof current === 'object') {
      if (seen.has(current)) break;
      seen.add(current);
    }

    current =
      current instanceof Error
        ? (current as Error & { cause?: unknown }).cause
        : typeof current === 'object' && current !== null && 'cause' in current
          ? (current as { cause?: unknown }).cause
          : undefined;
  }

  return false;
}

export async function runJob(
  env: RunJobEnv,
  jobId: string,
  db: TxDrizzleClient,
): Promise<RunResult> {
  try {
    // ── 1. Load row ──────────────────────────────────────────────────────────
    const jobRow = await loadJobRow(db, jobId);
    if (!jobRow) return { status: 'ok' };
    if (jobRow.status !== 'PENDING') return { status: 'ok' };

    // ── 2. Lookup kind ───────────────────────────────────────────────────────
    const kind = JOB_KIND_REGISTRY[jobRow.jobType];
    if (!kind) return { status: 'fatal', reason: 'unknown_kind' };

    // ── 3. Resolve provider + credentials (needed for both paths) ────────────
    let resolution: Awaited<ReturnType<typeof resolveQueueForJobType>>;
    try {
      resolution = await resolveQueueForJobType(
        db,
        env.PII_KEY,
        jobRow.jobType,
        jobRow.queueChainOverride,
      );
    } catch (err) {
      if (err instanceof LlmQueueConfigError) {
        // Permanent misconfiguration (queue missing/empty) — terminal, escalate now.
        // Retrying an unconfigured queue can never succeed, so do NOT leave it for
        // the Step-2 backstop (which would burn 3 retries and silently drop
        // non-DEAL_MODERATION kinds before any escalation fires).
        //
        // ORDER MATTERS: escalate FIRST, stamp terminal ONLY after escalation
        // succeeds. onRetryExhausted does an unguarded DB write and can throw; if
        // we stamped completed_at first, a throw here would fall to the outer catch
        // (terminal:false, which never CLEARS completed_at) and strand the row
        // permanently — backstop skips it, no human review ever fires. Escalating
        // first means a throw leaves completed_at NULL → backstop re-runs the whole
        // job → re-enters this branch → onRetryExhausted (idempotent, status-guarded)
        // retries until it lands, then we stamp terminal. Self-healing preserved.
        await kind.onRetryExhausted(db, jobRow, 'no_credentials');
        await markJobFailed(db, jobId, `no_credentials: ${err.code}`, {
          terminal: true,
        });
        return { status: 'fatal', reason: 'no_credentials' };
      }
      throw err; // transient/unknown (e.g. DB hiccup) → outer catch → retryable
    }

    const deps = {
      db,
      provider: resolution.provider,
      creds: {} as AiCredentials,
      model: resolution.model,
      r2Bucket: env.R2_BUCKET,
    };

    // ── 4. Idempotent retry: skip provider if output already stored ──────────
    // applied is informational only — always mark completed so job exits PENDING.
    // If applied=false someone else already actioned the entity; that's fine.
    if (jobRow.outputPayload != null) {
      const result = kind.hydrateResult(jobRow);
      const meta = kind.extractCompletionMeta?.(result);
      await kind.applyVerdict(deps, jobRow, result);
      await markJobCompleted(db, jobId, jobRow.outputPayload, meta);
      return { status: 'ok' };
    }

    // ── 5. Race guard: claim the job ─────────────────────────────────────────
    const claimed = await transitionToRunning(db, jobId);
    if (!claimed) return { status: 'ok' };

    // ── 6. Load input ────────────────────────────────────────────────────────
    let input: unknown;
    try {
      input = await kind.loadInput(deps, jobRow);
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      if (isTransientDbError(err)) {
        // Transient DB failure loading input — leave completed_at NULL so Step-2 retries. Do NOT escalate inline (would defeat the retry).
        await markJobFailed(db, jobId, `load_input_transient: ${msg}`, {
          terminal: false,
        });
        return { status: 'fatal', reason: 'load_input_transient' };
      }
      // Deterministic load failure (entity missing / wrong state) — terminal.
      // ORDER: escalate FIRST, stamp terminal AFTER (mirrors the config-error branch).
      // onRetryExhausted does unguarded DB writes and CAN throw; stamping terminal
      // first then throwing here drops to the outer catch (terminal:false, which never
      // CLEARS completed_at) → row stranded, never human-reviewed. Escalate-first
      // leaves completed_at NULL on a throw → backstop self-heals via re-run
      // (onRetryExhausted is idempotent across all queued kinds).
      await kind.onRetryExhausted(db, jobRow, msg);
      await markJobFailed(db, jobId, `load_input_failed: ${msg}`, {
        terminal: true,
      });
      return { status: 'fatal', reason: 'load_input_failed' };
    }

    // ── 7. Call provider ─────────────────────────────────────────────────────
    let result: unknown;
    try {
      result = await kind.runProvider(deps, jobRow, input);
    } catch (err) {
      const isRetryable =
        err instanceof ProviderRateLimitError || err instanceof ProviderTransientError;
      const isQuotaFatal =
        err instanceof ProviderQuotaExhaustedError ||
        err instanceof ProviderAuthError ||
        err instanceof ProviderFatalError ||
        err instanceof ProviderNotImplementedError;

      if (isRetryable) {
        const willRetry = await retryLlmJob(db, jobId);
        if (willRetry) {
          return {
            status: 'retry',
            reason: err instanceof Error ? err.message : 'transient',
          };
        }
        // Budget exhausted → treat as fatal
        // ORDER: escalate-first (see load_input_failed) — terminal stamp only after escalation.
        const reason = err instanceof Error ? err.message : 'retry_exhausted';
        await kind.onRetryExhausted(db, jobRow, reason);
        await markJobFailed(db, jobId, reason);
        return { status: 'fatal', reason: 'retry_exhausted' };
      }

      if (isQuotaFatal) {
        // ORDER: escalate-first (see load_input_failed) — terminal stamp only after escalation.
        const reason = err instanceof Error ? err.message : 'provider_fatal';
        await kind.onRetryExhausted(db, jobRow, reason);
        await markJobFailed(db, jobId, reason);
        return { status: 'fatal', reason: 'provider_fatal' };
      }

      // Unknown error — propagate as fatal
      // ORDER: escalate-first (see load_input_failed) — terminal stamp only after escalation.
      const reason = err instanceof Error ? err.message : 'unknown_provider_error';
      await kind.onRetryExhausted(db, jobRow, reason);
      await markJobFailed(db, jobId, reason);
      return { status: 'fatal', reason: 'provider_fatal' };
    }

    // ── 8. Apply verdict + complete ──────────────────────────────────────────
    const meta = kind.extractCompletionMeta?.(result);
    await kind.applyVerdict(deps, jobRow, result);
    await markJobCompleted(db, jobId, result, meta);
    return { status: 'ok' };
  } catch (err) {
    // Outer catch — runJob NEVER throws. Log for operator visibility.
    console.error('[ai.runJob] cause-chain:', JSON.stringify(serializeErrorCauseChain(err)));
    console.error('[ai.runJob] uncaught:', err instanceof Error ? (err.stack ?? err.message) : err);
    captureCaught(err, { scope: 'ai.runJob', severity: 'error' });
    const failReason = 'runner_error: ' + (err instanceof Error ? err.message : String(err));
    try {
      await markJobFailed(db, jobId, failReason, { terminal: false });
    } catch (markErr) {
      console.error('[ai.runJob] markJobFailed failed:', markErr);
    }
    return { status: 'fatal', reason: 'runner_error' };
  }
}