import { randomUUID } from "node:crypto";
import { zValidator } from "@hono/zod-validator";
import { sentry } from "@sentry/hono/bun";
import { and, eq, inArray, sql } from "drizzle-orm";
import type { MiddlewareHandler } from "hono";
import { Hono } from "hono";
import { bearerAuth } from "hono/bearer-auth";
import { z } from "zod";
import { analyticsJobs } from "@/lib/db/domain-schema";
import { db } from "./db";
import { env } from "./env";
import { claimNextJob, reapStaleJobs, STALE_RUNNING_MINUTES } from "./queue";
import { processJob } from "./worker";
import { startWorkerPool } from "./worker-pool";

// csv-import : payload historique. `type` optionnel (défaut csv-import) pour la
// compat — l'app existante n'envoie pas de `type`.
const CsvJob = z.object({
  type: z.literal("csv-import").default("csv-import"),
  siteId: z.string().min(1),
  source: z.enum(["semrush", "mangools", "ahrefs", "seranking"]),
  objectKey: z.string().min(1),
  // CSV carries no locale for Semrush/SE-Ranking → the app passes the chosen
  // language; the worker falls back to the site's primary language when absent.
  lang: z.string().min(2).optional(),
  destination: z.enum(["reservoir", "recherche"]).default("reservoir"),
});

// site-crawl : ni source ni objectKey (pas de fichier importé).
const CrawlJob = z.object({
  type: z.literal("site-crawl"),
  siteId: z.string().min(1),
});

// serp-semantics : payload structuré (keywordId, country, urls curées SERP,
// clientUrl optionnelle pour le gap analysis) — cf. plan S3 Task 4.
const SerpSemanticsJob = z.object({
  type: z.literal("serp-semantics"),
  siteId: z.string().min(1),
  payload: z.object({
    keywordId: z.string().min(1),
    country: z.string().length(2),
    urls: z
      .array(
        z.object({
          position: z.number().int().min(1).max(10),
          url: z.string().url(),
          included: z.boolean(),
        }),
      )
      .min(1)
      .max(10),
    clientUrl: z.string().url().optional(),
  }),
});

// crawl-onpage : re-crawle les compétiteurs SERP d'un mot-clé via Firecrawl —
// payload minimal (keywordId), cf. plan Firecrawl on-page Task 2.
const CrawlOnpageJob = z.object({
  type: z.literal("crawl-onpage"),
  siteId: z.string().min(1),
  payload: z.object({
    keywordId: z.string().min(1),
  }),
});

// keyword-enrich : enrichit une sélection de mots-clés via un connecteur BYOK
// (Mangools). Payload = provider + ids (≥1). Cf. plan Connecteur Mangools Task 8.
const KeywordEnrichJob = z.object({
  type: z.literal("keyword-enrich"),
  siteId: z.string().min(1),
  payload: z.object({
    provider: z.literal("mangools"),
    keywordIds: z.array(z.string().min(1)).min(1),
  }),
});

// Union tolérante : CrawlJob, SerpSemanticsJob, CrawlOnpageJob et
// KeywordEnrichJob d'abord (discriminés sur `type`), sinon CsvJob (qui absorbe
// l'absence de `type`).
const Enqueue = z.union([
  CrawlJob,
  SerpSemanticsJob,
  CrawlOnpageJob,
  KeywordEnrichJob,
  CsvJob,
]);

// Bearer guard on mutating routes — mirrors the WORKER_API_KEY precedent
// (gsc-token route). Enforced only when ANALYTICS_API_KEY is set, so an
// unconfigured dev setup keeps working (a boot warning is logged below).
// Read per-request (not at module init) so tests can toggle the env var.
const jobsAuth: MiddlewareHandler = async (c, next) => {
  const key = process.env.ANALYTICS_API_KEY;
  if (!key) return next();
  return bearerAuth({ token: key })(c, next);
};

export const app = new Hono();

// Sentry as early as possible (before auth/routes) so it captures every request
// + unhandled error. The DSN is publishable (client-embedded), not a secret:
// SENTRY_DSN overrides it; SENTRY_DSN="" disables; off in the test runner.
const DEFAULT_SENTRY_DSN =
  "https://32fcb4bd23f264e20c4f981be40caa84@o4511579881865216.ingest.us.sentry.io/4511579919613953";
