// Auteur : Simon-Pierre Boucher — contact@spboucher.ai
/**
 * Orchestrateur d'un connecteur : discover → fetch (hash) → extract → normalize
 * → validate → persistance, avec journal structuré et statistiques de run.
 * Utilisé par scripts/cost-sync.ts (planificateur) et l'API admin.
 */
import type Database from "better-sqlite3";
import { getCostDb } from "../db";
import { CONNECTORS, type RunnableKey } from "./registry";
import { closeLabourPeriods, ensureConnectorSchema, finishRun, isSourceActive, lastSuccessfulSync, pushLog, saveIndex, saveLabour, savePrice, saveRawObservation, saveUnchangedMarker, startRun } from "./store";
import type { CanonicalObservation, RawDocument, RunLogEntry, RunOptions, RunStats } from "./types";
import { APCHQ_PARSER_VERSION } from "./apchq";
import { STATCAN_PARSER_VERSION } from "./statcan";
import { CCQ_PARSER_VERSION } from "./ccq";
import { RETAIL_PARSER_VERSION } from "./retail";

const PARSER_VERSION: Record<RunnableKey, string> = { apchq: APCHQ_PARSER_VERSION, ccq: CCQ_PARSER_VERSION, statcan: STATCAN_PARSER_VERSION, canac: RETAIL_PARSER_VERSION, bmr: RETAIL_PARSER_VERSION, patrickmorin: RETAIL_PARSER_VERSION, rona: RETAIL_PARSER_VERSION, homedepot: RETAIL_PARSER_VERSION };

/** Le connecteur est-il dû (fréquence nominale écoulée) ? */
export function isDue(key: RunnableKey, d: Database.Database = getCostDb()): boolean {
  const last = lastSuccessfulSync(key, d);
  if (!last) return true;
  return Date.now() - new Date(last).getTime() >= CONNECTORS[key].config.refreshDays * 86400000 - 3600000;
}

