"""The `log_events` sink: the stored copy of the log the diagnostics console reads.

stdout stays the complete log (docker keeps 3×50 MB of it); this table keeps the part worth querying
for `log_retention_days`:

  * every WARNING and above,
  * the `http.request` line of every request at INFO or above (not the DEBUG-level health checks and
    static files),
  * every line of a request whose user raised their diagnostics level (`logs.LevelGate`
    marks them), which is the detail they consented to have kept,
  * nothing a caller marked `extra={logs.PERSIST: False}` — how the request log caps anonymous
    traffic per client.

What the table must never do is slow down or break the code that logs. So a line is redacted and
turned into a row on the CALLER's thread (the same `logs.record_fields` the stdout line is built
from), put on a bounded queue without blocking, and written in batches by one background thread
on its own connection — never inside a request's transaction. A full queue drops the row and
counts it; a failed write drops the batch and says so on stderr. Both are reported rather than
retried: a log that retries into a sick database is how a small outage becomes a big one.

Flood control is the second half. On 2026-09-16 a YouTube RSS outage made prod log 11k near-identical
warnings in four hours. stdout keeps them all; the table keeps `REPEATS_PER_MINUTE` of each message
template per minute and one `log.repeats_suppressed` row saying how many more there were.

Core layer: imports `logs`, `db`, `models`, `config` — nothing above them.
"""

import atexit
import json
import logging
import queue
import re
import sys
import threading
import time
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import delete, insert
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session

from app import logs
from app.config import settings
from app.db import engine
from app.models import LogEvent
from app.utils import iso

QUEUE_MAX = 10_000
BATCH_MAX = 500
FLUSH_SECONDS = 1.0
REPEATS_PER_MINUTE = 30
REPEAT_WINDOW_SECONDS = 60.0
# Distinct templates tracked per window. Past this many, a new template is stored without counting
# — the limiter must not become the unbounded thing it exists to prevent.
_REPEAT_KEYS_MAX = 2_000

# Column widths from models.LogEvent. A value past its column would fail the WHOLE batch it rides in.
_LIMITS = {
    "level": 10,
    "source": 16,
    "logger": 128,
    "event": 64,
    "request_id": 64,
    "debug_session_id": 64,
    "version": 64,
    "msg": 8_000,
    "exc": 32_000,
}

_log = logs.get_logger(__name__)


def should_store(record: logging.LogRecord) -> bool:
    """Whether this record belongs in `log_events` (see the module docstring)."""
    if getattr(record, logs.PERSIST, True) is False:
        return False
    if record.levelno >= logging.WARNING or getattr(record, "elevated", False):
        return True
    return getattr(record, "event", None) == "http.request" and record.levelno >= logging.INFO


