"""Worker de la file `jobs` (spec 2026-08-31 « socle jobs » §2).

Polling Postgres `FOR UPDATE SKIP LOCKED`, sans broker : une table et un process
(architecture §7). Un process, un job à la fois ; la montée en charge se fait en lançant
N process identiques — `SKIP LOCKED` les répartit nativement.

Le worker est **neutre** : il ne connaît aucun type métier, seulement le registre. Les
modules qui possèdent des types les enregistrent à l'import de leur code ; en T0 il n'y en
a aucun (les imports de modules métier arriveront avec eux).

Usage : uv run python -m app.jobs.worker
"""

import asyncio
import contextlib
import logging
import signal
import time
import uuid
from collections.abc import Mapping
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import case, func, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.core.config import get_settings
from app.jobs import registry
from app.jobs.loader import load_job_types
from app.jobs.models import Job

logger = logging.getLogger(__name__)

DEFAULT_POLL_INTERVAL = 1.0
DEFAULT_SWEEP_INTERVAL = 60.0
DEFAULT_ZOMBIE_AFTER = timedelta(minutes=10)
# Erreur d'infrastructure (Postgres qui redémarre, réseau coupé) : la boucle attend puis
# retente, de 1 s à 30 s. Elle ne meurt jamais dessus — sans supervision, une file arrêtée
# en silence est pire qu'une file lente.
DEFAULT_BACKOFF_MIN = 1.0
DEFAULT_BACKOFF_MAX = 30.0
# Au-delà, l'erreur n'a plus rien de transitoire : le log passe en CRITICAL pour qu'une
# file gelée reste visible même sans supervision.
FAILURES_BEFORE_ALERT = 5
MAX_ATTEMPTS_ERROR = "max attempts exceeded"
MAX_ERROR_LENGTH = 500  # le détail circonstancié va dans `report`, pas dans `error`


def _heartbeat(job: Job) -> datetime | None:
    """Dernier signe de vie du job : `progress.updated_at`, sinon `started_at` (spec §2.5).

    **Cette fonction est totale** : `progress` est du jsonb de forme libre, donc n'importe
    quelle valeur peut s'y trouver — un objet, mais aussi `[1, 2]`, `"x"` ou `3`, qui
    désérialisent en `list`/`str`/`int`. Tout ce qui n'est pas un horodatage lisible retombe
    sur `started_at`, sans jamais lever : une exception ici ferait échouer le sweep, donc
    (la boucle ne mourant plus, mais faisant `continue`) gèlerait toute la file pour une
    seule ligne empoisonnée.
    """
    progress = job.progress
    raw = progress.get("updated_at") if isinstance(progress, dict) else None
    if isinstance(raw, str):
        with contextlib.suppress(ValueError):
            beat = datetime.fromisoformat(raw)
            # Un horodatage sans fuseau (jamais écrit par `save_progress`) est lu en UTC :
            # comparer un naïf à un aware lèverait, et ferait retomber le cas dans le bug.
            return beat if beat.tzinfo is not None else beat.replace(tzinfo=UTC)
    return job.started_at


def _is_zombie(job: Job, threshold: datetime) -> bool:
    beat = _heartbeat(job)
    # Un `running` sans aucun horodatage (état qu'aucun claim ne produit) est balayé.
    return beat is None or beat < threshold


def _short(exc: Exception) -> str:
    return f"{type(exc).__name__}: {exc}"[:MAX_ERROR_LENGTH]


async def claim(session: AsyncSession) -> Job | None:
    """Prend le plus vieux job `queued`, **sans commiter** : la ligne reste verrouillée.

    Un seul UPDATE : le sous-select verrouille la ligne (`FOR UPDATE SKIP LOCKED`) et la
    même instruction la fait passer à `running`. Deux workers concurrents ne peuvent donc
    jamais prendre le même job — le second saute la ligne verrouillée et prend la suivante.

    La `dedup_key` est relâchée ici pour les types `dedup_scope='queue'` : le travail
    redevient enfilable dès que son run démarre (anti lost-wakeup, spec §1).
    """
    candidate = (
        select(Job.id)
        .where(Job.status == "queued")
        .order_by(Job.created_at)
        .limit(1)
        .with_for_update(skip_locked=True)
        .scalar_subquery()
    )
    job_id = await session.scalar(
        update(Job)
        .where(Job.id == candidate)
        .values(
            status="running",
            started_at=func.now(),
            attempts=Job.attempts + 1,
            dedup_key=case(
                (Job.type.in_(registry.queue_scoped_names()), None), else_=Job.dedup_key
            ),
        )
        .returning(Job.id)
        .execution_options(synchronize_session=False)
    )
    if job_id is None:
        return None
    return await _reload(session, job_id)


