"""Сторож собственных прогонов: журнал `job_runs` наконец кто-то читает.

Планировщик gateway пишет строку на каждый прогон джоба (`job_runs`: имя,
статус, хвост трейсбека в `error`). Запись была всегда, читателя не было
ни одного: канал к соседнему агенту лежал 38 часов подряд, планировщик
честно записал это 231 раз, и до человека не доехало ничего. Этот модуль -
вторая половина прибора: показание из таблицы превращается в тревогу.

Что здесь считается показанием:

- тревога берётся только с ХВОСТА: три и более `failed` подряд, считая от
  последнего прогона джоба. Одиночный упавший прогон - сетевой чих, а не
  болезнь, и будить на него человека значит выучить его не смотреть;
- окно счёта - неделя, а не сутки: иначе суточный джоб не наберёт трёх
  падений подряд никогда, и `отметка`, `закрытие`, `голосовой`, `память`
  не охранялись бы вовсе;
- отсутствие записей - тоже показание. Джоб с регулярным кроном, у которого
  последний прогон старше `FACTOR` его периодов, «не запускался» - это
  отдельная тревога, не «лежит». Период выводится из крона тем же
  `every_from_cron`, что у сторожа соседа, и расписание, которое не
  сводится к «раз в N часов», в давности не участвует и называется в логе
  неохраняемым: подставить ему число значило бы соврать. Джобы без крона
  (вебхуки, события) давностью не меряются вовсе, а джоб, которого в
  журнале нет ни разу, молча ждёт своей первой записи;
- пустая выдача - это показание о запросе, а не о здоровье джобов. Ноль
  строк за сутки (кроме своих) означает, что планировщик вообще не
  работал, и это отдельная тревога с отдельным текстом, а не тишина;
- упавший запрос к БД - не показание вовсе: решает джоб, здесь такого
  входа просто нет.

Свои собственные прогоны из счёта выкинуты (`me`): джоб, который лежит,
о себе не доложит - наблюдатель внутри объекта не сообщает о смерти
объекта. Это дыра прибора, и она названа в README, а не заклеена.

Текст тревоги проходит `mask`: из `error` берётся первая строка, режется
по длине и чистится - общим `redact` gateway плюс два своих правила на то,
что `redact` по конструкции не ловит внутри строки (`ИМЯ=значение` не с
начала строки и хвост запроса у URL). Тревога уезжает в чат к диспетчеру,
логи гейтвея видны в чатах - секрету тут негде быть.
"""

from __future__ import annotations

import itertools
import json
import re
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path  # noqa: TC003 - в аннотации dataclass
from typing import TYPE_CHECKING

from beaver_gateway.jobs.scheduler import next_run
from beaver_gateway.security.redact import MASK, is_secret_name, redact

from beaver_agent.hands.heartbeat import every_from_cron

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence
    from datetime import tzinfo

__all__ = [
    "DAY",
    "ERROR_CHARS",
    "FACTOR",
    "FAILED",
    "FORGET",
    "MAX_LINES",
    "REPEAT",
    "STREAK",
    "WINDOW",
    "Clear",
    "Quiet",
    "Report",
    "Resumed",
    "Run",
    "Runs",
    "Streak",
    "alarm_text",
    "awake",
    "clear_text",
    "empty_text",
    "first_line",
    "look",
    "mask",
    "quiet_text",
    "streak",
]