const sentryDsn = process.env.SENTRY_DSN ?? DEFAULT_SENTRY_DSN;
if (sentryDsn && process.env.NODE_ENV !== "test") {
  app.use(sentry(app, { dsn: sentryDsn }));
}

// "/jobs/*" so any future sub-route (e.g. GET /jobs/:id) is born protected.
app.use("/jobs", jobsAuth);
app.use("/jobs/*", jobsAuth);
app.post("/jobs", zValidator("json", Enqueue), async (c) => {
  const body = c.req.valid("json");
  const id = randomUUID();

  if (body.type === "site-crawl") {
    // Un seul crawl actif par site : refuse si un job site-crawl est déjà
    // queued/running pour ce site (spec §Garde).
    const [active] = await db
      .select({ id: analyticsJobs.id })
      .from(analyticsJobs)
      .where(
        and(
          eq(analyticsJobs.siteId, body.siteId),
          eq(analyticsJobs.type, "site-crawl"),
          inArray(analyticsJobs.status, ["queued", "running"]),
        ),
      )
      .limit(1);
    if (active) {
      return c.json({ error: "Un crawl est déjà en cours pour ce site." }, 409);
    }
    await db.insert(analyticsJobs).values({
      id,
      siteId: body.siteId,
      type: "site-crawl",
      status: "queued",
    });
    return c.json({ id });
  }

  if (body.type === "serp-semantics") {
    // Une seule analyse sémantique active par mot-clé : refuse si un job
    // serp-semantics est déjà queued/running pour ce keywordId (mirror de la
    // garde site-crawl, scopée `payload->>'keywordId'`).
    const [active] = await db
      .select({ id: analyticsJobs.id })
      .from(analyticsJobs)
      .where(
        and(
          eq(analyticsJobs.type, "serp-semantics"),
          inArray(analyticsJobs.status, ["queued", "running"]),
          sql`${analyticsJobs.payload}->>'keywordId' = ${body.payload.keywordId}`,
        ),
      )
      .limit(1);
    if (active) {
      return c.json(
        { error: "Une analyse sémantique est déjà en cours pour ce mot-clé." },
        409,
      );
    }
    await db.insert(analyticsJobs).values({
      id,
      siteId: body.siteId,
      type: "serp-semantics",
      payload: body.payload,
      status: "queued",
    });
    return c.json({ id });
  }

  if (body.type === "crawl-onpage") {
    // Un seul crawl-onpage actif par mot-clé : refuse si un job crawl-onpage
    // est déjà queued/running pour ce keywordId (mirror de la garde
    // serp-semantics, scopée `payload->>'keywordId'`).
    const [active] = await db
      .select({ id: analyticsJobs.id })
      .from(analyticsJobs)
      .where(
        and(
          eq(analyticsJobs.type, "crawl-onpage"),
          inArray(analyticsJobs.status, ["queued", "running"]),
          sql`${analyticsJobs.payload}->>'keywordId' = ${body.payload.keywordId}`,
        ),
      )
      .limit(1);
    if (active) {
      return c.json(
        { error: "Un crawl on-page est déjà en cours pour ce mot-clé." },
        409,
      );
    }
    await db.insert(analyticsJobs).values({
      id,
      siteId: body.siteId,
      type: "crawl-onpage",
      payload: body.payload,
      status: "queued",
    });
    return c.json({ id });
  }

  if (body.type === "keyword-enrich") {
    // Un seul enrichissement actif par site+provider : refuse si un job
    // keyword-enrich est déjà queued/running pour ce siteId et ce provider
    // (mirror des gardes ci-dessus, scopée `payload->>'provider'`).
    const [active] = await db
      .select({ id: analyticsJobs.id })
      .from(analyticsJobs)
      .where(
        and(
          eq(analyticsJobs.siteId, body.siteId),
          eq(analyticsJobs.type, "keyword-enrich"),
          inArray(analyticsJobs.status, ["queued", "running"]),
          sql`${analyticsJobs.payload}->>'provider' = ${body.payload.provider}`,
        ),
      )
      .limit(1);
    if (active) {
      return c.json(
        { error: "Un enrichissement est déjà en cours pour ce site." },
        409,
      );
    }
    await db.insert(analyticsJobs).values({
      id,
      siteId: body.siteId,
      type: "keyword-enrich",
      payload: body.payload,
      status: "queued",
    });
    return c.json({ id });
  }

  // csv-import (chemin inchangé — `type` reste au défaut de la colonne).
  await db.insert(analyticsJobs).values({
    id,
    siteId: body.siteId,
    source: body.source,
    objectKey: body.objectKey,
    lang: body.lang ?? null,
    destination: body.destination,
    status: "queued",
  });
  return c.json({ id });
});

