"""Export and replay by natural key (spec §8): the review on staging reaches production, and
production's decisions survive a staging refresh, by uid and natural key, never by id. The
export is JSON lines in `(decided_at, id)` order carrying no local id; replay resolves every
natural key on the target, matches rows by uid, keeps the source `decided_at` (an older row
over a newer target row is inserted and reported, never materialised), parks what it cannot
resolve in a remainder file, and applies through the same writer and appliers.

Sources of truth: this module, `writer.py`, `natural_keys.py`, `app/cli_replay.py`,
`scripts/staging-refresh.py`, `tests/test_replay.py`. What it cost before: `propose --file` carried
staging's row ids and could not load on production (plan W13); a refresh discarded every
decision taken on staging (the 15 Sep decisions §6).
"""

from __future__ import annotations

import json
import pathlib
import uuid as _uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import Decision, DecisionBatch, ProposalPass
from app.services.decisions import natural_keys
from app.services.decisions.effective import effective
from app.services.decisions.writer import Refused, batch as open_batch, finish, record, refuse_if_collecting

PARK_CODES = ("LINE_MISSING", "LINE_MISMATCH", "VARIANT_SPLIT", "LISTING_MISSING", "BRAND_MISSING", "PLACE_MISSING",
              "WORDING_MISSING", "PAIR_MISSING", "PAIR_SIDE_MISSING", "REVERSAL_TARGET_MISSING", "DECISION_HELD", "WOULD_MINT",
              "ENTITY_ALIASED", "PRODUCT_MERGED", "PAIR_SIDE_MERGED", "UNMERGE_CHAIN", "VALUE_INVALID", "FIELD_UNKNOWN",
              "LINE_HOUSES_DIFFER", "ENTITY_NOT_FOUND", "KEY_TYPE_MISMATCH")


def _iso(value: datetime | None) -> str | None:
    return value.isoformat() if value is not None else None


# --------------------------------------------------------------------------- export

