import { eq, sql } from "drizzle-orm";
import { analyticsJobs } from "@/lib/db/domain-schema";
import { db } from "./db";

export type Job = typeof analyticsJobs.$inferSelect;

/** Atomically claim one queued job (SKIP LOCKED) and mark it running. */
export async function claimNextJob(): Promise<Job | null> {
  // Alias snake_case columns → the camelCase `Job` fields. Raw db.execute() bypasses
  // Drizzle's column mapping, so `returning *` would yield `object_key`/`site_id`
  // (and `job.objectKey` would be undefined). Alias explicitly to stay type-honest.
  const rows = (await db.execute(sql`
    update analytics_jobs set status = 'running', updated_at = now()
    where id = (
      select id from analytics_jobs where status = 'queued'
      order by created_at asc for update skip locked limit 1
    )
    returning
      id,
      site_id as "siteId",
      type,
      source,
      object_key as "objectKey",
      lang,
      destination,
      status,
      totals,
      error,
      payload,
      created_at as "createdAt",
      updated_at as "updatedAt"
  `)) as unknown as Job[];
  return rows[0] ?? null;
}

export async function completeJob(
  id: string,
  totals: Job["totals"],
): Promise<void> {
  await db
    .update(analyticsJobs)
    .set({ status: "done", totals, updatedAt: new Date() })
    .where(eq(analyticsJobs.id, id));
}

/** Update a running job's `totals` jsonb without touching its status — used by
 *  site-crawl to publish live progress ({phase, done, total, …}). The `totals`
 *  column carries a shape per job kind (csv-import vs site-crawl); cast at the
 *  edge so the narrow inferred column type doesn't leak the union here. */
export async function updateJobTotals(
  id: string,
  totals: Record<string, unknown>,
): Promise<void> {
  await db
    .update(analyticsJobs)
    .set({ totals: totals as Job["totals"], updatedAt: new Date() })
    .where(eq(analyticsJobs.id, id));
}

export async function failJob(id: string, error: string): Promise<void> {
  await db
    .update(analyticsJobs)
    .set({ status: "failed", error, updatedAt: new Date() })
    .where(eq(analyticsJobs.id, id));
}

/** Default staleness window for the reaper (minutes). A job whose `updatedAt` is
 *  older than this while still `running` is assumed orphaned by a crashed worker. */
export const STALE_RUNNING_MINUTES = 10;

/** Reaper: mark jobs stuck in `running` (process died mid-import) as `failed`, so
 *  the UI stops polling forever. One UPDATE; safe to call repeatedly. Returns the
 *  number of jobs reaped. */
export async function reapStaleJobs(
  staleMinutes = STALE_RUNNING_MINUTES,
): Promise<number> {
  const rows = (await db.execute(sql`
    update analytics_jobs
    set status = 'failed',
        error = ${`orphaned: stuck in running > ${staleMinutes}m (worker likely crashed)`},
        updated_at = now()
    where status = 'running'
      and updated_at < now() - (${staleMinutes} * interval '1 minute')
    returning id
  `)) as unknown as { id: string }[];
  return rows.length;
}