WINDOW = timedelta(days=7)
"""Окно журнала: неделя. Суточный джоб набирает в нём три падения подряд за
трое суток, недельный (`память`) - нет: у него три падения это три недели,
и его охраняет давность, а не серия."""
DAY = timedelta(hours=24)
"""Окно тревоги о пустом журнале: сутки без единого чужого прогона."""
FACTOR = 2
"""Давность = столько периодов джоба без записи. Два - это «пропущено два
срабатывания подряд»: один пропуск случается честно (передеплой пришёлся на
минуту крона, а догонять пропущенное планировщик не умеет), два подряд -
уже не случайность."""
GRAIN = timedelta(minutes=1)
"""Шаг, в котором период крона должен сводиться к целому: крон минутный."""
LAG = timedelta(minutes=10)
"""Запись в журнал ложится по КОНЦУ прогона: срабатывание моложе этого
могло ещё не дописаться, и пропуском его не считаем."""
STREAK = 3
"""Сколько `failed` подряд считаем «джоб лежит»."""
REPEAT = timedelta(hours=24)
"""Чаще этого об одном и том же джобе не говорим."""
FORGET = timedelta(days=7)
"""Джоб, о котором тревожились, исчез из окна насовсем - забываем его молча:
починки мы не видели, врать «отбой» не о чем."""
ERROR_CHARS = 160
"""Сколько букв первой строки ошибки уезжает человеку."""
MAX_LINES = 10
"""Сколько джобов перечисляем в одной тревоге: сразу всё не читают."""
FAILED = "failed"
_DAYS_FROM = timedelta(hours=48)
STATE_VERSION = 1

_PAIR = re.compile(r"(?=([A-Za-z][\w.-]{2,})\s*([:=])\s*([^\s,;)\]}\"']{6,}))")
"""`ИМЯ=значение` где угодно в строке: общий `redact` ловит такие пары только
с начала строки, а трейсбек кладёт их в середину. Просмотр нулевой длины -
чтобы `RuntimeError: MSOS_MCP=…` не съедало секрет первой же парой: обычный
`sub` откусил бы `RuntimeError:` вместе со значением и до вложенной пары
не дошёл."""
_QUERY = re.compile(r"(https?://[^\s?#]+)\?[^\s#]*")
"""Хвост запроса у URL сносим целиком: имя параметра с ключом может быть любым."""


def _iso(when: datetime | None) -> str | None:
    if when is None:
        return None
    return when.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")


def _parse_iso(raw: str | None) -> datetime | None:
    if not raw:
        return None
    try:
        when = datetime.fromisoformat(raw)
    except ValueError:
        return None
    return when if when.tzinfo else when.replace(tzinfo=UTC)


def _plural(count: int, forms: tuple[str, str, str]) -> str:
    if count % 10 == 1 and count % 100 != 11:
        return forms[0]
    if count % 10 in (2, 3, 4) and count % 100 not in (12, 13, 14):
        return forms[1]
    return forms[2]


def _age(delta: timedelta) -> str:
    """Возраст человеку: «4 часа», «45 минут», «15 суток». Секунды не нужны."""
    minutes = max(int(delta.total_seconds()) // 60, 0)
    if minutes < 60:
        return f"{minutes} {_plural(minutes, ('минуту', 'минуты', 'минут'))}"
    hours = minutes // 60
    if delta >= _DAYS_FROM:
        days = hours // 24
        return f"{days} {_plural(days, ('сутки', 'суток', 'суток'))}"
    return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}"


def _period(delta: timedelta) -> str:
    """Период крона: «10 минут», «24 часа», «7 суток» - без округления."""
    minutes = int(delta.total_seconds()) // 60
    if minutes % (24 * 60) == 0 and delta >= _DAYS_FROM:
        days = minutes // (24 * 60)
        return f"{days} {_plural(days, ('сутки', 'суток', 'суток'))}"
    if minutes % 60 == 0:
        hours = minutes // 60
        return f"{hours} {_plural(hours, ('час', 'часа', 'часов'))}"
    return f"{minutes} {_plural(minutes, ('минута', 'минуты', 'минут'))}"


def _mask_pairs(text: str) -> str:
    out: list[str] = []
    pos = 0
    for match in _PAIR.finditer(text):
        if match.start(3) < pos or not is_secret_name(match[1]):
            continue
        out.append(text[pos : match.start(3)])
        out.append(MASK)
        pos = match.end(3)
    out.append(text[pos:])
    return "".join(out)


