Logging EAS Interrupts for As-Run Reconciliation

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timezone
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, field_validator

# Traffic-ops structured logging: timestamp | level | module | spot_id
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.eas.ledger")

GENESIS_HASH = "0" * 64          # prev_hash of the very first entry in a chain


class ReconStatus(str, Enum):
    CORROBORATED = "corroborated"     # as-run confirms the displacement
    MISSING_AS_RUN = "missing_as_run" # ledger says displaced, as-run silent
    UNEXPECTED_AIR = "unexpected_air" # as-run says aired despite displacement


def _require_utc(v: datetime) -> datetime:
    if v.tzinfo is None:
        raise ValueError("timestamp must be timezone-aware (UTC)")
    return v.astimezone(timezone.utc)


class InterruptLogEntry(BaseModel):
    # frozen=True makes an instance hashable and read-only after construction.
    model_config = ConfigDict(frozen=True)

    entry_id: str                    # Deterministic hash of (activation_id, spot_id)
    activation_id: str
    event_code: str                  # EAN, TOR, SVR, RWT, ...
    spot_id: str
    house_number: str
    isci: str
    overlap_ms: int = Field(gt=0)
    disposition: str
    makegood_eligible: bool
    interrupted_at: datetime         # window start that displaced this spot
    recorded_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    prev_hash: str = GENESIS_HASH    # links this entry to the one before it
    content_hash: str = ""           # filled in at Step 2; excluded from itself

    _v_int = field_validator("interrupted_at")(classmethod(lambda cls, v: _require_utc(v)))

def seal_entry(entry: InterruptLogEntry, prev_hash: str) -> InterruptLogEntry:
    # Serialize the immutable fields deterministically (sorted keys), excluding
    # the hash field itself, then chain onto the previous entry's hash.
    payload = entry.model_dump(exclude={"content_hash", "prev_hash"}, mode="json")
    material = json.dumps(payload, sort_keys=True, separators=(",", ":")) + prev_hash
    digest = hashlib.sha256(material.encode()).hexdigest()
    sealed = entry.model_copy(update={"prev_hash": prev_hash, "content_hash": digest})
    logger.info(
        "%s | sealed ledger entry | activation=%s hash=%s",
        sealed.spot_id, sealed.activation_id, digest[:12],
    )
    return sealed

class ImmutableLedger:
    """Append-only, hash-chained store of EAS interrupt log entries."""

    def __init__(self) -> None:
        self._entries: list[InterruptLogEntry] = []
        self._seen: set[str] = set()

    @property
    def head_hash(self) -> str:
        return self._entries[-1].content_hash if self._entries else GENESIS_HASH

    def append(self, entry: InterruptLogEntry) -> InterruptLogEntry:
        if entry.entry_id in self._seen:
            # Idempotent: a replayed signal is a no-op, not a second record.
            logger.info("%s | duplicate entry_id, skipped | %s",
                        entry.spot_id, entry.entry_id)
            return next(e for e in self._entries if e.entry_id == entry.entry_id)
        sealed = seal_entry(entry, self.head_hash)
        self._entries.append(sealed)
        self._seen.add(sealed.entry_id)
        return sealed

    def verify_chain(self) -> bool:
        # Re-derive every hash; any tampered entry breaks the chain from there on.
        prev = GENESIS_HASH
        for e in self._entries:
            expected = seal_entry(e, prev).content_hash
            if expected != e.content_hash:
                logger.error("%s | chain broken at entry %s", e.spot_id, e.entry_id)
                return False
            prev = e.content_hash
        return True

class AsRunRecord(BaseModel):
    spot_id: str
    aired_at: datetime | None        # None if the spot did not air
    aired_ms: int = Field(ge=0)      # duration actually broadcast

    _v_air = field_validator("aired_at")(
        classmethod(lambda cls, v: _require_utc(v) if v is not None else None)
    )


def reconcile(
    ledger: list[InterruptLogEntry], as_run: dict[str, AsRunRecord]
) -> dict[str, ReconStatus]:
    results: dict[str, ReconStatus] = {}
    for entry in ledger:
        record = as_run.get(entry.spot_id)
        if record is None or record.aired_ms == 0:
            # Ledger says displaced and as-run shows no airing → they agree.
            status = ReconStatus.CORROBORATED
        elif entry.disposition == "fully_displaced" and record.aired_ms > 0:
            # Ledger says fully displaced but as-run shows it aired → conflict.
            status = ReconStatus.UNEXPECTED_AIR
        else:
            # Partial displacement with a partial airing is internally consistent.
            status = ReconStatus.CORROBORATED
        results[entry.spot_id] = status
        level = logging.WARNING if status is ReconStatus.UNEXPECTED_AIR else logging.INFO
        logger.log(level, "%s | reconcile %s | activation=%s",
                   entry.spot_id, status.value, entry.activation_id)
    return results

def emit_discrepancies(
    ledger: list[InterruptLogEntry], statuses: dict[str, ReconStatus]
) -> list[dict[str, str]]:
    discrepancies: list[dict[str, str]] = []
    for entry in ledger:
        status = statuses[entry.spot_id]
        if status is ReconStatus.CORROBORATED:
            continue
        discrepancies.append({
            "entry_id": entry.entry_id,
            "spot_id": entry.spot_id,
            "house_number": entry.house_number,
            "activation_id": entry.activation_id,
            "status": status.value,
        })
        logger.warning("%s | discrepancy queued for review | %s",
                       entry.spot_id, status.value)
    return discrepancies

led = ImmutableLedger()
base = dict(activation_id="EAS-2026-0714-TOR-0007", event_code="TOR",
            house_number="HN-1", isci="AAAA1111", overlap_ms=30000,
            makegood_eligible=True,
            interrupted_at=datetime(2026, 7, 14, 21, 3, 0, tzinfo=timezone.utc))

e1 = led.append(InterruptLogEntry(entry_id="k1", spot_id="SP-A",
                disposition="fully_displaced", **base))
e2 = led.append(InterruptLogEntry(entry_id="k2", spot_id="SP-B",
                disposition="fully_displaced", **base))

# 1. Chaining: entry two links to entry one, and the chain verifies.
assert e2.prev_hash == e1.content_hash
assert led.verify_chain() is True

# 2. Idempotency: re-appending an existing entry_id is a no-op.
dup = led.append(InterruptLogEntry(entry_id="k1", spot_id="SP-A",
                 disposition="fully_displaced", **base))
assert dup.content_hash == e1.content_hash
assert len(led._entries) == 2

# 3. Reconciliation: SP-A did not air (corroborated); SP-B aired (conflict).
as_run = {
    "SP-A": AsRunRecord(spot_id="SP-A", aired_at=None, aired_ms=0),
    "SP-B": AsRunRecord(spot_id="SP-B",
                        aired_at=datetime(2026, 7, 14, 21, 3, 30, tzinfo=timezone.utc),
                        aired_ms=30000),
}
statuses = reconcile(led._entries, as_run)
assert statuses["SP-A"] is ReconStatus.CORROBORATED
assert statuses["SP-B"] is ReconStatus.UNEXPECTED_AIR
assert len(emit_discrepancies(led._entries, statuses)) == 1