def export(db: Session, *, since: datetime | None = None, batch_uid: str | None = None, pass_name: str | None = None,
           all_: bool = False, origin_host: str | None = None, out) -> dict[str, int]:
    """JSON lines: the passes referenced, the batches, every decision with `caused_by_id IS
    NULL` (consequences are derived and regenerate on the target with the same uuid5 uid), in
    `(decided_at, id)` order, no local id anywhere. `--batch` also exports every row that
    reverses or supersedes a row of the batch, or refuses `BATCH_UNDONE`."""
    stmt = select(Decision).where(Decision.caused_by_id.is_(None))
    if batch_uid:
        target = db.scalar(select(DecisionBatch).where(DecisionBatch.uid == _uuid.UUID(str(batch_uid))))
        if target is None:
            raise Refused("REVERSAL_TARGET_MISSING", f"No batch {batch_uid}.")
        undone = db.scalar(select(DecisionBatch).where(DecisionBatch.reverses_batch_id == target.id))
        if undone is not None:
            raise Refused("BATCH_UNDONE", f"Batch {batch_uid} was undone by batch {undone.uid}; export the undo too with --since.")
        own = select(Decision.id).where(Decision.batch_id == target.id)
        stmt = stmt.where((Decision.batch_id == target.id) | Decision.reverses_id.in_(own) | Decision.supersedes_id.in_(own))
    elif pass_name:
        p = db.scalar(select(ProposalPass).where(ProposalPass.name == pass_name))
        if p is None:
            raise Refused("REVERSAL_TARGET_MISSING", f"No pass {pass_name}.")
        stmt = stmt.where(Decision.pass_id == p.id)
    elif since is not None:
        stmt = stmt.where(Decision.decided_at >= since)
    elif not all_:
        raise Refused("VALUE_INVALID", "export needs --since, --batch, --pass or --all")
    if origin_host:
        stmt = stmt.where(Decision.origin_host == origin_host)
    rows = list(db.scalars(stmt.order_by(Decision.decided_at.asc(), Decision.id.asc())))
    batches = {b.id: b for b in db.scalars(select(DecisionBatch).where(DecisionBatch.id.in_({r.batch_id for r in rows})))} if rows else {}
    passes = {p.id: p for p in db.scalars(select(ProposalPass).where(ProposalPass.id.in_({r.pass_id for r in rows if r.pass_id})))} if rows else {}
    by_id = {r.id: r for r in rows}
    counts = {"passes": 0, "batches": 0, "decisions": 0}
    for p in passes.values():
        out.write(json.dumps({"kind": "pass", "name": p.name, "pass_kind": p.kind, "process_version": p.process_version,
                              "rules_version": p.rules_version, "generator": p.generator, "scope_brand_slug": p.scope_brand_slug,
                              "note": p.note}) + "\n")
        counts["passes"] += 1
    for b in sorted(batches.values(), key=lambda x: (x.opened_at, x.id)):
        reverses = batches.get(b.reverses_batch_id) if b.reverses_batch_id else None
        if reverses is None and b.reverses_batch_id:
            reverses = db.get(DecisionBatch, b.reverses_batch_id)
        out.write(json.dumps({"kind": "batch", "uid": str(b.uid), "batch_kind": b.kind, "mode": b.mode, "by_username": b.by_username,
                              "opened_at": _iso(b.opened_at), "origin_host": b.origin_host, "scope": b.scope, "note": b.note,
                              "pass_name": passes[b.pass_id].name if b.pass_id in passes else None,
                              "reverses_batch_uid": str(reverses.uid) if reverses is not None else None}) + "\n")
        counts["batches"] += 1
    for r in rows:
        line: dict[str, Any] = {
            "kind": "decision", "uid": str(r.uid), "decided_at": _iso(r.decided_at), "origin_host": r.origin_host,
            "entity_type": r.entity_type, "natural_key": r.natural_key, "natural_key_detail": r.natural_key_detail,
            "field": r.field, "effect": r.effect, "value": r.value, "rule_value": r.rule_value, "prior_value": r.prior_value,
            "origin": r.origin, "mode": r.mode, "reason": r.reason, "rules_version": r.rules_version,
            "decided_by_username": r.decided_by_username, "batch_uid": str(batches[r.batch_id].uid) if r.batch_id in batches else None,
            "pass_name": passes[r.pass_id].name if r.pass_id in passes else None,
            "detail": {k: v for k, v in (r.detail or {}).items() if k not in ("conflict",)} or None,
        }
        if r.reverses_id:
            reversed_row = by_id.get(r.reverses_id) or db.get(Decision, r.reverses_id)
            line["reverses_uid"] = str(reversed_row.uid)
            line["reversed"] = {"entity_type": reversed_row.entity_type, "natural_key": reversed_row.natural_key,
                                "field": reversed_row.field, "value": reversed_row.value}
        out.write(json.dumps(line) + "\n")
        counts["decisions"] += 1
    return counts


# --------------------------------------------------------------------------- replay