def mask(text: str) -> str:
    """Убрать из строки всё, что похоже на секрет, прежде чем её показывать."""
    return _QUERY.sub(r"\1?" + MASK, _mask_pairs(redact(text)))


def first_line(error: str | None) -> str:
    """Первая строка ошибки - чистая и короткая. Трейсбек человеку не нужен."""
    for line in mask(error or "").splitlines():
        head = line.strip()
        if not head:
            continue
        if len(head) > ERROR_CHARS:
            head = head[: ERROR_CHARS - 1].rstrip() + "…"
        return head
    return "без текста ошибки"


@dataclass(frozen=True, slots=True)
class Run:
    """Одна строка журнала: ровно то, на что мы смотрим, без ORM."""

    id: int
    job: str
    status: str
    started_at: datetime
    error: str | None = None

    @property
    def failed(self) -> bool:
        return self.status == FAILED


@dataclass(frozen=True, slots=True)
class Streak:
    """Хвост подряд упавших прогонов одного джоба."""

    job: str
    count: int
    since: datetime
    """Начало серии - первое падение из тех, что видно в окне."""
    age: timedelta
    error: str
    capped: bool
    """Серия упёрлась в край окна: падений было НЕ МЕНЬШЕ `count`, а сколько
    на самом деле - за неделей не видно, и врать точным числом нельзя."""


@dataclass(frozen=True, slots=True)
class Quiet:
    """Джоб с кроном, который перестал запускаться."""

    job: str
    age: timedelta
    """Сколько прошло с начала последнего прогона - любого, хоть упавшего."""
    period: timedelta


@dataclass(frozen=True, slots=True)
class Resumed:
    """Джоб, о молчании которого говорили, снова запустился."""

    job: str
    status: str


@dataclass(frozen=True, slots=True)
class Clear:
    """Джоб, о котором мы тревожились, больше не лежит."""

    job: str
    status: str
    """Статус последнего прогона."""
    count: int
    """Сколько подряд упавших осталось в хвосте: 0 - чисто."""


@dataclass(slots=True)
class Runs:
    """Состояние между прогонами: кому и когда уже сказали."""

    path: Path
    alerted: dict[str, datetime] = field(default_factory=dict)
    """Джоб → когда о нём ушла тревога. Он же признак «сейчас в тревоге»."""
    quiet: dict[str, datetime] = field(default_factory=dict)
    """То же для «не запускался»: отдельно, у тревог разные отбои."""
    empty_alerted: datetime | None = None

    @classmethod
    def load(cls, path: Path) -> Runs:
        try:
            raw = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return cls(path=path)
        if raw.get("version") != STATE_VERSION:
            return cls(path=path)
        return cls(
            path=path,
            alerted=_stamps(raw.get("alerted")),
            quiet=_stamps(raw.get("quiet")),
            empty_alerted=_parse_iso(raw.get("empty_alerted")),
        )

    def save(self) -> None:
        body = {
            "version": STATE_VERSION,
            "alerted": {job: _iso(when) for job, when in sorted(self.alerted.items())},
            "quiet": {job: _iso(when) for job, when in sorted(self.quiet.items())},
            "empty_alerted": _iso(self.empty_alerted),
        }
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(body, ensure_ascii=False, indent=1), encoding="utf-8")
        tmp.replace(self.path)


def _stamps(raw: dict[str, str] | None) -> dict[str, datetime]:
    out = {}
    for job, when in (raw or {}).items():
        parsed = _parse_iso(str(when))
        if parsed is not None:
            out[str(job)] = parsed
    return out


