"""Scheduler process (`aia schedule`): runs due connectors (adaptive intervals, Redis lock per connector), the job worker,
hourly stats/quality, embedding backlog, nightly backup. One process per node is enough; several nodes cooperate via locks."""
from __future__ import annotations

import asyncio
import logging
import signal
from datetime import UTC, datetime

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

from aiatlas.config import settings
from aiatlas.connectors import get, registry
from aiatlas.db import execute, fetch_all, fetch_one, transaction
from aiatlas.services import cache
from aiatlas.services.jobs import enqueue, requeue_stale, run_worker

log = logging.getLogger(__name__)

_running: set[str] = set()


async def run_connector(name: str, *, force: bool = False) -> None:
    if name in _running:
        return
    _running.add(name)
    try:
        async with cache.lock(f"connector:{name}", ttl_s=4 * 3600) as ok:
            if not ok:
                log.info("connector locked elsewhere", extra={"connector": name})
                return
            try:
                await get(name).run(force=force)
            except Exception as exc:  # noqa: BLE001
                log.warning("connector run failed", extra={"connector": name, "error": str(exc)})
            # a connector run can touch every read model (stats, changes, benchmarks, prices, models, frontier, pulse, families…):
            # flush the whole `aia:api:` namespace so new routes are covered without maintaining a prefix list
            await cache.cache_invalidate()
    finally:
        _running.discard(name)


LIGHT_CANONICALIZE_STEPS = ["variants", "artifacts", "families", "events", "anomalies"]


async def canonicalize_job(*, light: bool = False) -> None:
    """Nightly full `canonicalize --apply` (03:30 local) and a light pass every 6 h; one line of summary in the log."""
    from aiatlas.services.canonical import canonicalize

    async with cache.lock("canonicalize", ttl_s=2 * 3600) as ok:
        if not ok:
            log.info("canonicalize locked elsewhere, skipping", extra={"light": light})
            return
        try:
            report = await canonicalize(apply=True, steps=LIGHT_CANONICALIZE_STEPS if light else None)
            summary = ", ".join(f"{s.name}={s.changes}" for s in report.steps if s.changes) or "no changes"
            log.info("canonicalize %s pass: %d change(s) — %s", "light" if light else "full", report.changes, summary,
                     extra={"light": light, "changes": report.changes, "steps": {s.name: s.changes for s in report.steps}})
            if report.changes:
                await cache.cache_invalidate()
        except Exception as exc:  # noqa: BLE001
            log.warning("canonicalize pass failed", extra={"light": light, "error": str(exc)})


async def _pop_run_now() -> list[str]:
    """Admin `POST /admin/connectors/{name}/run` sets `aia:run-now:<name>`; consume (delete) every such key."""
    names: list[str] = []
    try:
        r = cache.redis()
        async for key in r.scan_iter(match="aia:run-now:*", count=100):
            if await r.delete(key):
                names.append(key.decode().rsplit(":", 1)[-1])
    except Exception as exc:  # noqa: BLE001
        log.debug("run-now scan skipped", extra={"error": str(exc)})
    return names


async def tick() -> None:
    known = registry()
    forced = [n for n in await _pop_run_now() if n in known and n not in _running]
    if forced:
        log.info("run-now connectors", extra={"connectors": forced})
        for n in forced[:3]:                       # one at a time: concurrent runs touch the same organisation rows and can deadlock Postgres
            await run_connector(n, force=True)
    async with transaction() as conn:
        due = await fetch_all(conn, """select name from connectors where enabled and (next_run_at is null or next_run_at <= now())
                                       and (circuit_open_until is null or circuit_open_until <= now()) order by priority, coalesce(next_run_at, 'epoch') limit 6""")
    names = [r["name"] for r in due if r["name"] in known and r["name"] not in _running and r["name"] not in forced]
    if names:
        log.info("due connectors", extra={"connectors": names})
        for n in names[:2]:
            await run_connector(n)
    await cache.heartbeat("scheduler", {"at": datetime.now(UTC).isoformat(), "running": sorted(_running), "due": names, "forced": forced})


async def hourly() -> None:
    from aiatlas.services.quality import recompute
    from aiatlas.services.stats import compute_stats

    async with cache.lock("hourly", ttl_s=3000) as ok:
        if not ok:
            return
        try:
            await compute_stats()
            await recompute(limit=5000)
            async with transaction() as conn:
                n = await requeue_stale(conn)
                if n:
                    log.info("requeued stale jobs", extra={"n": n})
            async with transaction() as conn:
                extraction_pending = await fetch_one(conn, "select 1 from jobs where kind = 'llm_extract' and status in ('queued','running') limit 1")
            if settings.llm_available and not extraction_pending:   # never interleave embeddings with extraction: the server would swap models
                from aiatlas.services.embeddings import pending_entity_ids

                ids = await pending_entity_ids(limit=400)
                if ids:
                    async with transaction() as conn:
                        for i in range(0, len(ids), 50):
                            await enqueue(conn, "embed_entity", {"entity_ids": ids[i:i + 50]}, priority=8, dedupe_key=f"embed:{ids[i]}")
            await cache.cache_invalidate()
        except Exception as exc:  # noqa: BLE001
            log.warning("hourly maintenance failed", extra={"error": str(exc)})


async def nightly_backup() -> None:
    from aiatlas.services.backup import backup_database

    async with cache.lock("backup", ttl_s=3600) as ok:
        if ok:
            try:
                path = await asyncio.to_thread(backup_database)
                log.info("backup done", extra={"path": str(path)})
            except Exception as exc:  # noqa: BLE001
                log.error("backup failed", extra={"error": str(exc)})


async def main(*, with_worker: bool = True) -> None:
    settings.ensure_dirs()
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, stop.set)
        except NotImplementedError:
            pass
    scheduler = AsyncIOScheduler(timezone="UTC")
    scheduler.add_job(tick, "interval", seconds=settings.scheduler_tick_s, max_instances=1, coalesce=True, id="tick")
    scheduler.add_job(hourly, "interval", minutes=60, max_instances=1, coalesce=True, id="hourly", next_run_time=datetime.now(UTC))
    minute, hour, *_ = settings.backup_cron.split()
    scheduler.add_job(nightly_backup, CronTrigger(minute=minute, hour=hour, timezone=settings.tz), id="backup")
    scheduler.add_job(canonicalize_job, CronTrigger(minute=30, hour=3, timezone=settings.tz), id="canonicalize-full", max_instances=1, coalesce=True)
    scheduler.add_job(canonicalize_job, "interval", hours=6, kwargs={"light": True}, id="canonicalize-light", max_instances=1, coalesce=True)
    scheduler.start()
    log.info("scheduler started", extra={"tick_s": settings.scheduler_tick_s, "connectors": len(registry()), "worker": with_worker})
    async with transaction() as conn:
        await execute(conn, "update connectors set health = 'disabled' where not enabled")
    worker_task = asyncio.create_task(run_worker(stop=stop)) if with_worker else None
    await stop.wait()
    scheduler.shutdown(wait=False)
    if worker_task:
        await worker_task
    await cache.close()


__all__ = ["LIGHT_CANONICALIZE_STEPS", "canonicalize_job", "hourly", "main", "run_connector", "tick"]