// Verify route — only registered with SENTRY_DEBUG=1, never a permanent prod
// endpoint. Hit it once to confirm Sentry captures the error, then unset.
if (process.env.SENTRY_DEBUG === "1") {
  app.get("/debug-sentry", () => {
    throw new Error("Sentry test error (analytics service)");
  });
}

/** Repli de WORKER_CONCURRENCY (pool de claim du worker, finding revue
 *  firecrawl-onpage) — au plus ce nombre de jobs en vol simultanément. */
const DEFAULT_WORKER_CONCURRENCY = 4;
const MIN_WORKER_CONCURRENCY = 1;
const MAX_WORKER_CONCURRENCY = 16;

/** Parse + clamp `WORKER_CONCURRENCY` (entier dans [1, 16]) ; toute valeur
 *  absente, non entière ou hors bornes retombe sur le défaut (4) plutôt que de
 *  laisser une valeur aberrante piloter la concurrence du pool — même pattern
 *  que `parseMaxRenderPosition` (serp-semantics.ts). */
function parseWorkerConcurrency(raw: string | undefined): number {
  if (!raw) return DEFAULT_WORKER_CONCURRENCY;
  const n = Number.parseInt(raw, 10);
  if (
    !Number.isFinite(n) ||
    n < MIN_WORKER_CONCURRENCY ||
    n > MAX_WORKER_CONCURRENCY
  ) {
    return DEFAULT_WORKER_CONCURRENCY;
  }
  return n;
}

export type AppType = typeof app;
// Bun auto-serves this default export when the file is the entrypoint. Do NOT
// also call Bun.serve() in the worker block below — that would double-bind the
// port (EADDRINUSE). Tests import the NAMED `app` export to drive app.request().
// hostname 127.0.0.1 par défaut : seul le process web local consomme ce service ;
// le bind toutes-interfaces reposait sur le firewall du provider (finding F3,
// docs/project/ops/serveur-prod.md). HOST=0.0.0.0 pour l'exposer volontairement.
export default {
  port: Number(process.env.PORT ?? 8787),
  hostname: process.env.HOST ?? "127.0.0.1",
  fetch: app.fetch,
};

// Worker loop (runs only when this file is the entrypoint, not when imported in tests).
if (import.meta.main) {
  // Fail fast on a misconfigured environment (DATABASE_URL, and R2_* unless local).
  env();
  if (!process.env.ANALYTICS_API_KEY) {
    // Fail closed in production: an exposed deployment must never run open.
    if (process.env.NODE_ENV === "production") {
      throw new Error(
        "[analytics] ANALYTICS_API_KEY is required in production — set it (and the same value on the Next app).",
      );
    }
    console.warn(
      "[analytics] ANALYTICS_API_KEY is not set — POST /jobs is UNAUTHENTICATED. Set it (and the same value on the Next app) before exposing this service.",
    );
  }
  // Reaper: at boot, and then every minute, fail jobs orphaned in `running` by a
  // crashed worker so the UI stops polling them indefinitely.
  const reap = async (): Promise<void> => {
    try {
      const n = await reapStaleJobs();
      if (n > 0)
        console.warn(
          `[worker] reaped ${n} orphaned job(s) stuck in running > ${STALE_RUNNING_MINUTES}m`,
        );
    } catch (e) {
      console.error("[worker] reaper", e);
    }
  };
  void reap();
  setInterval(reap, 60_000);

  // Pool de claim (finding revue firecrawl-onpage) : jusqu'à WORKER_CONCURRENCY
  // jobs en vol simultanément, plutôt qu'un seul job claimé puis `await`é
  // jusqu'au bout — un crawl-onpage/site-crawl lent ne bloque plus csv-import/
  // serp-semantics pour les autres orgs. `limitRenderConcurrency` et le
  // sémaphore de render.ts restent valides (voire nécessaires) avec plusieurs
  // jobs en vol : c'est leur raison d'être (cap global process-wide).
  startWorkerPool({
    claim: claimNextJob,
    process: processJob,
    concurrency: parseWorkerConcurrency(process.env.WORKER_CONCURRENCY),
    idleDelayMs: 2000,
    onError: (e) => console.error("[worker]", e),
  });
}