@dataclass(frozen=True, slots=True)
class Report:
    """Что решил один прогон сторожа журнала."""

    alarms: tuple[Streak, ...]
    clears: tuple[Clear, ...]
    empty: bool
    """Прогонов за сутки не было вовсе - тревога о планировщике, не о джобе."""
    seen: int
    """Сколько чужих прогонов было в окне."""
    mine: int
    """Сколько своих - их в счёт не берём, но в тексте называем."""
    note: str
    quiet: tuple[Quiet, ...] = ()
    resumed: tuple[Resumed, ...] = ()
    unguarded: tuple[str, ...] = ()
    """Джобы с кроном, у которого нет регулярного периода: давностью не меряем."""


def streak(runs: Sequence[Run], *, now: datetime) -> Streak | None:
    """Хвост падений одного джоба, считая от последнего прогона.

    `runs` - прогоны одного джоба, новейший первым. Серия, упёршаяся в край
    окна, помечается `capped`: её настоящую длину мы не знаем.
    """
    failures = list(itertools.takewhile(lambda run: run.failed, runs))
    if not failures:
        return None
    oldest = failures[-1]
    return Streak(
        job=failures[0].job,
        count=len(failures),
        since=oldest.started_at,
        age=now - oldest.started_at,
        error=first_line(failures[0].error),
        capped=len(failures) == len(runs),
    )


def look(
    state: Runs,
    runs: Sequence[Run],
    *,
    now: datetime,
    me: str,
    crons: Mapping[str, str] | None = None,
    latest: Mapping[str, Run] | None = None,
    tz: tzinfo = UTC,
    boot: datetime | None = None,
) -> Report:
    """Разобрать окно журнала и решить, о чём говорить человеку.

    `crons` - джоб → крон, только джобы с кроном; `latest` - последний прогон
    каждого джоба за ВСЮ историю журнала, а не за окно; `boot` - старт
    процесса. Без `crons` давность не проверяется вовсе.

    Состояние здесь только чистится от забытых джобов; отметки «сказали» и
    «сказали про пустоту» ставит джоб - и только когда тревога действительно
    ушла: несостоявшаяся доставка не должна заглушать следующий прогон.
    """
    mine = [run for run in runs if run.job == me]
    others = sorted(
        (run for run in runs if run.job != me), key=lambda run: run.id, reverse=True
    )
    if not any(now - run.started_at < DAY for run in others):
        quiet = state.empty_alerted is not None and now - state.empty_alerted < REPEAT
        return Report(
            alarms=(),
            clears=(),
            empty=not quiet,
            seen=len(others),
            mine=len(mine),
            note=(
                f"чужих прогонов за сутки нет вовсе (своих {len(mine)})"
                + ("; о пустоте уже говорили меньше суток назад" if quiet else "")
            ),
        )

    by_job: dict[str, list[Run]] = {}
    for run in others:
        by_job.setdefault(run.job, []).append(run)

    alarms: list[Streak] = []
    clears: list[Clear] = []
    held: list[str] = []
    for job, rows in by_job.items():
        tail = streak(rows, now=now)
        if tail is not None and tail.count >= STREAK:
            told = state.alerted.get(job)
            if told is None or now - told >= REPEAT:
                alarms.append(tail)
            else:
                held.append(job)
        elif job in state.alerted:
            clears.append(
                Clear(
                    job=job,
                    status=rows[0].status,
                    count=0 if tail is None else tail.count,
                )
            )

    forgotten = [
        job
        for job, told in state.alerted.items()
        if job not in by_job and now - told >= FORGET
    ]
    for job in forgotten:
        del state.alerted[job]

    note = (
        f"прогонов за неделю {len(others)} по {len(by_job)} джобам (своих {len(mine)})"
    )
    if alarms:
        note += "; лежат: " + ", ".join(sorted(tail.job for tail in alarms))
    if held:
        note += "; уже говорили меньше суток назад: " + ", ".join(sorted(held))
    if clears:
        note += "; починились: " + ", ".join(sorted(clear.job for clear in clears))
    if forgotten:
        note += "; забыты (пропали из окна): " + ", ".join(sorted(forgotten))

    silence = _silence(
        state,
        mine,
        now=now,
        me=me,
        crons=crons or {},
        latest=latest or {},
        tz=tz,
        boot=boot,
    )

    return Report(
        alarms=tuple(sorted(alarms, key=lambda tail: tail.count, reverse=True)),
        clears=tuple(sorted(clears, key=lambda clear: clear.job)),
        empty=False,
        seen=len(others),
        mine=len(mine),
        note=note + silence.note,
        quiet=silence.quiet,
        resumed=silence.resumed,
        unguarded=silence.unguarded,
    )


