"""Branch projection rules — deterministic summaries over audit_events.

Each projection function takes a SQLite cursor and a high-water mark
(or ``None`` for "everything") and returns a :class:`BranchSummary`.
Projections must be:

1. **Deterministic** — same inputs produce byte-identical
   ``summary_blob``.
2. **Idempotent** — running twice on the same slice is a no-op.
3. **Stable across schema-compatible inserts** — adding a new
   ``providence_cache`` row that doesn't carry a violation must not
   shift existing motif counts.

Projection rule version is committed at the snapshot level via
:data:`PROJECTION_VERSION`. Bumping it is a new ``memory-vN`` and
invalidates prior snapshots' digests (by design — projection rule
changes are causal).
"""

from __future__ import annotations

import json
import sqlite3
from collections import Counter
from typing import Optional

from arborist.memory.canonical import BranchSummary


PROJECTION_VERSION = "projections-v1"


def _violations_text(body: str) -> list[str]:
    """Extract the flat list of violation tags from a v9.8 audit body.

    Multiple shapes coexist in the wild:
    - ``providence_write`` events with verifier_method/audit_mode
    - ``providence_repair`` events with pre/post audit_mode
    The verifier-violation tags surface in the in-process verdict and
    are persisted on the providence_cache row, but in the audit body
    they show up under various keys depending on the run-DAG node.
    Projection v1 is deliberately conservative: it reads
    ``violations`` arrays only, falling through to empty otherwise.
    """
    try:
        obj = json.loads(body)
    except (TypeError, json.JSONDecodeError):
        return []
    raw = obj.get("violations")
    if isinstance(raw, list):
        return [str(t) for t in raw if isinstance(t, str)]
    return []


def project_failure_motif_violations(
    conn: sqlite3.Connection,
    *,
    high_water: Optional[str] = None,
) -> tuple[BranchSummary, str]:
    """Count violation tags found in providence_cache rows.

    Returns ``(BranchSummary, latest_event_hash)``.

    We project over ``providence_cache`` (where verifier-resolved
    violations are persisted as JSON in the merkle_proof / row) plus
    the ``audit_events`` table for `providence_write` / `providence_repair`.
    Projection v1 reads two columns:
    - ``providence_cache.unverified_quotes`` — non-null indicates
      partial verification.
    - ``audit_events.body`` `violations` array on `providence_write`
      events.
    """
    counter: Counter[str] = Counter()

    # Slice over providence_write events past the high-water.
    if high_water is None:
        rows = conn.execute(
            "SELECT body FROM audit_events "
            "WHERE event_type = 'providence_write' "
            "ORDER BY seq"
        ).fetchall()
    else:
        cur = conn.execute(
            "SELECT seq FROM audit_events WHERE event_hash = ?",
            (high_water,),
        ).fetchone()
        if cur is None:
            rows = conn.execute(
                "SELECT body FROM audit_events "
                "WHERE event_type = 'providence_write' "
                "ORDER BY seq"
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT body FROM audit_events "
                "WHERE event_type = 'providence_write' AND seq > ? "
                "ORDER BY seq",
                (cur["seq"],),
            ).fetchall()

    for row in rows:
        for tag in _violations_text(row["body"]):
            counter[tag] += 1

    # Latest event_hash overall (regardless of branch).
    latest = conn.execute(
        "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
    ).fetchone()
    latest_hash = latest["event_hash"] if latest else "genesis"

    payload = {
        "projection_version": PROJECTION_VERSION,
        "tag_counts": {k: counter[k] for k in sorted(counter)},
    }
    blob = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8", errors="surrogatepass")

    return (
        BranchSummary(
            branch_id="failure-motif:violations",
            summary_blob=blob,
            count=sum(counter.values()),
        ),
        latest_hash,
    )


def project_audit_mode_distribution(
    conn: sqlite3.Connection,
    *,
    high_water: Optional[str] = None,
) -> BranchSummary:
    """Count audit_mode outcomes (STRICT/HYBRID/UNGROUNDED) on writes.

    Pure read — does not advance high-water; that's the caller's job.
    """
    counter: Counter[str] = Counter()

    if high_water is None:
        rows = conn.execute(
            "SELECT body FROM audit_events "
            "WHERE event_type = 'providence_write' "
            "ORDER BY seq"
        ).fetchall()
    else:
        cur = conn.execute(
            "SELECT seq FROM audit_events WHERE event_hash = ?",
            (high_water,),
        ).fetchone()
        cutoff = cur["seq"] if cur else -1
        rows = conn.execute(
            "SELECT body FROM audit_events "
            "WHERE event_type = 'providence_write' AND seq > ? "
            "ORDER BY seq",
            (cutoff,),
        ).fetchall()

    for row in rows:
        try:
            obj = json.loads(row["body"])
        except (TypeError, json.JSONDecodeError):
            continue
        mode = obj.get("audit_mode")
        if isinstance(mode, str):
            counter[mode] += 1

    payload = {
        "projection_version": PROJECTION_VERSION,
        "audit_mode_counts": {k: counter[k] for k in sorted(counter)},
    }
    blob = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8", errors="surrogatepass")
    return BranchSummary(
        branch_id="audit-mode-distribution",
        summary_blob=blob,
        count=sum(counter.values()),
    )


def project_falsification_counts(
    conn: sqlite3.Connection,
    *,
    high_water: Optional[str] = None,
) -> BranchSummary:
    """Count falsification states across providence_cache rows.

    Reflects current state, not historical — useful for snapshot-time
    triage. The high_water arg is accepted but ignored for this branch
    (state is a current value, not a temporal accumulation).
    """
    rows = conn.execute(
        "SELECT falsification_state, COUNT(*) AS n FROM providence_cache "
        "GROUP BY falsification_state"
    ).fetchall()
    counts = {r["falsification_state"]: int(r["n"]) for r in rows}
    payload = {
        "projection_version": PROJECTION_VERSION,
        "falsification_counts": {k: counts[k] for k in sorted(counts)},
    }
    blob = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8", errors="surrogatepass")
    return BranchSummary(
        branch_id="falsification-state",
        summary_blob=blob,
        count=sum(counts.values()),
    )


# Default branch set the snapshotter calls.
DEFAULT_BRANCHES = (
    project_failure_motif_violations,
    project_audit_mode_distribution,
    project_falsification_counts,
)
