/**
 * Cron: Process LLM Jobs - runs every 30 minutes as a queue-loss backstop.
 *
 * Picks up to 1 PENDING llm_job in FIFO order and processes it via runJob().
 * All active LlmJobTypes (IMAGE_APPROVAL, DEAL_MODERATION, VENDOR_VIOLATION) are
 * registered in JOB_KIND_REGISTRY. runJob() owns retry/escalate internally.
 *
 * Step 2 backstop: auto-retry stale FAILED jobs (e.g. left by pre-runner runs).
 * For DEAL_MODERATION: escalate to PENDING_APPROVAL after max retries.
 *
 * Also runs human-review SLA alerting: finds deals stuck in PENDING_APPROVAL
 * longer than llm_sla_hours (default 4h) and marks sla_alerted_at to prevent
 * duplicate alerts.
 */

import { eq, isNull, and, lte, asc } from 'drizzle-orm';
// withSentry bypassed — dynamic import('@sentry/cloudflare') fails in this worker's bundle
import { createDbService } from '@/server/services/db.js';
import { type DrizzleClient, type TxDrizzleClient } from '../db/client.js';
import { llmJobs, deals, systemConfig, users } from '../db/schema.js';
import { retryLlmJob } from '../admin/resources/llm-jobs/actions.js';
import { runJob } from '../ai/runner.js';
import { captureCaught } from '../observability/capture.server.js';
import { scrubErrorForLog } from '../observability/pii-scrub.js';
import type { NotificationInsert } from '../db/queries/live-notifications.js';
import { insertOutboxRow } from '../db/queries/outbox.js';
import { insertNotification } from '../db/queries/live-notifications.js';
import {
  markLlmJobCompleted,
  markLlmJobsFailedAndCompleted,
  resetStaleRunningJobs,
} from '../db/queries/llm-jobs.js';
import {
  markDealsSlaAlerted,
  setDealPendingApprovalIfUnderReview,
} from '../db/queries/deals-extra.js';

export interface LlmCronEnv {
  DATABASE_URL: string;
  PII_KEY: string;
  R2_BUCKET?: R2Bucket;
  OUTBOX_QUEUE?: Queue<{ outboxId: string }>;
  db?: DrizzleClient;
}

const MAX_JOBS_PER_RUN = 1;
const MAX_BREACHED_DEALS_PER_RUN = 5;
const DEFAULT_SLA_HOURS = 4;

async function getSlaHours(db: DrizzleClient): Promise<number> {
  const rows = await db
    .select({ value: systemConfig.value })
    .from(systemConfig)
    .where(eq(systemConfig.key, 'llm_sla_hours'));
  const v = rows[0]?.value;
  if (v) {
    const n = parseInt(v, 10);
    if (!isNaN(n) && n > 0) return n;
  }
  return DEFAULT_SLA_HOURS;
}