@dataclass(frozen=True, slots=True)
class _Silence:
    quiet: tuple[Quiet, ...]
    resumed: tuple[Resumed, ...]
    unguarded: tuple[str, ...]
    note: str


def awake(
    mine: Sequence[Run], *, now: datetime, step: timedelta, boot: datetime | None
) -> datetime:
    """С какого момента планировщик заведомо крутил некритичные джобы.

    Показание - собственные прогоны сторожа: он сам некритичный, и когда
    гейтвей лежит или бюджет зажат (`throttled` откладывает некритичный крон
    без записи в журнал), его прогонов нет так же, как и чужих. Цепочка своих
    прогонов без разрыва длиннее `step`, считая от сейчас, плюс старт
    процесса - это время, за которое чужой пропуск действительно пропуск,
    а не общий простой.
    """
    anchor = now
    for run in sorted(mine, key=lambda run: run.started_at, reverse=True):
        if anchor - run.started_at > step:
            break
        anchor = min(anchor, run.started_at)
    return anchor if boot is None else max(anchor, boot)


def _silence(
    state: Runs,
    mine: Sequence[Run],
    *,
    now: datetime,
    me: str,
    crons: Mapping[str, str],
    latest: Mapping[str, Run],
    tz: tzinfo,
    boot: datetime | None,
) -> _Silence:
    """Давность: какие джобы с кроном перестали запускаться."""
    own = every_from_cron(crons.get(me), tz, now=now, grain=GRAIN)
    guarded: dict[str, timedelta] = {}
    unguarded: list[str] = []
    for job, cron in crons.items():
        if job == me:
            continue
        period = every_from_cron(cron, tz, now=now, grain=GRAIN)
        if period is None:
            unguarded.append(job)
        else:
            guarded[job] = period

    note = ""
    if unguarded:
        note += "; давностью не охраняются (крон без регулярного периода): " + (
            ", ".join(sorted(unguarded))
        )
    dropped = [job for job in state.quiet if job not in guarded]
    for job in dropped:
        del state.quiet[job]
    if dropped:
        note += "; сняты с давности (нет в расписании): " + ", ".join(sorted(dropped))
    if not guarded:
        return _Silence((), (), tuple(sorted(unguarded)), note)
    if own is None:
        # Свой период не выведен - нет и опоры, когда планировщик заведомо
        # работал. Любое число тут было бы выдумано.
        note += "; свой крон без регулярного периода - давность не проверяется"
        return _Silence((), (), tuple(sorted(unguarded)), note)

    since = awake(mine, now=now, step=own * FACTOR, boot=boot)
    quiet: list[Quiet] = []
    resumed: list[Resumed] = []
    held: list[str] = []
    fresh: list[str] = []
    for job, period in guarded.items():
        last = latest.get(job)
        if last is None:
            fresh.append(job)
            continue
        age = now - last.started_at
        # Порог не тоньше собственного: чаще раза в час сторож не смотрит, и
        # пропуск короче двух своих периодов неотличим от его же слепоты.
        overdue = (
            age > max(period, own) * FACTOR
            and next_run(crons[job], tz, max(last.started_at, since)) <= now - LAG
        )
        told = state.quiet.get(job)
        if overdue:
            if told is None or now - told >= REPEAT:
                quiet.append(Quiet(job=job, age=age, period=period))
            else:
                held.append(job)
        elif told is not None and last.started_at > told:
            resumed.append(Resumed(job=job, status=last.status))

    if fresh:
        note += "; ещё ни разу не запускались: " + ", ".join(sorted(fresh))
    if quiet:
        note += "; не запускаются: " + ", ".join(sorted(q.job for q in quiet))
    if held:
        note += "; о молчании уже говорили меньше суток назад: " + ", ".join(
            sorted(held)
        )
    if resumed:
        note += "; снова запускаются: " + ", ".join(sorted(r.job for r in resumed))
    return _Silence(
        quiet=tuple(sorted(quiet, key=lambda q: q.age, reverse=True)),
        resumed=tuple(sorted(resumed, key=lambda r: r.job)),
        unguarded=tuple(sorted(unguarded)),
        note=note,
    )


