"""Background jobs on Procrastinate (Postgres-backed queue; no Redis).

One periodic task is generated per registered adapter/enricher from its `schedule` (cron, UTC) and
`queue`. Run workers per queue so heavy jobs don't block light ones:

    procrastinate --app=politrack.worker.app worker --queues=scrape,social
    procrastinate --app=politrack.worker.app worker --queues=bulk
    procrastinate --app=politrack.worker.app worker --queues=enrich

Periodic tasks are deferred by every running worker and de-duplicated by Procrastinate; queueing locks
prevent two runs of the same adapter at once. Workers can run on other machines that reach the DB.

Procrastinate only remembers in memory when it last deferred a task, and skips a slot that is more than
ten minutes overdue. In November 2026 every daily and weekly source silently missed two days that way,
and jobs of killed workers stayed "doing" forever. The watchdog below makes both self-healing.
"""

from __future__ import annotations

import asyncio
import logging
import os
import socket
from datetime import datetime, timedelta, timezone
from logging.handlers import RotatingFileHandler
from pathlib import Path

from croniter import croniter
from procrastinate import App, PsycopgConnector
from procrastinate.exceptions import AlreadyEnqueued
from procrastinate.jobs import Status

from politrack.config import get_settings
from politrack.core import registry
from politrack.core.runner import run
from politrack.db import connection

LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
log = logging.getLogger("politrack.watchdog")


def _log_to_file() -> None:
    """Container logs vanish when a container is recreated (every deploy); keep a rotating copy in
    $POLITRACK_LOG_DIR/<hostname>.log so failures can still be traced afterwards."""
    log_dir = os.environ.get("POLITRACK_LOG_DIR")
    if not log_dir:
        return
    try:
        handler = RotatingFileHandler(Path(log_dir) / f"{socket.gethostname()}.log", maxBytes=20_000_000,
                                      backupCount=5, encoding="utf-8")
    except OSError as exc:  # e.g. the host folder belongs to root: logs stay on stdout only
        log.warning("cannot write logs to %s: %s", log_dir, exc)
        return
    handler.setFormatter(logging.Formatter(LOG_FORMAT))
    logging.getLogger().addHandler(handler)


_log_to_file()

app = App(connector=PsycopgConnector(conninfo=get_settings().database_url))
TASKS: dict = {}
SCHEDULES: dict[str, str] = {}

STALLED_AFTER = 300                       # seconds without a worker heartbeat
OVERDUE_GRACE = timedelta(minutes=15)     # normal deferral happens within a minute of the slot
ABANDONED_RUN = timedelta(hours=6)        # CLI runs have no job; don't close them while they may still run


def _make_task(name: str, queue: str, schedule: str | None) -> None:
    def job(timestamp: int | None = None, **params) -> dict:
        return run(name, params=params or None)

    job.__name__ = f"run_{name.replace('.', '_')}"
    task = app.task(name=f"politrack.run.{name}", queue=queue, queueing_lock=name, pass_context=False)(job)
    TASKS[name] = task
    if schedule:
        SCHEDULES[name] = schedule
        app.periodic(cron=schedule, periodic_id=name)(task)


registry.load_builtin()
for _name, _cls in {**registry.ADAPTERS, **registry.ENRICHERS}.items():
    _make_task(_name, _cls.queue, _cls.schedule)


def overdue_tasks(now: datetime) -> list[str]:
    """Scheduled tasks whose latest slot passed without a run, a waiting job or a running job."""
    due: list[str] = []
    with connection() as conn:
        for name, schedule in SCHEDULES.items():
            slot = croniter(schedule, now).get_prev(datetime)
            if now - slot < OVERDUE_GRACE:
                continue
            seen = conn.execute(
                """SELECT EXISTS (SELECT 1 FROM run WHERE adapter = %(name)s AND started_at >= %(slot)s)
                       OR EXISTS (SELECT 1 FROM procrastinate_jobs WHERE task_name = %(task)s
                                  AND status IN ('todo', 'doing')) AS seen""",
                {"name": name, "slot": slot, "task": f"politrack.run.{name}"},
            ).fetchone()["seen"]
            if not seen:
                due.append(name)
    return due


def close_abandoned_runs() -> int:
    """Runs still marked running although no job for them is in progress (the worker was stopped)."""
    with connection() as conn:
        closed = conn.execute(
            """UPDATE run r SET status = 'error', finished_at = now(),
                   error = 'abandoned: the worker stopped before the run finished (closed by the watchdog)'
               WHERE r.status = 'running' AND r.started_at < now() - %s
                 AND NOT EXISTS (SELECT 1 FROM procrastinate_jobs j
                                 WHERE j.status = 'doing' AND j.task_name = 'politrack.run.' || r.adapter)""",
            (ABANDONED_RUN,),
        ).rowcount
        conn.commit()
    return closed


@app.periodic(cron="*/10 * * * *", periodic_id="watchdog")
@app.task(name="politrack.watchdog", queue="social", queueing_lock="watchdog", pass_context=False)
async def watchdog(timestamp: int | None = None) -> dict:
    stats = {"stalled_failed": 0, "overdue_deferred": 0}
    for job in await app.job_manager.get_stalled_jobs(seconds_since_heartbeat=STALLED_AFTER):
        await app.job_manager.finish_job(job, status=Status.FAILED, delete_job=False)
        stats["stalled_failed"] += 1
        log.warning("failed stalled job %s (%s): its worker is gone", job.id, job.task_name)
    stats["abandoned_runs_closed"] = await asyncio.to_thread(close_abandoned_runs)
    for name in await asyncio.to_thread(overdue_tasks, datetime.now(timezone.utc)):
        try:
            await TASKS[name].defer_async()
            stats["overdue_deferred"] += 1
            log.warning("deferred %s: its last scheduled slot passed without a run", name)
        except AlreadyEnqueued:
            pass
    return stats