"""The control hook: what a running collector reads between requests, and what a page action
must satisfy before it starts one.

Sources of truth: this module, `services/ingest.run_collector` (binds it), `collectors/fetch.py`
(calls `wait()` where it slept), `routers/collectors.py` (the six routes), the brief
`.logs/planning/streams/AW4-live-page.md` D2, D3, D7.

The skip-sink pattern, one ContextVar: unbound, `wait()` sleeps as the fetch port always did, so
collectors stay usable from a shell and in tests; bound by `run_collector`, every request
boundary commits the open batch, heartbeats the run row, re-reads the source's control, kill
switch and pace, and sleeps the pace in `CONTROL_POLL` slices, re-reading between slices. So a
pause, a stop, a pace change and the kill switch all land within two seconds plus the request in
flight, and a paused run is never silent: it heartbeats every poll.

`RunStopped` is a `BaseException`, like `KeyboardInterrupt`: a collector's broad
`except Exception` (the Dubai category-name enrichment has one) cannot swallow a stop. The SIGTERM
handler `run_collector` installs only sets a flag; the raise happens here at the next poll, never
mid-flush. On 19 Sep the guard's shed left four `running` rows behind and the approval gate held
the review for the 24 hours the timer had left; a run that ends through this hook ends `stopped`.

`effective_control` is why the loop never writes `sources.control`: a stop pressed yesterday
must not kill tonight's shell or admit-queue run, so the field applies only when it was set after
the run began, and nothing ever resets it.
"""

from __future__ import annotations

import inspect
import logging
import subprocess
import sys
import threading
import time
from collections.abc import Callable
from contextvars import ContextVar
from dataclasses import dataclass, field
from datetime import UTC, datetime
from functools import lru_cache
from typing import Any
from urllib.parse import urlsplit

from sqlalchemy import select, update
from sqlalchemy.orm import Session

from app.models import CollectionRun, Source

logger = logging.getLogger(__name__)

#: Below one second between requests is a request storm, whatever a person types.
OUR_FLOOR = 1.0
#: Slower than any crawl delay a host has asked of us (the slowest met is 60 s).
PACE_CAP = 600.0
#: The control is re-read this often while sleeping or paused: a page action lands in two
#: seconds plus the request in flight, and a paused run heartbeats this often.
CONTROL_POLL = 2.0
#: After a stop is requested, the page offers Stop now (SIGTERM) once this has passed: the
#: longest request in flight is a 40 s fetch or a 90 s render.
STOP_GRACE = 60.0

MODES = ("discover", "recheck")
CONTROLS = ("run", "pause", "stop")


class RunStopped(BaseException):
    """The run must end now, cooperatively. `by` is what the run row's `stopped_by` records."""

    def __init__(self, reason: str, by: str) -> None:
        super().__init__(reason)
        self.reason = reason
        self.by = by


class CollectorLocked(RuntimeError):
    """Another process holds this source's collector lock; no run row was made."""


def effective_control(control: str | None, set_at: datetime | None, run_started_at: datetime | None) -> str:
    """The control that applies to a run: what was set, only if it was set after the run began."""
    if control not in ("pause", "stop"):
        return "run"
    if set_at is None or run_started_at is None:
        return "run"
    return control if _aware(set_at) >= _aware(run_started_at) else "run"


def _aware(value: datetime) -> datetime:
    return value if value.tzinfo else value.replace(tzinfo=UTC)


@dataclass(frozen=True, slots=True)
class ControlRow:
    control: str
    control_set_by: str | None
    control_set_at: datetime | None
    delay_seconds: float
    enabled: bool