def _read(path: pathlib.Path) -> list[dict]:
    rows = []
    with path.open("r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if line:
                rows.append(json.loads(line))
    return rows


def _pass(db: Session, spec: dict) -> ProposalPass:
    row = db.scalar(select(ProposalPass).where(ProposalPass.name == spec["name"]))
    if row is None:
        row = ProposalPass(name=spec["name"], kind=spec.get("pass_kind") or "session", process_version=spec.get("process_version") or "1",
                           rules_version=spec.get("rules_version") or "0", generator=spec.get("generator") or "replay",
                           scope_brand_slug=spec.get("scope_brand_slug"), note=spec.get("note"))
        db.add(row)
        db.flush()
    return row


def _reversal_target(db: Session, line: dict, entity, entity_type: str) -> Decision | None:
    if line.get("reverses_uid"):
        row = db.scalar(select(Decision).where(Decision.uid == _uuid.UUID(line["reverses_uid"])))
        if row is not None:
            return row
    reversed_spec = line.get("reversed") or {}
    now = effective(db, entity_type, [entity.id], [line["field"]]).get((entity.id, line["field"]))
    if now is not None and now.value == reversed_spec.get("value"):
        return now
    return None


def _mint_line(db: Session, line: dict, *, check: bool):
    """A `name` decision on `line:new:<slug>` mints the line on the target when no row holds the
    slug (spec §2, §6.3): brand from the detail, key `decided:<slug>`."""
    from app.models import Brand, ProductLine

    detail = line.get("natural_key_detail") or {}
    kind, parts = natural_keys.parse(line["natural_key"])
    slug = parts.get("slug") or detail.get("slug")
    brand = db.scalar(select(Brand).where(Brand.slug == detail.get("brand_slug") or ""))
    if brand is None or not slug or line["field"] != "name":
        return natural_keys.Parked("LINE_MISSING", detail=line["natural_key"])
    if check:
        return natural_keys.Parked("WOULD_MINT", detail=slug)
    row = ProductLine(brand_id=brand.id, key=f"decided:{slug}", slug=slug, name=str(line["value"])[:200])
    db.add(row)
    db.flush()
    return row


def replay(db: Session, file, *, check: bool = False, label: str | None = None, force: bool = False) -> dict[str, Any]:
    """Apply an export on this host (spec §8): per source batch one `decision_batches` row with the
    SAME uid (kind `replay`; present already -> reused); per row in file order, in a savepoint:
    uid present -> `already`; the entity and every value resolved by natural key; the reverser by
    uid, else by the effective row whose value equals the recorded one; the writer's own checks
    (`DECISION_HELD` parks); an older row than the target's effective row is inserted, not
    materialised, and reported `superseded`; a parked row leaves nothing behind and goes to the
    remainder file. One rekey and fold per batch at its tail; one commit at the end. `--check`
    rolls everything back and prints the counts."""
    refuse_if_collecting(db, force)
    path = pathlib.Path(file)
    remainder_path = path.with_suffix(path.suffix + ".remainder.jsonl") if path.suffix else path.with_name(path.name + ".remainder.jsonl")
    rows = _read(path)
    retry = [r for r in _read(remainder_path)] if remainder_path.exists() else []
    stamp = datetime.now(UTC).strftime("%Y-%m-%d-%H%M")
    replayed_from = f"{label or 'export'}@{stamp}"
    counts = {"passes": 0, "batches": 0, "applied": 0, "already": 0, "superseded": 0, "held": 0, "parked": 0, "consequences": 0}
    by_type: dict[str, dict[str, int]] = {}
    parked: list[dict] = []
    passes: dict[str, ProposalPass] = {}
    batches: dict[str, Any] = {}
    batch_lines = {r["uid"]: r for r in rows if r.get("kind") == "batch"}
    for spec in rows:
        if spec.get("kind") == "pass":
            passes[spec["name"]] = _pass(db, spec)
            counts["passes"] += 1
    retry_uids = {r["uid"] for r in retry if r.get("kind") == "decision"}
    decisions = [r for r in retry if r.get("kind") == "decision"] + [r for r in rows if r.get("kind") == "decision" and r["uid"] not in retry_uids]

    def batch_for(uid: str):
        if uid in batches:
            return batches[uid]
        spec = batch_lines.get(uid, {})
        existing = db.scalar(select(DecisionBatch).where(DecisionBatch.uid == _uuid.UUID(uid)))
        if existing is not None:
            from app.services.decisions.writer import Batch
            from app.services import keying

            b = Batch(db=db, row=existing, kind=existing.kind, mode=existing.mode, by_id=existing.by_account_id,
                      by_username=existing.by_username, pass_id=existing.pass_id, maps=keying.load_maps(db), tail=True, commit=False)
            batches[uid] = b
            return b
        reverses_id = None
        if spec.get("reverses_batch_uid"):
            rb = db.scalar(select(DecisionBatch).where(DecisionBatch.uid == _uuid.UUID(spec["reverses_batch_uid"])))
            reverses_id = rb.id if rb is not None else None
        kind = "undo" if reverses_id is not None else "replay"
        ctx = open_batch(db, kind, spec.get("mode") or "individual", spec.get("by_username"), scope={"source_batch_uid": uid, "file": path.name},
                         pass_id=passes[spec["pass_name"]].id if spec.get("pass_name") in passes else None, note=spec.get("note"),
                         commit=False, tail=True, reverses_batch_id=reverses_id, uid=_uuid.UUID(uid), replayed_from=replayed_from)
        b = ctx.__enter__()
        b._ctx = ctx  # closed at the end, in order
        batches[uid] = b
        counts["batches"] += 1
        return b

    for line in decisions:
        et = line["entity_type"]
        per = by_type.setdefault(et, {"applied": 0, "already": 0, "superseded": 0, "held": 0, "parked": 0})
        if db.scalar(select(Decision.id).where(Decision.uid == _uuid.UUID(line["uid"]))) is not None:
            counts["already"] += 1; per["already"] += 1
            continue
        b = batch_for(line["batch_uid"]) if line.get("batch_uid") else batch_for(str(_uuid.uuid5(_uuid.NAMESPACE_URL, "replay:" + path.name)))
        savepoint = db.begin_nested()
        try:
            entity = natural_keys.resolve(db, et, line["natural_key"], line.get("natural_key_detail"), mint=not check)
            if isinstance(entity, natural_keys.Parked) and et == "product_line" and line["natural_key"].startswith("line:new:"):
                entity = _mint_line(db, line, check=check)
            if isinstance(entity, natural_keys.Parked):
                raise Refused(entity.code, entity.detail or entity.code)
            reverses = None
            if line.get("reverses_uid") or line.get("reversed"):
                reverses = _reversal_target(db, line, entity, et)
                if reverses is None:
                    raise Refused("REVERSAL_TARGET_MISSING", f"nothing to reverse for {line['natural_key']} {line['field']}")
            value = line.get("value")
            if line["effect"] == "release":
                value = None
            row = record(b, et, entity, line["field"], value, origin=line.get("origin") or "person", reason=line.get("reason"),
                         decided_at=datetime.fromisoformat(line["decided_at"]), effect=line["effect"], mode=line.get("mode"),
                         decided_by=line.get("decided_by_username"), uid=_uuid.UUID(line["uid"]), replayed_from=replayed_from,
                         reverses=reverses, detail={**(line.get("detail") or {}), "replayed_from_values": {"rule_value": line.get("rule_value"), "prior_value": line.get("prior_value")}})
            savepoint.commit()
            if (row.detail or {}).get("conflict"):
                counts["superseded"] += 1; per["superseded"] += 1
            else:
                counts["applied"] += 1; per["applied"] += 1
        except Refused as exc:
            savepoint.rollback()
            code = exc.code if exc.code in PARK_CODES else "REFUSED"
            key = "held" if exc.code == "DECISION_HELD" else "parked"
            counts[key] += 1; per[key] += 1
            parked.append({**line, "park_code": code, "park_detail": exc.summary})
    for b in batches.values():
        ctx = getattr(b, "_ctx", None)
        if ctx is not None:
            ctx.__exit__(None, None, None)
        else:
            finish(b, close=False)  # reused: closed on its first replay; the tail still rekeys what this run touched
    counts["consequences"] = sum(len([d for d in b.decisions if d.caused_by_id is not None]) for b in batches.values())
    counts["by_type"] = by_type
    if check:
        db.rollback()
        counts["check"] = True
    else:
        db.commit()
        if parked:
            with remainder_path.open("w", encoding="utf-8") as fh:
                for item in parked:
                    fh.write(json.dumps(item) + "\n")
        elif remainder_path.exists():
            remainder_path.unlink()
    counts["remainder"] = str(remainder_path) if parked else None
    counts["parked_rows"] = [{"uid": p["uid"], "code": p["park_code"], "why": p["park_detail"]} for p in parked[:50]]
    return counts