export async function runConnector(key: RunnableKey, opts: RunOptions = {}): Promise<RunStats> {
  const d = opts.db ?? getCostDb();
  ensureConnectorSchema(d);
  const c = CONNECTORS[key];
  const run = startRun(key, d);
  const log = (e: RunLogEntry) => pushLog(run, e, opts.log);
  const ts = () => new Date().toISOString();
  if (!isSourceActive(key, d) && !opts.force) {
    run.status = "skipped";
    run.message = "source inactive (cost_sources.is_active = 0)";
    log({ connector: key, url: null, status: "skipped", duration: 0, observations: 0, validationErrors: 0, timestamp: ts(), message: run.message });
    return finishRun(run, d);
  }
  let docs: Awaited<ReturnType<typeof c.discover>> = [];
  try {
    docs = await c.discover({ ...opts, db: d, log });
    log({ connector: key, url: null, status: "ok", duration: 0, observations: docs.length, validationErrors: 0, timestamp: ts(), message: `${docs.length} document(s) découvert(s)` });
  } catch (e) {
    run.errors++;
    run.status = "error";
    run.message = `découverte échouée : ${(e as Error).message}`;
    log({ connector: key, url: null, status: "error", duration: 0, observations: 0, validationErrors: 0, timestamp: ts(), message: run.message });
    return finishRun(run, d);
  }
  const max = opts.maxPages ?? c.config.maxPages;
  const queue = docs.slice(0, max);
  const workers = Math.max(1, c.config.concurrency);
  let idx = 0;
  const worker = async () => {
    for (;;) {
      const i = idx++;
      if (i >= queue.length) return;
      const doc = queue[i];
      const t0 = Date.now();
      let raw: RawDocument | null = null;
      try {
        raw = await c.fetch(doc, { ...opts, db: d, log });
        run.pages++;
        if (raw.statusCode != null && raw.statusCode >= 400) {
          run.errors++;
          log({ connector: key, url: doc.url, status: "error", duration: Date.now() - t0, observations: 0, validationErrors: 0, timestamp: ts(), message: `HTTP ${raw.statusCode}` });
          continue;
        }
        if (raw.unchanged) {
          run.unchanged++;
          saveUnchangedMarker(key, raw, d);
          log({ connector: key, url: doc.url, status: "unchanged", duration: Date.now() - t0, observations: 0, validationErrors: 0, timestamp: ts() });
          continue;
        }
        const rawObs = await c.extract(raw, { ...opts, db: d, log });
        if (!rawObs.length) {
          const unavailable = key === "ccq" && !(raw.metadata.hasRateTable as boolean | undefined);
          if (unavailable) { run.status = "unavailable"; run.message = String(raw.metadata.incident ?? "aucune grille de taux détectable"); }
          else run.errors++;
          saveRawObservation(key, { externalId: null, sourceUrl: doc.url, retrievedAt: raw.fetchedAt, effectiveDate: null, title: String(raw.metadata.title ?? doc.title ?? ""), description: null, unit: null, price: null, regularPrice: null, salePrice: null, currency: "CAD", location: null, category: null, payload: { statusCode: raw.statusCode, unavailable } }, raw, "rejected", unavailable ? "source indisponible" : "aucune observation extraite (structure inattendue ou page sans prix)", PARSER_VERSION[key], d);
          log({ connector: key, url: doc.url, status: unavailable ? "unavailable" : "rejected", duration: Date.now() - t0, observations: 0, validationErrors: 1, timestamp: ts(), message: unavailable ? run.message : "aucune observation extraite" });
          continue;
        }
        const canon = await c.normalize(rawObs, raw, { ...opts, db: d, log });
        const { accepted, rejected } = c.validate(canon, { ...opts, db: d, log });
        run.observations += canon.length;
        const tx = d.transaction(() => {
          for (const o of accepted) persist(key, o, raw!, PARSER_VERSION[key], d);
          for (const r of rejected) saveRawObservation(key, rawOf(r.obs, raw!, doc.url), raw!, "rejected", r.reason, PARSER_VERSION[key], d);
        });
        tx();
        run.accepted += accepted.length;
        run.rejected += rejected.length;
        log({ connector: key, url: doc.url, status: accepted.length ? "ok" : "rejected", duration: Date.now() - t0, observations: accepted.length, validationErrors: rejected.length, timestamp: ts(), message: rejected.length ? rejected.slice(0, 3).map((r) => r.reason).join(" ; ") : undefined });
      } catch (e) {
        run.errors++;
        log({ connector: key, url: doc.url, status: "error", duration: Date.now() - t0, observations: 0, validationErrors: 0, timestamp: ts(), message: (e as Error).message });
      }
    }
  };
  await Promise.all(Array.from({ length: workers }, worker));
  if (key === "apchq" && run.accepted) closeLabourPeriods("apchq", d);
  if (run.status === "unavailable" && run.accepted === 0) return finishRun(run, d);
  return finishRun(run, d);
}

function rawOf(o: CanonicalObservation, raw: RawDocument, url: string) {
  if (o.kind === "index") return { externalId: o.indexCode, sourceUrl: url, retrievedAt: raw.fetchedAt, effectiveDate: o.period, title: o.indexCode, description: null, unit: "index", price: o.value, regularPrice: null, salePrice: null, currency: "CAD", location: o.geography, category: "index", payload: {} };
  return o.raw;
}

function persist(key: RunnableKey, o: CanonicalObservation, raw: RawDocument, parser: string, d: Database.Database): void {
  if (o.kind === "index") { saveIndex(o, d); return; }
  if (o.kind === "labour") {
    const rawId = saveRawObservation(key, o.raw, raw, "accepted", null, parser, d);
    saveLabour(key, o, rawId, d);
    return;
  }
  const rawId = saveRawObservation(key, o.raw, raw, "accepted", null, parser, d);
  const conf = Math.round(60 + (o.isRegularPrice ? 15 : 0) + (o.conversionFactor === 1 ? 10 : 5));
  savePrice(key, o, rawId, conf, false, null, d);
}