@dataclass
class RunControl:
    """One run's hook. `db` is the ingest session itself, never a second one: the boundary
    commit is the batch commit. `sleeper` and `clock` are injectable so the tests run in
    milliseconds with a fake clock."""

    run_id: int
    source_id: int
    slug: str
    db: Session
    run_started_at: datetime
    sleeper: Callable[[float], None] = time.sleep
    clock: Callable[[], float] = time.monotonic
    stop_flag: threading.Event = field(default_factory=threading.Event)
    #: The host's Crawl-delay as read THIS run, and whether it still has to be written.
    robots_delay: float | None = None
    robots_pending: tuple[float | None, datetime] | None = None
    polls: int = 0
    boundaries: int = 0

    # -- what the hook writes and reads ------------------------------------------------------

    def note_robots(self, crawl_delay: float | None) -> None:
        self.robots_delay = crawl_delay
        self.robots_pending = (crawl_delay, datetime.now(UTC))

    def _poll(self, boundary: bool) -> ControlRow:
        now = datetime.now(UTC)
        values: dict[str, Any] = {"heartbeat_at": now}
        if boundary:
            values["requests_made"] = CollectionRun.requests_made + 1
        result = self.db.execute(
            update(CollectionRun)
            .where(CollectionRun.id == self.run_id, CollectionRun.status == "running")
            .values(**values)
            .execution_options(synchronize_session=False)
        )
        if result.rowcount == 0:
            # Someone else closed this row (a newer run of the source, `backfill stuck_runs`,
            # Mark as ended on the page): this process no longer owns it.
            self.db.rollback()
            raise RunStopped("superseded", "rules")
        if self.robots_pending is not None:
            delay, read_at = self.robots_pending
            self.db.execute(
                update(Source).where(Source.id == self.source_id)
                .values(robots_crawl_delay=delay, robots_read_at=read_at)
                .execution_options(synchronize_session=False)
            )
            self.robots_pending = None
        # A column SELECT, never the identity-mapped Source: under expire_on_commit=False the
        # object the loop holds is the row as it was when the run began.
        row = self.db.execute(
            select(Source.control, Source.control_set_by, Source.control_set_at,
                   Source.delay_seconds, Source.enabled).where(Source.id == self.source_id)
        ).one()
        self.db.commit()
        self.polls += 1
        if boundary:
            self.boundaries += 1
        return ControlRow(row[0], row[1], row[2], float(row[3] or 0.0), bool(row[4]))

    def _decide(self, row: ControlRow) -> str:
        """`run` or `pause`; raises RunStopped for the kill switch, a signal, or a stop."""
        if not row.enabled:
            raise RunStopped("kill switch", "kill-switch")
        if self.stop_flag.is_set():
            raise RunStopped("signal", "signal")
        effective = effective_control(row.control, row.control_set_at, self.run_started_at)
        if effective == "stop":
            raise RunStopped("stopped from the page", row.control_set_by or "page")
        return "pause" if effective == "pause" else "run"

    def _settle(self, row: ControlRow) -> ControlRow:
        """The pause loop: heartbeat every CONTROL_POLL until the control says run again."""
        while self._decide(row) == "pause":
            self.sleeper(CONTROL_POLL)
            row = self._poll(boundary=False)
        return row

    def _wait_for(self, requested: float, row: ControlRow) -> float:
        return max(float(requested), float(self.robots_delay or 0.0), row.delay_seconds)

    # -- the boundary ---------------------------------------------------------------------------

    def wait(self, requested: float) -> None:
        """Called where the fetch port slept. Commits the batch, heartbeats, reads the control,
        holds while paused, then sleeps the pace in slices, re-reading between them."""
        if self.db.in_nested_transaction():
            raise AssertionError("the control hook was reached inside a savepoint")
        self.db.commit()
        row = self._settle(self._poll(boundary=True))
        if requested == 0:
            return  # the deliberate first fetch; the control was still read once
        t0 = self.clock()
        deadline = t0 + self._wait_for(requested, row)
        while True:
            remaining = deadline - self.clock()
            if remaining <= 0:
                return
            self.sleeper(min(remaining, CONTROL_POLL))
            row = self._settle(self._poll(boundary=False))
            # A pause held the deadline (two requests are never closer than the pace); a pace
            # change lengthens this wait and never shortens it.
            deadline = t0 + max(deadline - t0, self._wait_for(requested, row))


_control: ContextVar[RunControl | None] = ContextVar("collector_run_control", default=None)


