import { eq } from "drizzle-orm";
import { sites } from "@/lib/db/domain-schema";
import { processCrawlOnpage } from "./crawl-onpage";
import { db } from "./db";
import { processKeywordEnrich } from "./keyword-enrich";
import { processCsv } from "./processor";
import { completeJob, failJob, type Job } from "./queue";
import { processSerpSemantics } from "./serp-semantics";
import { processSiteCrawl } from "./site-crawl";
import { deleteObject } from "./storage";
import { upsertCanonical } from "./upsert";

/** Default ISO-3166 alpha-2 country per market language — keeps the keyword
 *  locale consistent with the chosen `lang` (no « fr-US ») for sources that
 *  carry no per-row country. Mirrors the app's `LANG_COUNTRY`. */
const LANG_COUNTRY: Record<string, string> = {
  en: "US",
  fr: "FR",
  es: "ES",
  de: "DE",
  it: "IT",
  nl: "NL",
  pt: "PT",
};

/** csv-import (chemin historique) : parse le CSV importé → upsert des keywords. */
async function processCsvImport(job: Job): Promise<void> {
  try {
    const [site] = await db
      .select({
        lang: sites.primaryLanguage,
        countries: sites.targetCountries,
      })
      .from(sites)
      .where(eq(sites.id, job.siteId));
    // The user-picked import language wins (so « fr » yields FR even if the site
    // targets US); fall back to the site's primary language, then "en".
    const lang = job.lang ?? site?.lang ?? "en";
    // Derive the default country from the chosen language first, then the site's
    // first target country — don't hardcode "US" for sources that lack a column.
    const country = LANG_COUNTRY[lang] ?? site?.countries?.[0] ?? "US";
    // `source`/`objectKey` are now nullable (site-crawl jobs carry neither).
    // The csv-import path requires both — fail loud rather than pass null down.
    if (!job.objectKey || !job.source) {
      throw new Error("csv-import job missing objectKey/source");
    }
    const { rows, totals } = await processCsv(
      job.objectKey,
      job.source,
      lang,
      country,
    );
    const up = await upsertCanonical(
      job.siteId,
      rows,
      job.destination === "recherche" ? ["research"] : [],
    );
    await completeJob(job.id, {
      read: totals.read,
      matched: up.matched,
      created: up.created,
      upserted: up.upserted,
    });
  } catch (e) {
    await failJob(job.id, (e as Error).message); // object retained for retry
    return;
  }
  // Best-effort cleanup AFTER the job is marked done. A failure here must NOT
  // re-mark a successful import as failed — the R2 lifecycle policy reaps orphans.
  try {
    // objectKey is nullable now; only csv-import jobs (which reach here) have one.
    if (job.objectKey) await deleteObject(job.objectKey); // no retention on success
  } catch (e) {
    console.error("[worker] deleteObject failed (object retained)", e);
  }
}

/** Dispatch par registre : chaque `type` de job mappe vers son handler. Un type
 *  inconnu échoue proprement (jamais un job coincé en `running`). Ajouter un
 *  nouveau type de job = une entrée ici, sans toucher au reste de la boucle. */
const HANDLERS: Record<string, (job: Job) => Promise<void>> = {
  "csv-import": processCsvImport,
  "site-crawl": processSiteCrawl,
  "serp-semantics": processSerpSemantics,
  "crawl-onpage": processCrawlOnpage,
  "keyword-enrich": processKeywordEnrich,
};

export async function processJob(job: Job): Promise<void> {
  const handler = HANDLERS[job.type];
  if (!handler) {
    await failJob(job.id, `type de job inconnu: ${job.type}`);
    return;
  }
  await handler(job);
}