export async function runProcessLlmJobs(env: LlmCronEnv): Promise<void> {
  try {
    const db = createDbService({ DATABASE_URL: env.DATABASE_URL });

    // ── Step 0: reap RUNNING jobs abandoned by a killed worker ─────────────────
    // Worker death (CF CPU/wall-clock limit) leaves jobs RUNNING forever — step 1
    // only picks PENDING. Reset any job stuck RUNNING for >10 min back to PENDING.
    await resetStaleRunningJobs(db);

    // ── Step 1: process pending jobs (backstop — queue consumer is primary) ────
    const pendingJobs = await db
      .select({ id: llmJobs.id, jobType: llmJobs.jobType, retryCount: llmJobs.retryCount })
      .from(llmJobs)
      .where(and(eq(llmJobs.status, 'PENDING'), isNull(llmJobs.e2eRunId)))
      .orderBy(asc(llmJobs.createdAt))
      .limit(MAX_JOBS_PER_RUN);

    for (const job of pendingJobs) {
      try {
        const r = await runJob(env, job.id, db);
        if (r.status === 'retry' || r.status === 'fatal') {
          console.warn(
            JSON.stringify({
              event: 'llm_job_runner_non_ok',
              jobId: job.id,
              jobType: job.jobType,
              status: r.status,
              reason: r.reason,
            }),
          );
        }
      } catch (err) {
        console.error(
          '[process-llm-jobs] runJob threw (unexpected):',
          err instanceof Error ? err.message : String(err),
        );
      }
    }

    // ── Step 2: auto-retry stale FAILED / TIMED_OUT jobs ──────────────────────
    const staleJobs = await db
      .select({ id: llmJobs.id, retryCount: llmJobs.retryCount, jobType: llmJobs.jobType })
      .from(llmJobs)
      .where(
        and(eq(llmJobs.status, 'FAILED'), isNull(llmJobs.completedAt), isNull(llmJobs.e2eRunId)),
      )
      .orderBy(asc(llmJobs.createdAt))
      .limit(5);

    for (const job of staleJobs) {
      const retried = await retryLlmJob(db, job.id);
      if (!retried) {
        if (job.jobType === 'DEAL_MODERATION') {
          await escalateDealToHumanReview(db, job.id);
        } else {
          await markLlmJobCompleted(db, job.id);
        }
      }
    }

    // ── Step 3: SLA alerting for stuck PENDING_APPROVAL deals (batched) ────────
    const slaHours = await getSlaHours(db);
    const slaThreshold = new Date(Date.now() - slaHours * 60 * 60 * 1000);

    const slaBreachedCandidates = await db
      .select({ id: deals.id })
      .from(deals)
      .where(
        and(
          eq(deals.dealState, 'PENDING_APPROVAL'),
          isNull(deals.slaAlertedAt),
          lte(deals.createdAt, slaThreshold),
        ),
      )
      .limit(20);

    if (slaBreachedCandidates.length > MAX_BREACHED_DEALS_PER_RUN) {
      console.warn(
        JSON.stringify({
          event: 'llm_sla_breach_cap',
          found: slaBreachedCandidates.length,
          processed: MAX_BREACHED_DEALS_PER_RUN,
        }),
      );
    }

    const slaBreached = slaBreachedCandidates.slice(0, MAX_BREACHED_DEALS_PER_RUN);

    if (slaBreached.length > 0) {
      const admins = await db
        .select({ id: users.id })
        .from(users)
        .where(eq(users.isAdmin, true))
        .limit(100);

      const dealIds = slaBreached.map((d) => d.id);
      await markDealsSlaAlerted(db, dealIds);

      const expiresAt = new Date(Date.now() + 7 * 24 * 60 * 60 * 1000);
      const notifValues: NotificationInsert[] = [];

      for (const deal of slaBreached) {
        for (const admin of admins) {
          notifValues.push({
            userId: admin.id,
            event: 'admin.llm_sla_breach',
            tier: 'transactional',
            titleHe: `עסקה תקועה באישור — SLA הופר`,
            titleEn: `Deal stuck in review — SLA breached`,
            bodyHe: `דיל ${deal.id} עבר את זמן הבדיקה האנושי`,
            bodyEn: `Deal ${deal.id} exceeded human-review SLA`,
            link: `/admin/moderation/deals/${deal.id}`,
            payload: { dealId: deal.id },
            expiresAt,
          });
        }
      }

      if (notifValues.length > 0) {
        const notifRows: Array<{ id: string; userId: string }> = [];
        for (const notif of notifValues) {
          const row = await insertNotification(db, notif);
          notifRows.push({ id: row.id, userId: row.userId });
        }

        const outboxValues = notifRows.map((row) => ({
          aggregateType: 'notification' as const,
          aggregateId: row.id,
          eventType: 'notification.created' as const,
          payload: { notificationId: row.id, userId: row.userId },
        }));

        const outboxRows: Array<{ id: string }> = [];
        for (const outboxValue of outboxValues) {
          outboxRows.push(await insertOutboxRow(db, outboxValue));
        }

        if (env.OUTBOX_QUEUE) {
          await Promise.all(
            outboxRows.map((row) =>
              env.OUTBOX_QUEUE!.send({ outboxId: row.id }).catch((e: unknown) => {
                captureCaught(e, { scope: 'cron.llm.outbox_queue_send', severity: 'warning' });
              }),
            ),
          );
        }
      }
    }
  } catch (err) {
    console.error(
      JSON.stringify({
        event: 'process_llm_jobs_fatal',
        error: scrubErrorForLog(err),
      }),
    );
  }
}

export async function runOwnedE2eLlmJob(
  env: LlmCronEnv,
  runId: string,
  jobId: string,
): Promise<Awaited<ReturnType<typeof runJob>>> {
  const db = env.db ?? createDbService({ DATABASE_URL: env.DATABASE_URL });
  const [owned] = await db
    .select({ id: llmJobs.id })
    .from(llmJobs)
    .where(and(eq(llmJobs.id, jobId), eq(llmJobs.e2eRunId, runId)))
    .limit(1);
  if (!owned) throw new Error('LLM job is not owned by E2E run');
  return runJob(env, jobId, db as TxDrizzleClient);
}

// ─── Helpers ────────────────────────────────────────────────────────────────────────────────

async function getDealIdForJob(db: DrizzleClient, jobId: string): Promise<string | null> {
  const rows = await db
    .select({ targetId: llmJobs.targetId })
    .from(llmJobs)
    .where(eq(llmJobs.id, jobId));
  return rows[0]?.targetId ?? null;
}

/** Fail-safe escalation when LLM cannot process after max retries. */
async function escalateDealToHumanReview(db: DrizzleClient, jobId: string): Promise<void> {
  const dealId = await getDealIdForJob(db, jobId);
  if (!dealId) return;

  await setDealPendingApprovalIfUnderReview(db, dealId);
  await markLlmJobsFailedAndCompleted(
    db,
    jobId,
    'Max retries exceeded - escalated to human review',
  );
}