async def _reload(session: AsyncSession, job_id: uuid.UUID) -> Job:
    job = await session.scalar(
        select(Job).where(Job.id == job_id).execution_options(populate_existing=True)
    )
    assert job is not None  # la ligne vient d'être écrite dans cette transaction
    return job


async def finish(
    session: AsyncSession,
    job_id: uuid.UUID,
    status: str,
    *,
    attempts: int,
    report: Mapping[str, Any] | None = None,
    error: str | None = None,
) -> Job | None:
    """Termine un job : `done`/`failed`, `finished_at`, et relâchement de la `dedup_key`.

    La clé est toujours mise à NULL : pour `dedup_scope='queue'` elle l'était déjà au claim,
    pour `dedup_scope='active'` c'est ici que le travail redevient enfilable (spec §2.3).

    `attempts` est le **jeton de propriété** pris au claim (fencing token). Un worker vivant
    mais silencieux plus longtemps que le seuil de zombie se fait voler son job par le
    sweep ; s'il finalisait sans garde, il écraserait le run de son repreneur et relâcherait
    la `dedup_key` d'un type `'active'` pendant que l'autre court — la garantie « jamais deux
    instances actives » tomberait. La garde `status='running' AND attempts=:attempts` fait de
    cette finalisation périmée un no-op logué : `None` est rendu, la ligne n'est pas touchée.
    """
    values: dict[str, Any] = {
        "status": status,
        "error": error,
        "finished_at": func.now(),
        "dedup_key": None,
    }
    # `report` n'est écrasé que si le job en rend un : un rapport partiel écrit par le
    # handler avant son échec survit à la finalisation.
    if report is not None:
        values["report"] = dict(report)
    result = await session.execute(
        update(Job)
        .where(Job.id == job_id, Job.status == "running", Job.attempts == attempts)
        .values(**values)
        .execution_options(synchronize_session=False)
    )
    await session.commit()
    if result.rowcount == 0:
        logger.warning(
            "job %s : finalisation périmée ignorée (jeton attempts=%s) — le job a été repris",
            job_id,
            attempts,
        )
        return None
    return await _reload(session, job_id)


async def _call_on_fail(session: AsyncSession, job_type: registry.JobType | None, job: Job) -> None:
    """Best-effort : le module remet son état de domaine en cohérence après un `failed`."""
    if job_type is None or job_type.on_fail is None:
        return
    try:
        await job_type.on_fail(session, job)
        await session.commit()
    except Exception:
        await session.rollback()
        logger.exception("job %s (%s) : callback on_fail en échec", job.id, job.type)


async def execute(session: AsyncSession, job: Job) -> Job | None:
    """Dispatche vers le handler du registre puis finalise. Ne lève jamais (spec §2.3).

    L'identité et le jeton de propriété sont capturés **avant** d'appeler le handler : un
    `rollback()` expire les attributs de l'instance, et les relire déclencherait un
    chargement paresseux impossible en asynchrone.
    """
    job_id, job_name, attempts = job.id, job.type, job.attempts
    job_type = registry.get(job_name)
    if job_type is None:
        # Rolling deploy : un worker plus vieux que l'api voit un type qu'il ne connaît pas.
        # L'échec est définitif — une tentative consommée, jamais rejouée après correction
        # du déploiement — mais bruyant et traçable, jamais avalé (comportement assumé).
        logger.error("job %s : type inconnu %r, échec définitif", job_id, job_name)
        return await finish(
            session,
            job_id,
            "failed",
            attempts=attempts,
            error=f"type de job inconnu : {job_name}",
        )
    try:
        report = await job_type.handler(session, job)
    except Exception as exc:
        # Échec déterministe : le rejeu automatique n'aiderait pas — `failed` immédiat et
        # définitif, jamais un worker mort.
        logger.exception("job %s (%s) : handler en échec", job_id, job_name)
        await session.rollback()
        failed = await finish(session, job_id, "failed", attempts=attempts, error=_short(exc))
        if failed is not None:
            await _call_on_fail(session, job_type, failed)
        return failed
    return await finish(session, job_id, "done", attempts=attempts, report=report)


