"""`catlas schedule` — the long-running crawl process (spec §15–16, §62–63, §124–128).

Every `settings.scheduler_tick_s` the scheduler claims due sensors from Postgres (`for update skip locked`, so any number of
scheduler processes on any machines can share one database), runs them through `pipeline.run_sensor` under a global semaphore
(the per-domain governor lives in `fetch.Fetcher`), releases claims, writes a `crawl_runs` row and a heartbeat in `settings_kv`.
It also hosts the onboarding worker (`discovery.onboard_pending`) and every periodic task registered through
`services.periodic` (interval + cron via APScheduler in `settings.tz`). Claims older than `CLAIM_TTL` are considered abandoned.
"""
from __future__ import annotations

import asyncio
import logging
import os
import signal
import socket
import time
from datetime import UTC, datetime
from typing import Any

from companyatlas.config import settings
from companyatlas.db import execute, fetch_all, jsonb, transaction
from companyatlas.fetch import Fetcher
from companyatlas.ids import new_id
from companyatlas.services import periodic
from companyatlas.services.pipeline import RunOutcome, run_sensor

log = logging.getLogger(__name__)

CLAIM_TTL = "15 minutes"
IDLE_RUN_ROW_EVERY = 40            # write a crawl_runs row on idle ticks only every N ticks
ONBOARD_BATCH = 20
ONBOARD_EVERY_TICKS = 4


def worker_name() -> str:
    return f"{socket.gethostname().split('.')[0]}:{os.getpid()}"


async def claim_due_sensors(worker: str, limit: int) -> list[dict[str, Any]]:
    async with transaction() as conn:
        return await fetch_all(conn, f"""
            with due as (
                select id from sensors
                where status in ('active', 'failing', 'pending') and next_run_at <= now()
                  and (claimed_at is null or claimed_at < now() - interval '{CLAIM_TTL}')
                order by priority desc, next_run_at
                limit :limit for update skip locked)
            update sensors s set claimed_by = :worker, claimed_at = now() from due where s.id = due.id returning s.*
        """, limit=limit, worker=worker)


async def release_claims(worker: str) -> None:
    async with transaction() as conn:
        await execute(conn, "update sensors set claimed_by = null, claimed_at = null where claimed_by = :w", w=worker)


async def heartbeat(worker: str, payload: dict[str, Any]) -> None:
    async with transaction() as conn:
        for key in ("scheduler:heartbeat", f"scheduler:heartbeat:{worker}"):
            await execute(conn, """insert into settings_kv (key, value) values (:k, cast(:v as jsonb))
                                   on conflict (key) do update set value = excluded.value, updated_at = now()""", k=key, v=jsonb(payload))


async def record_run(kind: str, worker: str, started_at: datetime, stats: dict[str, Any], error: str | None = None) -> None:
    async with transaction() as conn:
        await execute(conn, """insert into crawl_runs (id, kind, worker, started_at, finished_at, stats, error)
                               values (:id, :kind, :worker, :started, now(), cast(:stats as jsonb), :error)""",
                      id=new_id("crawl_run"), kind=kind, worker=worker, started=started_at, stats=jsonb(stats), error=error)