def streak_line(tail: Streak) -> str:
    """«джоб `вайбграм` лежит 4 часа, 24 прогона подряд с 12.09 18:00 UTC: …»."""
    runs = f"{tail.count} {_plural(tail.count, ('прогон', 'прогона', 'прогонов'))}"
    return (
        f"джоб `{tail.job}` лежит {_age(tail.age)}"
        f"{' и дольше' if tail.capped else ''}, {runs} подряд с "
        f"{tail.since.astimezone(UTC).strftime('%d.%m %H:%M UTC')}"
        + (" (это всё окно недели, серия началась раньше)" if tail.capped else "")
        + f": {tail.error}"
    )


def alarm_text(report: Report) -> str:
    """Тревога человеку: что лежит, сколько и с какой ошибкой."""
    lines = [f"- {streak_line(tail)}" for tail in report.alarms[:MAX_LINES]]
    rest = len(report.alarms) - len(lines)
    if rest > 0:
        lines.append(f"- и ещё {rest} {_plural(rest, ('джоб', 'джоба', 'джобов'))}")
    return "\n".join(lines)


def quiet_line(quiet: Quiet) -> str:
    """«джоб `закрытие` не запускался 50 часов при периоде 24 часа»."""
    return (
        f"джоб `{quiet.job}` не запускался {_age(quiet.age)} "
        f"при периоде {_period(quiet.period)}"
    )


def quiet_text(report: Report) -> str:
    """Тревога о молчании: какие джобы с кроном перестали запускаться."""
    lines = [f"- {quiet_line(quiet)}" for quiet in report.quiet[:MAX_LINES]]
    rest = len(report.quiet) - len(lines)
    if rest > 0:
        lines.append(f"- и ещё {rest} {_plural(rest, ('джоб', 'джоба', 'джобов'))}")
    return "\n".join(lines)


def resumed_line(resumed: Resumed) -> str:
    return f"отбой: джоб `{resumed.job}` снова запускается ({resumed.status})"


def clear_line(clear: Clear) -> str:
    if clear.count == 0:
        return f"отбой: джоб `{clear.job}` снова проходит ({clear.status})"
    return (
        f"отбой: джоб `{clear.job}` больше не лежит подряд "
        f"(упавших в хвосте {clear.count}, порог {STREAK})"
    )


def clear_text(report: Report) -> str:
    lines = [clear_line(clear) for clear in report.clears]
    lines += [resumed_line(resumed) for resumed in report.resumed]
    return "\n".join(f"- {line}" for line in lines)


def empty_text(report: Report) -> str:
    """Отдельная тревога: за сутки не отработал вообще никто."""
    return (
        f"За последние сутки в журнале прогонов (`job_runs`) нет ни одной записи, "
        f"кроме моих собственных (их {report.mine}): не отработал ни один джоб. "
        "Это показание о планировщике, а не о каком-то одном джобе. Смотреть "
        "стоит три вещи: поднялся ли планировщик вообще (в логе gateway "
        "`scheduler: no postgres driver - cron and webhooks are off`), живы ли "
        "кроны в pgqueuer, и не заведена ли база заново - на пустой базе эта же "
        "тревога выглядит так же."
    )