def _without_nul(value: Any) -> Any:
    """`value` with every NUL character replaced, in keys and nested values alike. Postgres refuses
    NUL in text and `\\u0000` in JSONB, and a refused row fails the WHOLE batch it rides in — every
    user's rows with it. Anything logged can carry one (`GET /assets/%00` logs the decoded path)."""
    if isinstance(value, str):
        return value.replace("\x00", "�")
    if isinstance(value, dict):
        return {_without_nul(k): _without_nul(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_without_nul(v) for v in value]
    return value


def to_row(fields: dict[str, Any]) -> dict[str, Any]:
    """`logs.record_fields` output → a `log_events` row: known columns only, text clipped to its
    column and free of NUL, attrs made JSON-safe (anything exotic becomes its `str`, as on stdout)."""
    row = {
        column: fields.get(column) for column in LogEvent.__table__.columns.keys() if column != "id"
    }
    for column, limit in _LIMITS.items():
        value = row.get(column)
        if isinstance(value, str):
            value = _without_nul(value)
            row[column] = value[: limit - 1] + "…" if len(value) > limit else value
    if row["attrs"] is not None:
        row["attrs"] = _without_nul(
            json.loads(json.dumps(row["attrs"], ensure_ascii=False, default=str))
        )
    return row


def without_erased(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """The rows, with the account taken off any whose account was erased meanwhile
    (`logs.forget_user`). Checked right before the INSERT, under `logs.erasure_guard()` held until
    the commit: a row built before the erasure can still be waiting for its batch."""
    for row in rows:
        if logs.is_erased(row.get("user_id")):
            row["user_id"] = None
    return rows


def _notice(code: str, msg: str, **attrs: Any) -> dict[str, Any]:
    """A row the sink writes about ITSELF. Built directly, not logged: logging it would route it back
    through this sink (and the writer thread's own records are deliberately ignored)."""
    return to_row(
        {
            "ts": datetime.now(timezone.utc),
            "level": "WARNING",
            "sev": logs.SEVERITY["WARNING"],
            "source": logs.current_source(),
            "logger": _log.name,
            "event": code,
            "msg": msg,
            "attrs": attrs,
            "version": settings.app_version,
        }
    )


_DIGITS = re.compile(r"\d+")


def _template(record: logging.LogRecord) -> str:
    """The flood limiter's grouping key: the message template. Except when the template is a bare
    `%s` — a pass-through for someone else's text (yt-dlp's lines, `formats._YdlLog`), where the
    template says nothing and every distinct line would share one bucket. Then the text itself
    groups, with its numbers folded, so `Retrying fragment 5 (1/10)` and `(2/10)` are still one
    repeat and a different warning is not crowded out by them."""
    msg = str(record.msg)
    if msg != "%s" or not record.args:
        return msg
    return _DIGITS.sub("#", record.getMessage())[:200]


class _Repeats:
    """Per-template admission for one minute at a time. Used by the writer thread only (unlocked)."""

    def __init__(self, per_window: int, window_seconds: float):
        self.per_window = per_window
        self.window = window_seconds
        self.started = time.monotonic()
        self.seen: dict[tuple[str, str, int], int] = {}

    def admit(self, key: tuple[str, str, int] | None) -> bool:
        if key is None:
            return True
        count = self.seen.get(key)
        if count is None and len(self.seen) >= _REPEAT_KEYS_MAX:
            return True
        self.seen[key] = (count or 0) + 1
        return self.seen[key] <= self.per_window

    def roll(self, now: float, *, final: bool = False) -> list[dict[str, Any]]:
        """Close the window if it is over — or, `final`, because the process is: one notice per
        template that went past the limit. Without `final` a process that lives under a minute
        (every download child) would drop its repeats and never say so."""
        if not final and now - self.started < self.window:
            return []
        notices = [
            _notice(
                "log.repeats_suppressed",
                f"{count - self.per_window} more like this were not stored",
                logger=logger,
                template=logs.redact_text(template),
                suppressed=count - self.per_window,
                window_seconds=int(self.window),
            )
            for (logger, template, _level), count in self.seen.items()
            if count > self.per_window
        ]
        self.seen.clear()
        self.started = now
        return notices


class LogSink(logging.Handler):
    """The handler plus its writer. `start()` runs the writer thread; a test drives `write_pending()`
    directly instead, on its own thread, with no timing involved."""

    def __init__(self, bind: Engine = engine):
        super().__init__(level=logging.NOTSET)
        # Correlation fields are contextvars: capture them now, on the thread that logged.
        self.addFilter(logs.ContextFilter())
        self._bind = bind
        self._queue: queue.Queue[tuple[tuple[str, str, int] | None, dict[str, Any]]] = queue.Queue(
            QUEUE_MAX
        )
        self._dropped = 0
        self._dropped_lock = threading.Lock()
        self._repeats = _Repeats(REPEATS_PER_MINUTE, REPEAT_WINDOW_SECONDS)
        self._stopping = threading.Event()
        self._writer: threading.Thread | None = None

    # --- caller side ---------------------------------------------------------------------------
    def emit(self, record: logging.LogRecord) -> None:
        if threading.current_thread() is self._writer:
            return  # the writer's own trouble (a driver warning mid-insert) must not feed itself
        if not should_store(record):
            return
        try:
            row = to_row(logs.record_fields(record))
        # silent-ok: `handleError` is the report — logging's own out-of-band channel to stderr.
        # A log call here would be handed to this handler, which is the one that just failed.
        except Exception:  # noqa: BLE001 — a record we cannot turn into a row is reported, not raised
            self.handleError(record)
            return
        # Request lines are one per request by nature; only the repeated SAME message is a flood.
        key = (
            None
            if row["event"] == "http.request"
            else (record.name, _template(record), record.levelno)
        )
        try:
            self._queue.put_nowait((key, row))
        # silent-ok: counted, not logged — the count goes out as the `log.sink.dropped` notice row
        # the writer emits, which is the one report that does not feed this queue again.
        except queue.Full:
            with self._dropped_lock:
                self._dropped += 1

    # --- writer side ---------------------------------------------------------------------------
    def write_pending(self, wait: float = 0.0, *, final: bool = False) -> int:
        """Write one batch: whatever is queued (up to BATCH_MAX), waiting up to `wait` seconds for the
        first item. Returns the number of rows written, notices included. `final` closes the
        repeat window early — the last batch before the process exits."""
        rows: list[dict[str, Any]] = []
        try:
            item = self._queue.get(timeout=wait) if wait > 0 else self._queue.get_nowait()
        # silent-ok: an idle sink. Nothing failed; there was simply nothing waiting.
        except queue.Empty:
            item = None
        while item is not None:
            key, row = item
            if self._repeats.admit(key):
                rows.append(row)
            if len(rows) >= BATCH_MAX:
                break
            try:
                item = self._queue.get_nowait()
            # silent-ok: the batch ends here because the queue ran dry, which is the normal end.
            except queue.Empty:
                item = None
        rows.extend(self._repeats.roll(time.monotonic(), final=final))
        with self._dropped_lock:
            dropped, self._dropped = self._dropped, 0
        if dropped:
            rows.append(
                _notice("log.sink.dropped", f"{dropped} log events were dropped", dropped=dropped)
            )
        if not rows:
            return 0
        try:
            # The connection is checked out BEFORE the guard, and every request path takes them in
            # that order too (its session already holds a connection when it reaches the guard). The
            # other way round, this thread would hold a process-global lock while waiting up to
            # `pool_timeout` for a connection, and the requests that could return one are the ones
            # blocked on the lock. The guard still spans the check, the insert and the commit.
            with self._bind.connect() as conn, logs.erasure_guard(), conn.begin():
                conn.execute(insert(LogEvent), without_erased(rows))
        except Exception as exc:  # noqa: BLE001 — the database is down or refused the batch
            # silent-ok / noqa T201: the sink's OWN failure. A log call here would be handed to
            # this sink, which is the thing that just failed — straight into a loop. stderr is
            # the one way out that does not re-enter.
            print(  # noqa: T201
                f"log sink: could not store {len(rows)} log events: "
                f"{logs.redact_text(str(exc))[:300]}",
                file=sys.stderr,
            )
            return 0
        return len(rows)

    def _run(self) -> None:
        while not self._stopping.is_set():
            self.write_pending(wait=FLUSH_SECONDS)
        while self.write_pending():  # drain what arrived before the stop
            pass
        self.write_pending(final=True)  # and the repeat notices of a window cut short

    def start(self) -> None:
        self._writer = threading.Thread(target=self._run, name="log-sink", daemon=True)
        self._writer.start()

    def stop(self, timeout: float = 5.0) -> None:
        self._stopping.set()
        if self._writer is not None:
            self._writer.join(timeout)


_sink: LogSink | None = None


def start() -> None:
    """Attach the sink to the root logger and start its writer. Call after `logs.configure_logging`
    (which replaces the root handlers) and once the schema exists. A no-op when `LOG_DB_SINK` is off
    or it is already running."""
    global _sink
    if _sink is not None or not settings.log_db_sink:
        return
    _sink = LogSink()
    # The same level gate as stdout: the root may be lowered for one user's elevated requests, and
    # everyone else's detail must not reach the table through that gap either.
    _sink.addFilter(logs.LevelGate())
    logging.getLogger().addHandler(_sink)
    _sink.start()
    # A short-lived process (the download child) exits without a lifespan: flush on the way out.
    atexit.register(stop)


def stop() -> None:
    """Detach the sink and write what is still queued."""
    global _sink
    if _sink is None:
        return
    logging.getLogger().removeHandler(_sink)
    _sink.stop()
    _sink = None


def gc(db: Session, retention_days: int) -> int:
    """Stage the deletion of rows older than `retention_days` (at least one day — stored logs always
    expire). The caller commits. Returns the number of rows removed."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days))
    result = db.execute(delete(LogEvent).where(LogEvent.ts < cutoff))
    return result.rowcount or 0


def serialize(row: LogEvent) -> dict:
    """A stored row as the API sends it — one shape for every reader: the admin console and
    a debug session's evidence list."""
    return {
        "id": row.id,
        "ts": iso(row.ts),
        "level": row.level,
        "source": row.source,
        "logger": row.logger,
        "event": row.event,
        "msg": row.msg,
        "attrs": row.attrs,
        "exc": row.exc,
        "request_id": row.request_id,
        "user_id": row.user_id,
        "job_id": row.job_id,
        "run_id": row.run_id,
        "debug_session_id": row.debug_session_id,
        "version": row.version,
    }