def bind_control(control: RunControl | None):
    return _control.set(control)


def unbind_control(token) -> None:
    _control.reset(token)


def current() -> RunControl | None:
    return _control.get()


def wait(requested: float) -> None:
    """The one call the fetch port makes where it used to sleep."""
    control = _control.get()
    if control is None:
        if requested and requested > 0:
            time.sleep(requested)
        return
    control.wait(requested)


def note_robots(robots: Any) -> None:
    """`check_allowed` hands the run the host's rules; the next poll publishes the crawl delay."""
    control = _control.get()
    if control is not None:
        control.note_robots(getattr(robots, "crawl_delay", None))


# --------------------------------------------------------------------------- what a Start needs

@dataclass(frozen=True, slots=True)
class Refusal:
    code: str
    summary: str


def is_rendered(collector: Any) -> bool:
    """A collector whose module draws pages in the browser sidecar (`fetch.render`)."""
    module = inspect.getmodule(type(collector))
    if module is None:
        return False
    from app.services.collectors import fetch

    return getattr(module, "render", None) is fetch.render


def has_read_path(collector: Any) -> bool:
    """Whether `read_one` reads anything. A family with no permitted read path refuses every
    read (its body is one `raise SourceBlocked`), and a recheck of it would write a `blocked`
    run and make the source refused for good. Read from the method's syntax tree, offline: a
    docstring that mentions a fetch is not a fetch."""
    import ast
    import textwrap

    try:
        tree = ast.parse(textwrap.dedent(inspect.getsource(type(collector).read_one)))
    except (OSError, TypeError, AttributeError, SyntaxError):
        return False
    function = next((n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "read_one"), None)
    if function is None:
        return False
    body = [n for n in function.body if not (isinstance(n, ast.Expr) and isinstance(getattr(n, "value", None), ast.Constant))]
    return not all(isinstance(n, ast.Raise) for n in body)


def recheck_refusal(collector: Any) -> Refusal | None:
    """Why a recheck is not possible on this collector, or None; a fact of the class, cached."""
    return _recheck_refusal_of(type(collector))


@lru_cache(maxsize=64)
def _recheck_refusal_of(cls: type) -> Refusal | None:
    collector = cls.__new__(cls)
    if is_rendered(collector):
        from app.services.collectors.fetch import RENDER_DEFAULT_CAP

        return Refusal("MODE_UNSUPPORTED",
                       f"recheck is not supported on a rendered collector: the render cap ends the walk at "
                       f"{RENDER_DEFAULT_CAP} pages.")
    if not has_read_path(collector):
        return Refusal("MODE_UNSUPPORTED",
                       "recheck is not supported here: this platform has no permitted way to read one listing back.")
    return None


def host_of(collector: Any, source: Source) -> str:
    homepage = getattr(collector, "homepage", None) or ""
    return urlsplit(homepage).hostname or source.slug


def held_listings_count(db: Session, collector: Any) -> int:
    """How many listings at this collector's shops have a URL to read back."""
    from sqlalchemy import func

    from app.models import Listing, Retailer, Shop

    codes = [spec.code for spec in collector.shops()]
    if not codes:
        return 0
    return db.scalar(
        select(func.count(Listing.id))
        .join(Shop, Shop.id == Listing.shop_id)
        .join(Retailer, Retailer.id == Shop.retailer_id)
        .where(Retailer.slug == collector.retailer_slug, Shop.code.in_(codes), Listing.url.isnot(None))
    ) or 0


def refused_since_of(db: Session, source_id: int) -> datetime | None:
    """The newest BLOCKED verification check on the source, when the table exists."""
    from sqlalchemy import func

    from app.models.quality import VerificationCheck
    from app.services.collector_view import _has_checks

    if not _has_checks(db):
        return None
    return db.scalar(
        select(func.max(VerificationCheck.checked_at))
        .where(VerificationCheck.verdict == "BLOCKED", VerificationCheck.source_id == source_id)
    )


def _when(value: datetime | None) -> str:
    return _aware(value).strftime("%d %b %H:%M") if value else "an unknown time"