class Scheduler:
    def __init__(self, *, concurrency: int | None = None, onboarding: bool = True, worker: str | None = None, claim_batch: int | None = None):
        self.concurrency = max(1, concurrency or settings.fetch_concurrency)
        self.onboarding = onboarding
        self.worker = worker or worker_name()
        self.claim_batch = claim_batch or settings.scheduler_claim_batch
        self.stop = asyncio.Event()
        self.inflight = 0
        self.ticks = 0
        self.fetcher = Fetcher()
        self._onboard_task: asyncio.Task[Any] | None = None
        self._aps: Any = None

    # ------------------------------------------------------------------ one tick
    async def tick(self) -> dict[str, Any]:
        started = datetime.now(UTC)
        t0 = time.perf_counter()
        sensors = await claim_due_sensors(self.worker, self.claim_batch)
        stats = {"claimed": len(sensors), "ok": 0, "changed": 0, "meaningful": 0, "failed": 0, "not_modified": 0, "unchanged": 0, "skipped": 0, "redirected": 0}
        sem = asyncio.Semaphore(self.concurrency)

        async def one(row: dict[str, Any]) -> None:
            async with sem:
                self.inflight += 1
                try:
                    outcome: RunOutcome = await run_sensor(row, fetcher=self.fetcher, worker=self.worker)
                except Exception as exc:
                    log.exception("sensor run crashed", extra={"sensor_id": row.get("id")})
                    stats["failed"] += 1
                    async with transaction() as conn:
                        await execute(conn, """update sensors set claimed_by = null, claimed_at = null, last_error = :e,
                                               next_run_at = now() + interval '30 minutes' where id = :id""", id=row["id"], e=f"{exc.__class__.__name__}: {exc}"[:500])
                    return
                finally:
                    self.inflight -= 1
                if outcome.status in ("ok", "changed", "unchanged", "not_modified"):
                    stats["ok"] += 1
                if outcome.status == "changed":
                    stats["changed"] += 1
                    if outcome.kind in ("meaningful", "major", "critical"):
                        stats["meaningful"] += 1
                elif outcome.status in stats:
                    stats[outcome.status] += 1

        if sensors:
            await asyncio.gather(*(one(s) for s in sensors))
        stats["tick_ms"] = int((time.perf_counter() - t0) * 1000)
        self.ticks += 1
        if sensors or self.ticks % IDLE_RUN_ROW_EVERY == 0:
            await record_run("scheduler_tick", self.worker, started, stats)
        due = await self._due_count()
        await heartbeat(self.worker, {"worker": self.worker, "at": datetime.now(UTC).isoformat(), "inflight": self.inflight, "due": due, "tick_ms": stats["tick_ms"],
                                      "claimed": stats["claimed"], "concurrency": self.concurrency, "tasks": periodic.snapshot()})
        log.info("tick", extra={"worker": self.worker, **stats, "due": due})
        return stats

    async def _due_count(self) -> int:
        async with transaction() as conn:
            rows = await fetch_all(conn, "select count(*) as n from sensors where status in ('active','failing','pending') and next_run_at <= now()")
        return int(rows[0]["n"]) if rows else 0

    # ------------------------------------------------------------------ onboarding
    async def _maybe_onboard(self) -> None:
        if not self.onboarding or (self._onboard_task and not self._onboard_task.done()):
            return
        from companyatlas.services.discovery import onboard_pending

        async def go() -> None:
            try:
                stats = await onboard_pending(ONBOARD_BATCH, settings.onboarding_concurrency, fetcher=self.fetcher, worker=self.worker)
                if stats.get("claimed"):
                    await record_run("onboarding", self.worker, datetime.now(UTC), stats)
            except Exception:
                log.exception("onboarding batch failed")

        self._onboard_task = asyncio.create_task(go())

    # ------------------------------------------------------------------ periodic tasks
    def _start_periodic(self) -> None:
        loaded = periodic.load_task_modules()
        try:
            from apscheduler.schedulers.asyncio import AsyncIOScheduler
            from apscheduler.triggers.cron import CronTrigger
            from apscheduler.triggers.interval import IntervalTrigger
        except ImportError:  # pragma: no cover
            log.warning("apscheduler unavailable — periodic tasks disabled")
            return
        aps = AsyncIOScheduler(timezone=settings.tz)
        for task in periodic.tasks().values():
            if task.cron:
                trigger: Any = CronTrigger.from_crontab(task.cron, timezone=settings.tz)
            else:
                trigger = IntervalTrigger(seconds=float(task.every_s or 60), start_date=datetime.now(UTC).timestamp() + task.initial_delay_s and None)
            aps.add_job(task.run, trigger=trigger, id=task.name, name=task.name, max_instances=1, coalesce=True, misfire_grace_time=60)
        aps.start()
        self._aps = aps
        log.info("periodic tasks started", extra={"modules": loaded, "tasks": list(periodic.tasks())})

    # ------------------------------------------------------------------ main loop
    async def run(self, *, once: bool = False) -> None:
        await self.fetcher.open()
        loop = asyncio.get_running_loop()
        for sig in (signal.SIGINT, signal.SIGTERM):
            try:
                loop.add_signal_handler(sig, self.stop.set)
            except (NotImplementedError, RuntimeError):  # pragma: no cover - non-unix
                pass
        try:
            from companyatlas.sdk.connector import sync_connectors_table

            async with transaction() as conn:
                await sync_connectors_table(conn)
            if not once:
                self._start_periodic()
            log.info("scheduler started", extra={"worker": self.worker, "concurrency": self.concurrency, "tick_s": settings.scheduler_tick_s, "onboarding": self.onboarding})
            while not self.stop.is_set():
                try:
                    if self.onboarding and (self.ticks % ONBOARD_EVERY_TICKS == 0):
                        await self._maybe_onboard()
                    await self.tick()
                except Exception:
                    log.exception("tick failed")
                if once:
                    if self._onboard_task:
                        await self._onboard_task
                    break
                try:
                    await asyncio.wait_for(self.stop.wait(), settings.scheduler_tick_s)
                except TimeoutError:
                    pass
        finally:
            if self._aps is not None:
                self._aps.shutdown(wait=False)
            if self._onboard_task and not self._onboard_task.done():
                self._onboard_task.cancel()
            await release_claims(self.worker)
            await self.fetcher.close()
            log.info("scheduler stopped", extra={"worker": self.worker, "ticks": self.ticks})


async def run_scheduler(*, concurrency: int | None = None, onboarding: bool = True, once: bool = False, worker: str | None = None) -> None:
    await Scheduler(concurrency=concurrency, onboarding=onboarding, worker=worker).run(once=once)


__all__ = ["Scheduler", "claim_due_sensors", "heartbeat", "record_run", "release_claims", "run_scheduler", "worker_name"]