class Worker:
    def __init__(
        self,
        session_factory: async_sessionmaker[AsyncSession],
        *,
        poll_interval: float = DEFAULT_POLL_INTERVAL,
        sweep_interval: float = DEFAULT_SWEEP_INTERVAL,
        zombie_after: timedelta = DEFAULT_ZOMBIE_AFTER,
        backoff_min: float = DEFAULT_BACKOFF_MIN,
        backoff_max: float = DEFAULT_BACKOFF_MAX,
    ) -> None:
        self._session_factory = session_factory
        self._poll_interval = poll_interval
        self._sweep_interval = sweep_interval
        self._zombie_after = zombie_after
        self._backoff_min = backoff_min
        self._backoff_max = backoff_max
        self._stop = asyncio.Event()

    def request_stop(self) -> None:
        """Arrêt propre : plus aucun job n'est pris, le job courant va au bout (spec §2.4)."""
        self._stop.set()

    async def run_once(self) -> bool:
        """Prend et exécute un job. Rend False si la file est vide."""
        async with self._session_factory() as session:
            job = await claim(session)
            if job is None:
                await session.rollback()
                return False
            # Le claim est committé avant l'exécution : un crash du handler laisse un job
            # `running` (rattrapé par le sweep), jamais un job perdu de la file.
            await session.commit()
            await execute(session, job)
            return True

    async def sweep(self) -> int:
        """Requalifie les jobs zombies (worker mort à mi-course : kill -9, crash natif).

        Sous le seuil de tentatives le job repart en file — le rejeu est sûr, les jobs sont
        idempotents par contrat. Au plafond, il est `failed` définitivement : un job
        empoisonné ne monopolise jamais le worker (architecture §9).
        """
        threshold = datetime.now(UTC) - self._zombie_after
        async with self._session_factory() as session:
            # Phase 1, **sans verrou** : présélection en base sur `started_at`, qui borne le
            # heartbeat par le bas (il est toujours écrit après le démarrage), puis verdict
            # en Python sur le heartbeat (voir `_heartbeat`). Verrouiller dès ici bloquerait
            # le `save_progress` des jobs longs mais parfaitement sains.
            candidates = (
                await session.scalars(
                    select(Job)
                    .where(
                        Job.status == "running",
                        or_(Job.started_at < threshold, Job.started_at.is_(None)),
                    )
                    .order_by(Job.created_at)
                )
            ).all()
            suspects = [job.id for job in candidates if _is_zombie(job, threshold)]
            if not suspects:
                await session.rollback()
                return 0
            # Phase 2 : on ne verrouille que les suspects, et on re-tranche sur l'état
            # verrouillé — entre les deux lectures, un job a pu finir ou reprendre son
            # heartbeat. `populate_existing` force la relecture des instances déjà en
            # mémoire, sans quoi le verdict porterait sur la version périmée de la phase 1.
            locked = (
                await session.scalars(
                    select(Job)
                    .where(Job.id.in_(suspects), Job.status == "running")
                    .order_by(Job.created_at)
                    .with_for_update(skip_locked=True)
                    .execution_options(populate_existing=True)
                )
            ).all()
            zombies = [job for job in locked if _is_zombie(job, threshold)]
            if not zombies:
                await session.rollback()
                return 0
            poisoned: list[Job] = []
            for job in zombies:
                job_type = registry.get(job.type)
                max_attempts = job_type.max_attempts if job_type else registry.DEFAULT_MAX_ATTEMPTS
                if job.attempts < max_attempts:
                    # `started_at` remis à NULL : le prochain claim le repose, et le
                    # heartbeat déjà écrit reste le curseur de reprise du handler.
                    values: dict[str, Any] = {"status": "queued", "started_at": None}
                else:
                    values = {
                        "status": "failed",
                        "error": MAX_ATTEMPTS_ERROR,
                        "finished_at": func.now(),
                        "dedup_key": None,
                    }
                    poisoned.append(job)
                logger.warning(
                    "job zombie %s (%s) : %s (tentative %s/%s)",
                    job.id,
                    job.type,
                    values["status"],
                    job.attempts,
                    max_attempts,
                )
                await session.execute(
                    update(Job)
                    .where(Job.id == job.id)
                    .values(**values)
                    .execution_options(synchronize_session=False)
                )
            # Un seul commit : les verrous FOR UPDATE tiennent jusque-là.
            await session.commit()
            for job in poisoned:
                await _call_on_fail(session, registry.get(job.type), await _reload(session, job.id))
            return len(zombies)

    async def run_periodics(self) -> dict[str, int]:
        """Fait une passe sur chaque tâche périodique du registre. **Ne lève jamais.**

        C'est le planificateur du produit (spec Imports §9 : « planificateur = sweep périodique
        du worker existant ») : les imports récurrents dus y sont enfilés. Le worker, lui, reste
        neutre — il ne sait pas ce que ces tâches font.

        Chaque tâche est isolée : une session neuve (pour qu'une transaction avortée n'emporte
        pas la suivante) et un `try` propre. Une tâche cassée ne doit ni geler la file, ni
        empêcher une autre tâche de tourner, ni faire remonter une exception dans la boucle —
        où elle déclencherait le backoff et affamerait l'exécution des jobs.
        """
        done: dict[str, int] = {}
        for task in registry.periodic_tasks():
            async with self._session_factory() as session:
                try:
                    done[task.name] = await task.run(session)
                except Exception:
                    await session.rollback()
                    logger.exception("tâche périodique %s en échec", task.name)
        return done

    async def run_forever(self) -> None:
        """Boucle du process worker. Seul un signal d'arrêt en sort (spec §2.4).

        Une erreur d'infrastructure (Postgres redémarré, réseau coupé — `pool_pre_ping` ne
        couvre que le checkout) est loguée puis retentée avec un backoff exponentiel borné :
        la file ne s'arrête jamais en silence. Les échecs de job, eux, sont absorbés bien
        plus tôt par `execute`, qui ne lève pas.
        """
        # Balayage de démarrage (spec §2.5) : les zombies laissés par le process précédent,
        # avant d'entrer dans la boucle. Explicite plutôt que déduit d'un `last_sweep`
        # initial : `time.monotonic()` est l'uptime de la machine, pas l'âge du process —
        # un worker lancé par systemd au boot aurait sauté ce balayage. Sous garde comme le
        # reste : un balayage en échec ne doit pas empêcher le worker de démarrer.
        try:
            await self.sweep()
        except Exception:
            logger.exception("worker jobs : balayage de démarrage en échec, la file démarre")
        last_sweep = time.monotonic()
        backoff = self._backoff_min
        failures = 0
        while not self._stop.is_set():
            try:
                # `run_once` avant `sweep` : un balayage qui échoue en boucle ne doit jamais
                # affamer l'exécution des jobs.
                worked = await self.run_once()
                if time.monotonic() - last_sweep >= self._sweep_interval:
                    await self.sweep()
                    #  Après le balayage des zombies, sur la même cadence : les tâches
                    #  périodiques enfilent (elles ne consomment pas), donc les faire tourner
                    #  ici ne retarde jamais un job — la boucle repasse tout de suite par
                    #  `run_once`. Elles n'entrent pas dans le `try` de leur côté : elles
                    #  avalent déjà leurs propres erreurs (`run_periodics`).
                    await self.run_periodics()
                    last_sweep = time.monotonic()
            except Exception:
                failures += 1
                # Le backoff vise les pannes transitoires. Un défaut persistant (bug, base
                # inaccessible durablement) ne doit pas se traduire par une file arrêtée en
                # silence : à partir du seuil, on crie. Quand la supervision existera (T2),
                # c'est ici que le process sortira pour se faire redémarrer.
                level = logging.CRITICAL if failures >= FAILURES_BEFORE_ALERT else logging.ERROR
                logger.log(
                    level,
                    "worker jobs : %s échec(s) consécutif(s), nouvelle tentative dans %.0f s",
                    failures,
                    backoff,
                    exc_info=True,
                )
                await self._sleep(backoff)
                backoff = min(backoff * 2, self._backoff_max)
                continue
            failures = 0
            backoff = self._backoff_min
            if not worked:
                await self._sleep(self._poll_interval)

    async def _sleep(self, seconds: float) -> None:
        # Interruptible : SIGTERM n'attend pas la fin du sleep de polling.
        with contextlib.suppress(TimeoutError):
            await asyncio.wait_for(self._stop.wait(), seconds)


async def _run() -> None:
    # **Avant tout le reste** : sans cet appel, le registre est vide et le worker met
    # `failed` — « type de job inconnu » — tous les jobs de la file (voir `jobs/loader.py`).
    # C'est le seul endroit du process worker qui sait qu'il existe des types métier ; le
    # worker lui-même reste neutre.
    types = load_job_types()
    # Moteur dédié : le worker est un process autonome, il ne partage pas celui de l'api.
    engine = create_async_engine(str(get_settings().database_url), pool_pre_ping=True)
    worker = Worker(async_sessionmaker(engine, expire_on_commit=False))
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, worker.request_stop)
    logger.info("worker jobs démarré (types : %s)", ", ".join(types) or "aucun")
    try:
        await worker.run_forever()
    finally:
        logger.info("worker jobs arrêté")
        await engine.dispose()


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
    asyncio.run(_run())


if __name__ == "__main__":
    main()