"""
Cola de trabajos de proyección.

La proyección tarda minutos y consume núcleos, así que no puede correr dentro
del request. Este módulo la ejecuta en segundo plano con tres límites:

  * un solo trabajo pesado a la vez (semáforo), para que varios usuarios
    disparando proyecciones no se peleen por la CPU;
  * un tope de núcleos por trabajo, dejando margen a las otras apps del
    servidor;
  * caché por combinación exacta de parámetros, de modo que repetir la misma
    consulta devuelva el resultado ya calculado en lugar de recalcularlo.
"""

from __future__ import annotations

import hashlib
import json
import threading
import time
import traceback
import uuid
from collections import OrderedDict
from typing import Any, Callable

MAX_CONCURRENTES = 1
CACHE_MAX = 32
TTL_JOB = 3600  # segundos que se conserva un trabajo terminado

_sem = threading.Semaphore(MAX_CONCURRENTES)
_lock = threading.Lock()
_jobs: dict[str, dict] = {}
_cache: OrderedDict[str, Any] = OrderedDict()


def clave(params: dict) -> str:
    crudo = json.dumps(params, sort_keys=True, ensure_ascii=False, default=str)
    return hashlib.sha256(crudo.encode()).hexdigest()[:16]


def _limpiar():
    """Descarta trabajos terminados hace rato para no acumularlos en memoria."""
    ahora = time.time()
    for jid in [
        j for j, d in _jobs.items()
        if d["estado"] in ("listo", "error") and ahora - d.get("fin", ahora) > TTL_JOB
    ]:
        _jobs.pop(jid, None)


def lanzar(params: dict, trabajo: Callable[[Callable[[dict], None]], Any]) -> dict:
    """
    Encola un trabajo. Si el resultado ya está en caché lo devuelve al instante;
    si ya hay uno idéntico corriendo, se engancha a ese en vez de duplicarlo.
    """
    k = clave(params)

    with _lock:
        _limpiar()

        if k in _cache:
            _cache.move_to_end(k)
            jid = str(uuid.uuid4())[:8]
            _jobs[jid] = {
                "id": jid, "clave": k, "estado": "listo", "desde_cache": True,
                "progreso": {"fase": "listo", "mensaje": "Resultado en caché"},
                "resultado": _cache[k], "inicio": time.time(), "fin": time.time(),
                "params": params,
            }
            return _jobs[jid]

        for d in _jobs.values():
            if d["clave"] == k and d["estado"] in ("en_cola", "corriendo"):
                return d

        jid = str(uuid.uuid4())[:8]
        job = {
            "id": jid, "clave": k, "estado": "en_cola", "desde_cache": False,
            "progreso": {"fase": "en_cola", "mensaje": "Esperando un núcleo libre"},
            "resultado": None, "error": None,
            "inicio": time.time(), "fin": None, "params": params,
        }
        _jobs[jid] = job

    def correr():
        # El semáforo se toma dentro del hilo: la petición HTTP ya respondió y
        # el trabajo espera aquí su turno sin bloquear a nadie.
        with _sem:
            with _lock:
                job["estado"] = "corriendo"
                job["progreso"] = {"fase": "iniciando", "mensaje": "Preparando la serie"}
            try:
                def avance(p: dict):
                    with _lock:
                        job["progreso"] = {**job.get("progreso", {}), **p}

                res = trabajo(avance)

                with _lock:
                    job["resultado"] = res
                    job["estado"] = "listo"
                    job["fin"] = time.time()
                    job["progreso"] = {"fase": "listo", "mensaje": "Proyección completa"}
                    _cache[k] = res
                    while len(_cache) > CACHE_MAX:
                        _cache.popitem(last=False)
            except Exception as e:
                with _lock:
                    job["estado"] = "error"
                    job["error"] = str(e)
                    job["traza"] = traceback.format_exc()[-1500:]
                    job["fin"] = time.time()
                    job["progreso"] = {"fase": "error", "mensaje": str(e)}

    threading.Thread(target=correr, daemon=True).start()
    return job


def desde_repositorio(params: dict, resultado: Any) -> dict:
    """
    Envuelve una proyección recuperada del repositorio en disco con la misma
    forma que un trabajo recién corrido, para que el frontend no distinga.
    """
    jid = str(uuid.uuid4())[:8]
    ahora = time.time()
    with _lock:
        _jobs[jid] = {
            "id": jid, "clave": clave(params), "estado": "listo", "desde_cache": True,
            "origen": "repositorio",
            "progreso": {
                "fase": "listo",
                "mensaje": f"Recuperada del repositorio (calculada el {resultado.get('_guardada_en')})",
            },
            "resultado": resultado, "inicio": ahora, "fin": ahora, "params": params,
        }
    return _jobs[jid]


def estado(jid: str) -> dict | None:
    with _lock:
        job = _jobs.get(jid)
        if not job:
            return None
        salida = {k: v for k, v in job.items() if k != "traza"}
        salida["duracion"] = round((job.get("fin") or time.time()) - job["inicio"], 1)
        # Cuántos trabajos hay delante en la cola.
        salida["en_espera"] = sum(
            1 for d in _jobs.values()
            if d["estado"] == "en_cola" and d["inicio"] < job["inicio"]
        )
        return salida