"""Declared emission cadence, evaluated by the observer (AUDIT-WP-0009-T05).

The contract is info-tech-canon's, not ours: the EmissionCadenceDeclaration
wire schema 0.1 (``infospace/schemas/emission-cadence.schema.yaml``), read at
contract digest ``b08b4d95fc4b0bd3``. The source owns and publishes the
declaration; audit-core accepts it on the sender registration and evaluates it
against the stream it actually holds. The standard keeps those roles apart and
so does this module: a declaration is a statement of intended cadence, never
evidence that the emission happens.

**What is evaluated here.** Only the ``expected-rate`` form: at least
``expected_min`` events of a class in each trailing window, below which a
finding is raised. That is the high-volume case, and the one question an
archive can answer on its own — how many it holds.

**What is accepted but evaluated elsewhere.** ``heartbeat-or-reconciliation``
entries are validated and kept, not re-evaluated. Their heartbeat half is the
registration's ``heartbeat_classes`` (T04) and their reconciliation half is
``GET /v1/reconciliation`` (T06), whose count is the ``compare_observed`` side
of the comparison. The ``compare_local`` side belongs to the source; audit-core
cannot see it and does not pretend to.

**The bound.** A rate held says the stream was not silent, not that it was
complete. A rate missed says the source may have stopped, not that it did.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, Iterable

WIRE_SCHEMA_VERSION = "0.1"
CONTRACT_DIGEST = "b08b4d95fc4b0bd3"

EXPECTED_RATE = "expected-rate"
HEARTBEAT_OR_RECONCILIATION = "heartbeat-or-reconciliation"
FORMS = (EXPECTED_RATE, HEARTBEAT_OR_RECONCILIATION)

BELOW_CADENCE = "below_declared_cadence"

_ISO = re.compile(
    r"^P(?=\d|T\d)(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?)?$"
)
_SHORT = re.compile(r"^([1-9][0-9]*)(s|m|h|d)$")
_UNIT = {"s": 1, "m": 60, "h": 3600, "d": 86400}

_TOP_KEYS = {"schema_version", "declaration_id", "source", "stream_id", "sources", "extensions"}
_ENTRY_KEYS = {
    "source_id", "source_system", "event_class", "form", "window", "window_seconds",
    "expected_min", "drop_below", "heartbeat", "reconciliation", "extensions",
}
_HEARTBEAT_KEYS = {"event_class", "interval", "interval_seconds", "assertion", "missing", "extensions"}
_RECONCILIATION_KEYS = {"compare_local", "compare_observed", "divergence", "undrained_local", "extensions"}


def duration_seconds(value: Any) -> int:
    """Seconds in a schema ``duration``: ISO-8601 ``PnDTnHnMnS`` or ``90s``/``5m``/``1h``/``1d``."""
    text = str(value)
    short = _SHORT.match(text)
    if short:
        return int(short.group(1)) * _UNIT[short.group(2)]
    iso = _ISO.match(text)
    if iso and text not in ("P", "PT"):
        days, hours, minutes, seconds = (int(g or 0) for g in iso.groups())
        total = days * 86400 + hours * 3600 + minutes * 60 + seconds
        if total > 0:
            return total
    raise ValueError(f"not a positive schema duration: {value!r}")


@dataclass(frozen=True)
class RateEntry:
    """One ``expected-rate`` entry: at least ``expected_min`` per ``window_seconds``."""

    source_id: str
    event_class: str
    window_seconds: int
    expected_min: int
    drop_below: str


@dataclass(frozen=True)
class CadenceDeclaration:
    """A source-owned declaration, validated against wire schema 0.1."""

    declaration_id: str
    source: str
    stream_id: str | None
    rates: tuple[RateEntry, ...]
    # heartbeat-or-reconciliation entries, kept verbatim: accepted, and
    # evaluated by T04/T06 rather than here.
    other_entries: tuple[dict, ...]
    document: dict

    def summary(self) -> dict[str, Any]:
        return {
            "declaration_id": self.declaration_id,
            "source": self.source,
            "stream_id": self.stream_id,
            "contract_digest": CONTRACT_DIGEST,
            "expected_rate_classes": {
                r.event_class: {"window_seconds": r.window_seconds, "expected_min": r.expected_min}
                for r in self.rates
            },
            "heartbeat_or_reconciliation_classes": sorted(
                e["event_class"] for e in self.other_entries
            ),
        }


def _text(obj: dict, key: str, where: str, *, required: bool = True) -> str | None:
    if key not in obj:
        if required:
            raise ValueError(f"{where}: {key} is required")
        return None
    value = obj[key]
    if not isinstance(value, str) or not value:
        raise ValueError(f"{where}: {key} must be a non-empty string")
    return value


def _closed(obj: Any, allowed: set[str], where: str) -> dict:
    if not isinstance(obj, dict):
        raise ValueError(f"{where} must be an object")
    extra = set(obj) - allowed
    if extra:
        raise ValueError(f"{where}: unknown properties {sorted(extra)}")
    return obj


def _one_of(obj: dict, a: str, b: str, where: str) -> int:
    if (a in obj) == (b in obj):
        raise ValueError(f"{where}: exactly one of {a} or {b} is required")
    if a in obj:
        return duration_seconds(obj[a])
    value = obj[b]
    if not isinstance(value, int) or isinstance(value, bool) or value < 1:
        raise ValueError(f"{where}: {b} must be a positive integer")
    return value


def parse_declaration(document: Any) -> CadenceDeclaration:
    """Validate a declaration against wire schema 0.1 and index it.

    Hand-checked rather than via a JSON Schema library, to keep the receiver
    stdlib-only. Refuses what the schema refuses; it does not relax a rule to
    admit a declaration, since a relaxed reading would be a competing shape.
    """
    doc = _closed(document, _TOP_KEYS, "emission cadence declaration")
    if doc.get("schema_version") != WIRE_SCHEMA_VERSION:
        raise ValueError(
            f"emission cadence declaration: schema_version must be "
            f"{WIRE_SCHEMA_VERSION!r}, got {doc.get('schema_version')!r}"
        )
    declaration_id = _text(doc, "declaration_id", "declaration")
    source = _text(doc, "source", "declaration")
    stream_id = _text(doc, "stream_id", "declaration", required=False)
    entries = doc.get("sources")
    if not isinstance(entries, list) or not entries:
        raise ValueError("declaration: sources must be a non-empty list")

    rates: list[RateEntry] = []
    others: list[dict] = []
    for index, raw in enumerate(entries):
        where = f"declaration sources[{index}]"
        entry = _closed(raw, _ENTRY_KEYS, where)
        source_id = _text(entry, "source_id", where)
        event_class = _text(entry, "event_class", where)
        _text(entry, "source_system", where, required=False)
        form = entry.get("form")
        if form not in FORMS:
            raise ValueError(f"{where}: form must be one of {FORMS}, got {form!r}")
        if form == EXPECTED_RATE:
            if "heartbeat" in entry or "reconciliation" in entry:
                raise ValueError(f"{where}: expected-rate carries no heartbeat or reconciliation")
            window = _one_of(entry, "window", "window_seconds", where)
            expected_min = entry.get("expected_min")
            if not isinstance(expected_min, int) or isinstance(expected_min, bool) or expected_min < 0:
                raise ValueError(f"{where}: expected_min must be a non-negative integer")
            drop_below = _text(entry, "drop_below", where)
            rates.append(RateEntry(source_id, event_class, window, expected_min, drop_below))
            continue
        forbidden = {"window", "window_seconds", "expected_min", "drop_below"} & set(entry)
        if forbidden:
            raise ValueError(f"{where}: heartbeat-or-reconciliation may not carry {sorted(forbidden)}")
        if "heartbeat" not in entry and "reconciliation" not in entry:
            raise ValueError(f"{where}: heartbeat-or-reconciliation needs a heartbeat or a reconciliation")
        if "heartbeat" in entry:
            beat = _closed(entry["heartbeat"], _HEARTBEAT_KEYS, f"{where}.heartbeat")
            for key in ("event_class", "assertion", "missing"):
                _text(beat, key, f"{where}.heartbeat")
            _one_of(beat, "interval", "interval_seconds", f"{where}.heartbeat")
        if "reconciliation" in entry:
            rec = _closed(entry["reconciliation"], _RECONCILIATION_KEYS, f"{where}.reconciliation")
            for key in ("compare_local", "compare_observed", "divergence"):
                _text(rec, key, f"{where}.reconciliation")
            _text(rec, "undrained_local", f"{where}.reconciliation", required=False)
        others.append(dict(entry))

    classes = [r.event_class for r in rates]
    if len(classes) != len(set(classes)):
        # Two rates for one class would each be satisfied by the other's
        # events; which one is meant is not ours to guess.
        raise ValueError("declaration: an event class has more than one expected-rate entry")
    return CadenceDeclaration(
        declaration_id=declaration_id, source=source, stream_id=stream_id,
        rates=tuple(rates), other_entries=tuple(others), document=dict(doc),
    )


@dataclass(frozen=True)
class CadenceFinding:
    """A stream observed below the rate its source declared."""

    sender: str
    source: str
    event_class: str
    declaration_id: str
    window_seconds: int
    expected_min: int
    observed: int
    since: str
    until: str

    def as_dict(self) -> dict[str, Any]:
        return {
            "kind": BELOW_CADENCE,
            "sender": self.sender,
            "source": self.source,
            "class": self.event_class,
            "declaration_id": self.declaration_id,
            "contract_digest": CONTRACT_DIGEST,
            "window_seconds": self.window_seconds,
            "expected_min": self.expected_min,
            "observed": self.observed,
            "since": self.since,
            "until": self.until,
            "means": (
                "audit-core holds fewer events of this class in the window than "
                "the source declared it would emit. The source may have stopped, "
                "or the declaration may be wrong; this is not proof of "
                "suppression, and a rate held would not prove completeness."
            ),
        }


def evaluate(
    identities: Iterable[Any],
    counter: Callable[[str, str, str], list[dict]],
    *,
    now: datetime | None = None,
) -> list[CadenceFinding]:
    """Findings for every declared expected-rate class observed below its rate.

    ``counter(source, since, until)`` is the backend's per-class count — the
    same numbers ``/v1/reconciliation`` returns, so a finding and a count
    cannot disagree. Each declared class is evaluated over the trailing window
    ending ``now``, per permitted source. A wildcard source is refused, as for
    heartbeats: there is no determinate stream to hold to a rate.
    """
    now = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
    until = now.isoformat()
    findings: list[CadenceFinding] = []
    for identity in identities:
        declaration = getattr(identity, "emission_cadence", None)
        if declaration is None or not declaration.rates:
            continue
        for source in sorted(identity.sources):
            if source == "*":
                continue
            cache: dict[int, dict[str, int]] = {}
            for rate in declaration.rates:
                if rate.window_seconds not in cache:
                    since = (now - timedelta(seconds=rate.window_seconds)).isoformat()
                    cache[rate.window_seconds] = {
                        row["class"]: int(row["count"]) for row in counter(source, since, until)
                    }
                observed = cache[rate.window_seconds].get(rate.event_class, 0)
                if observed < rate.expected_min:
                    findings.append(CadenceFinding(
                        sender=identity.name, source=source, event_class=rate.event_class,
                        declaration_id=declaration.declaration_id,
                        window_seconds=rate.window_seconds, expected_min=rate.expected_min,
                        observed=observed,
                        since=(now - timedelta(seconds=rate.window_seconds)).isoformat(),
                        until=until,
                    ))
    return findings