def start_refusal(
    db: Session, source: Source, collector: Any, mode: str, limit: int | None, memory: Any,
    now: datetime, *, last: CollectionRun | None = None, running: CollectionRun | None = None,
    refused_since: datetime | None = None, prefetched: bool = False, held_listings: int | None = None,
) -> Refusal | None:
    """Why a Start would be refused, in the order the route refuses: the legal reasons first
    (`SOURCE_DISABLED`, `SOURCE_REFUSED`, final), then the operational ones. The route and the
    live read call this one function, so `start_blocked` on the page is exactly what the POST
    answers. `prefetched=True` says the caller supplied `last`, `running` and `refused_since`;
    `held_listings` is the count already in hand (the live read's one statement for every
    collector), else it is counted here."""
    from app.services import procinfo
    from app.services.ingest import is_stuck

    if not source.enabled:
        return Refusal("SOURCE_DISABLED", f"{source.slug} is switched off (the kill switch); switch it on first.")
    if not prefetched:
        # `last` is the newest FINISHED run, exactly as `live()` ranks it: a dead `running` row
        # newer than a `blocked` one once hid the refusal from the route while the page showed
        # it, so POST /start let a refused source through. The two must never diverge.
        last = db.scalar(
            select(CollectionRun).where(CollectionRun.source_id == source.id, CollectionRun.status != "running")
            .order_by(CollectionRun.started_at.desc(), CollectionRun.id.desc()).limit(1)
        )
        running = running_run(db, source.id)
        refused_since = refused_since_of(db, source.id)
    host = host_of(collector, source)
    if last is not None and last.status == "blocked":
        return Refusal("SOURCE_REFUSED",
                       f"{host} refused us on {_when(last.started_at)}: {(last.error or 'refused')[:160]}. "
                       f"A refusal is final; only a decision recorded by rian lifts it.")
    if refused_since is not None and (last is None or last.started_at is None or _aware(last.started_at) < _aware(refused_since)):
        return Refusal("SOURCE_REFUSED",
                       f"{host} refused our declared identity on {_when(refused_since)} (a verification check). "
                       f"A refusal is final; only a decision recorded by rian lifts it.")
    if running is not None and not is_stuck(running, now):
        return Refusal("SOURCE_RUNNING", f"run {running.id} has been collecting since {_when(running.started_at)}; stop it first.")
    if mode not in MODES:
        return Refusal("MODE_UNSUPPORTED", f"unknown mode {mode!r}; discover or recheck.")
    if mode == "recheck":
        refusal = recheck_refusal(collector)
        if refusal is not None:
            return refusal
        if (held_listings_count(db, collector) if held_listings is None else held_listings) == 0:
            return Refusal("NOTHING_TO_RECHECK", "no held listing at this source has a URL to read back.")
    freeze = procinfo.freeze_state(freeze_marker(), now)
    if freeze.frozen:
        age = freeze.age(now)
        minutes = int(age.total_seconds() // 60) if age else 0
        return Refusal("COLLECT_FROZEN", f"collection is frozen for a deploy: {freeze.stamp or 'no stamp'} ({minutes} min ago).")
    if memory is not None and getattr(memory, "readable", False) and memory.used_bytes is not None \
            and memory.used_bytes > procinfo.START_HEADROOM_BYTES:
        return Refusal("MEMORY_LOW",
                       f"the app container is using {_gib(memory.used_bytes)} GiB; starts are refused above "
                       f"{_gib(procinfo.START_HEADROOM_BYTES)} GiB (the guard sheds at {_gib(procinfo.SHED_AT_BYTES)} GiB).")
    return None


def _gib(value: int) -> str:
    return f"{value / 1024 ** 3:.1f}"


def freeze_marker():
    from app.config import settings

    return settings.collect_freeze_file


# --------------------------------------------------------------------------- the pace and its floor

@dataclass(frozen=True, slots=True)
class Pace:
    delay_seconds: float
    robots_crawl_delay: float | None
    robots_read_at: datetime | None
    floor: float
    floor_reason: str  # robots | ours | render
    render_floor: float | None
    recommended: float
    cap: float
    set_by: str | None
    set_at: datetime | None

    def as_dict(self) -> dict[str, Any]:
        return {
            "delay_seconds": self.delay_seconds, "robots_crawl_delay": self.robots_crawl_delay,
            "robots_read_at": self.robots_read_at.isoformat() if self.robots_read_at else None,
            "floor": self.floor, "floor_reason": self.floor_reason, "render_floor": self.render_floor,
            "recommended": self.recommended, "cap": self.cap, "set_by": self.set_by,
            "set_at": self.set_at.isoformat() if self.set_at else None,
        }


def pace_of(source: Source, collector: Any) -> Pace:
    """The floor is the host's robots crawl delay when a run has read one, else our own second;
    a rendered source never goes under the render floor. Recommended is the slower of the host's
    ask and the floor."""
    from app.services.collectors.fetch import RENDER_FLOOR_DELAY

    robots = float(source.robots_crawl_delay) if source.robots_crawl_delay is not None else None
    floor, reason = (robots, "robots") if robots is not None else (OUR_FLOOR, "ours")
    render_floor = None
    if collector is not None and is_rendered(collector):
        render_floor = max(RENDER_FLOOR_DELAY, float(getattr(collector, "render_floor_seconds", 0) or 0))
        if render_floor > floor:
            floor, reason = render_floor, "render"
    return Pace(
        delay_seconds=float(source.delay_seconds or 0.0), robots_crawl_delay=robots,
        robots_read_at=source.robots_read_at, floor=floor, floor_reason=reason, render_floor=render_floor,
        recommended=max(robots or 0.0, floor), cap=PACE_CAP, set_by=source.delay_set_by, set_at=source.delay_set_at,
    )


def pace_refusal(source: Source, collector: Any, delay_seconds: float) -> Refusal | None:
    pace = pace_of(source, collector)
    if delay_seconds < pace.floor:
        if pace.floor_reason == "robots":
            why = f"the host asks for at least {pace.floor:g} s (robots.txt read {_when(pace.robots_read_at)})"
        elif pace.floor_reason == "render":
            why = f"a rendered source waits at least {pace.floor:g} s"
        else:
            why = f"our own floor is {pace.floor:g} s"
        return Refusal("PACE_BELOW_FLOOR", f"{delay_seconds:g} s is under the floor: {why}.")
    if delay_seconds > pace.cap:
        return Refusal("PACE_ABOVE_CAP", f"{delay_seconds:g} s is over the cap of {pace.cap:g} s.")
    return None


# --------------------------------------------------------------------------- the spawn

def collect_argv(slug: str, mode: str, limit: int | None, by: str) -> list[str]:
    argv = [sys.executable, "-m", "app.cli", "collect", "--source", slug, "--mode", mode, "--by", by]
    if limit:
        argv += ["--limit", str(int(limit))]
    return argv


def spawn_collector(slug: str, mode: str, limit: int | None, by: str) -> subprocess.Popen:
    """One collector process, its own session, stdio inherited (the container's log). A daemon
    thread reaps it so no zombie is left, and logs how it ended."""
    argv = collect_argv(slug, mode, limit, by)
    proc = subprocess.Popen(argv, start_new_session=True, close_fds=True)  # noqa: S603 - argv is a list we built
    logger.info("collector_spawned source=%s mode=%s limit=%s by=%s pid=%d", slug, mode, limit, by, proc.pid)

    def reap() -> None:
        code = proc.wait()
        logger.info("collector_exited source=%s pid=%d code=%d", slug, proc.pid, code)

    threading.Thread(target=reap, name=f"reap-{slug}", daemon=True).start()
    return proc


# --------------------------------------------------------------------------- what a route reads

def running_run(db: Session, source_id: int) -> CollectionRun | None:
    return db.scalar(
        select(CollectionRun).where(CollectionRun.source_id == source_id, CollectionRun.status == "running")
        .order_by(CollectionRun.started_at.desc()).limit(1)
    )