#!/usr/bin/env python3
"""Executable delivery/retry/replay failure matrix (AUDIT-WP-0005-T05).

The controlling assertion is one sentence: **one source outbox event produces
exactly one durable normalized event**, across retries, replay, and
infrastructure disruption. Every scenario below either supports that claim or
fails loudly.

Modes:
    MODE=local   (default) start PostgreSQL and the receiver locally in Docker
                 and run the full matrix, including infrastructure disruption
    MODE=remote  run against an already-deployed receiver at BASE_URL; the
                 disruption scenarios are skipped unless DISRUPT=1, because
                 restarting a production database is not this script's call

Local mode exists so the matrix is a rehearsed script by the time it runs
against Railiance, rather than something improvised during a deployment.

Evidence is written to evidence/failure-matrix-<timestamp>.json and contains no
secrets — it is intended to be handed back to NK-WP-0024.
"""

from __future__ import annotations

import json
import os
import pathlib
import subprocess
import sys
import time
import urllib.error
import urllib.request
import uuid
from dataclasses import dataclass, field

REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))

MODE = os.environ.get("MODE", "local")
BASE_URL = os.environ.get("BASE_URL", "http://127.0.0.1:8091")
DISRUPT = os.environ.get("DISRUPT", "1" if MODE == "local" else "0") == "1"

PG_CONTAINER = os.environ.get("FM_PG_CONTAINER", "fm-audit-pg")
PG_PORT = int(os.environ.get("FM_PG_PORT", "55460"))
APP_PORT = int(os.environ.get("FM_APP_PORT", "8091"))

SENDER_TOKEN = os.environ.get("FM_SENDER_TOKEN", "sender-current")
SENDER_NEXT = os.environ.get("FM_SENDER_NEXT", "sender-next")
READER_TOKEN = os.environ.get("FM_READER_TOKEN", "reader-token")
TENANT = "tenant:friendly:binky"

# Retry policy modelled on a bounded transactional-outbox sender. Which codes
# are retried is not arbitrary — it is the response contract in
# audit_core/ingestion.py, and getting it wrong is how senders either lose
# events or duplicate them.
RETRYABLE = {503, 500}
TERMINAL = {400, 401, 403, 409}


@dataclass
class Result:
    name: str
    ok: bool
    detail: str = ""
    skipped: bool = False


@dataclass
class Matrix:
    results: list[Result] = field(default_factory=list)

    def record(self, name: str, ok: bool, detail: str = "", skipped: bool = False):
        self.results.append(Result(name, ok, detail, skipped))
        mark = "SKIP" if skipped else ("PASS" if ok else "FAIL")
        colour = {"PASS": "\033[32m", "FAIL": "\033[31m", "SKIP": "\033[33m"}[mark]
        print(f"  {colour}{mark}\033[0m  {name}" + (f" — {detail}" if detail else ""))
        return ok

    @property
    def failed(self) -> list[Result]:
        return [r for r in self.results if not r.ok and not r.skipped]


# --- transport --------------------------------------------------------------

def request(method: str, path: str, token: str | None = None,
            body: dict | None = None, idempotency_key: str | None = None,
            timeout: float = 10.0) -> tuple[int, dict]:
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(f"{BASE_URL}{path}", data=data, method=method)
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    if idempotency_key:
        req.add_header("Idempotency-Key", idempotency_key)
    if data:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as response:
            return response.status, json.loads(response.read() or b"{}")
    except urllib.error.HTTPError as exc:
        raw = exc.read() or b"{}"
        try:
            return exc.code, json.loads(raw)
        except ValueError:
            return exc.code, {"raw": raw.decode("utf-8", "replace")}
    except (urllib.error.URLError, TimeoutError, ConnectionError) as exc:
        # No response at all — indistinguishable to a sender from a timeout,
        # and therefore retryable.
        return 0, {"error": f"unreachable: {exc}"}


def event(event_id: str, *, tenant: str = TENANT, correlation: str = "corr-matrix",
          data: dict | None = None, subject: str = "membership-1") -> dict:
    return {
        "id": event_id, "type": "membership.added", "source": "user-engine",
        "subject": subject, "tenant": tenant, "correlation_id": correlation,
        "occurred_at": "2026-08-09T00:00:00Z", "data": data or {"membership_id": "m-1"},
    }


def deliver(payload: dict, token: str = SENDER_TOKEN, *, attempts: int = 8,
            backoff: float = 0.5) -> tuple[int, dict, int]:
    """Deliver like a bounded outbox sender. Returns (status, body, tries)."""
    tries = 0
    status, body = 0, {}
    for attempt in range(attempts):
        tries += 1
        status, body = request("POST", "/v1/events", token, payload, payload["id"])
        if status in TERMINAL or status in (200, 202):
            return status, body, tries
        if status in RETRYABLE or status == 0:
            time.sleep(backoff * (attempt + 1))
            continue
        return status, body, tries
    return status, body, tries


def stored_count(event_id: str) -> int:
    """Count custody records for an event id, read straight from the database.

    Deliberately not via the API: the assertion is about what is *stored*, and
    asking the service under test to vouch for itself is weaker evidence.
    """
    if MODE != "local":
        status, body = request("GET", f"/v1/events/{event_id}", READER_TOKEN)
        return 1 if status == 200 else 0
    out = subprocess.run(
        ["docker", "exec", PG_CONTAINER, "psql", "-At", "-U", "postgres",
         "-d", "audit_core", "-c",
         f"SELECT count(*) FROM audit_core.events WHERE event_id = '{event_id}'"],
        capture_output=True, text=True,
    )
    try:
        return int(out.stdout.strip())
    except ValueError:
        return -1


# --- local stack ------------------------------------------------------------

REPO = pathlib.Path(__file__).resolve().parent.parent


def start_stack() -> subprocess.Popen:
    subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
    subprocess.run(
        ["docker", "run", "-d", "--name", PG_CONTAINER,
         "-e", "POSTGRES_PASSWORD=matrix", "-e", "POSTGRES_DB=audit_core",
         "-p", f"127.0.0.1:{PG_PORT}:5432", "postgres:16-alpine"],
        check=True, capture_output=True,
    )
    wait_for_postgres()

    senders = json.dumps([
        {"name": "user-engine", "tokens": [SENDER_TOKEN, SENDER_NEXT],
         "sources": ["user-engine"], "tenants": [TENANT]},
        {"name": "ops-reader", "tokens": [READER_TOKEN], "sources": ["*"],
         "tenants": ["*"], "may_write": False, "may_read": True},
    ])
    env = {
        **os.environ,
        "AUDIT_CORE_DATABASE_URL":
            f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core",
        "AUDIT_CORE_SENDERS": senders,
        "AUDIT_CORE_HOST": "127.0.0.1",
        "AUDIT_CORE_HTTP_PORT": str(APP_PORT),
        "AUDIT_CORE_REQUIRE_CUSTODY_CLASS": "operational",
        "AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS": "5000",
        "AUDIT_CORE_LOG_LEVEL": "WARNING",
    }
    python = REPO / ".venv" / "bin" / "python"
    app = subprocess.Popen(
        [str(python if python.exists() else sys.executable), "-m", "audit_core.ingestion"],
        cwd=REPO, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    wait_for_app()
    return app


def wait_for_postgres(timeout: int = 60) -> None:
    for _ in range(timeout):
        probe = subprocess.run(
            ["docker", "exec", PG_CONTAINER, "pg_isready", "-U", "postgres"],
            capture_output=True,
        )
        if probe.returncode == 0:
            return
        time.sleep(1)
    raise SystemExit("postgres did not become ready")


def wait_for_app(timeout: int = 40) -> bool:
    for _ in range(timeout):
        status, _ = request("GET", "/healthz", timeout=2)
        if status == 200:
            return True
        time.sleep(1)
    return False


def stop_stack(app: subprocess.Popen | None) -> None:
    if app and app.poll() is None:
        app.terminate()
        try:
            app.wait(timeout=15)
        except subprocess.TimeoutExpired:
            app.kill()
    subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)


# --- the matrix -------------------------------------------------------------

def run(matrix: Matrix, app: subprocess.Popen | None) -> None:
    print("\n== delivery ==")
    e1 = f"fm-deliver-{uuid.uuid4().hex[:8]}"
    status, body, tries = deliver(event(e1))
    matrix.record("S01 successful delivery is accepted once",
                  status == 202 and body.get("status") == "accepted" and stored_count(e1) == 1,
                  f"status={status} tries={tries} stored={stored_count(e1)}")

    status, body, _ = deliver(event(e1))
    matrix.record("S02 resubmission reconciles as duplicate, no second record",
                  status == 200 and body.get("status") == "duplicate" and stored_count(e1) == 1,
                  f"status={status} stored={stored_count(e1)}")

    status, _, _ = deliver(event(e1, subject="changed"))
    matrix.record("S03 same id, different payload is a conflict",
                  status == 409 and stored_count(e1) == 1, f"status={status}")

    print("\n== rejection and dead-letter visibility ==")
    e2 = f"fm-tenant-{uuid.uuid4().hex[:8]}"
    status, body, tries = deliver(event(e2, tenant="tenant:coulomb"))
    matrix.record("S04 cross-tenant claim refused, terminal for the sender",
                  status == 400 and body.get("error") == "tenant_not_allowed" and tries == 1,
                  f"status={status} tries={tries}")

    status, dead = request("GET", "/v1/dead-letters", READER_TOKEN)
    visible = any(d.get("event_id") == e2 for d in dead.get("dead_letters", []))
    matrix.record("S05 rejected event is visible as a dead letter", status == 200 and visible)

    status, _ = request("POST", "/v1/events", "not-a-token", event("fm-x"), "fm-x")
    matrix.record("S06 bad credential is rejected and not retried", status == 401,
                  f"status={status}")

    print("\n== redaction ==")
    e3 = f"fm-redact-{uuid.uuid4().hex[:8]}"
    status, _, _ = deliver(event(e3, data={"membership_id": "m-1", "auth_token": "s3cret"}))
    _, record = request("GET", f"/v1/events/{e3}", READER_TOKEN)
    data = record.get("details", {}).get("data", {})
    redaction = record.get("details", {}).get("redaction", {})
    matrix.record("S07 secret-shaped field is redacted, event still stored",
                  status == 202 and data.get("auth_token") == "[redacted]"
                  and data.get("membership_id") == "m-1"
                  and redaction.get("paths") == ["data.auth_token"],
                  f"stored auth_token={data.get('auth_token')!r}")

    _, findings = request("GET", "/v1/secret-findings", READER_TOKEN)
    counted = any(f["field_path"] == "data.auth_token" and f["outcome"] == "redacted"
                  for f in findings.get("secret_findings", []))
    matrix.record("S08 redaction is counted by field path", counted)

    print("\n== correlation lookup ==")
    correlation = f"corr-{uuid.uuid4().hex[:8]}"
    ids = [f"fm-corr-{i}-{uuid.uuid4().hex[:6]}" for i in range(3)]
    for event_id in ids:
        deliver(event(event_id, correlation=correlation))
    status, found = request("GET", f"/v1/events?correlation_id={correlation}", READER_TOKEN)
    matrix.record("S09 correlation lookup returns every related event",
                  status == 200 and {e["event_id"] for e in found.get("events", [])} == set(ids),
                  f"found={len(found.get('events', []))}/3")

    print("\n== privilege separation ==")
    status, _ = request("GET", f"/v1/events/{e1}", SENDER_TOKEN)
    matrix.record("S10 sender credential cannot read the trail back", status == 403,
                  f"status={status}")
    status, _ = request("POST", "/v1/events", READER_TOKEN, event("fm-ro"), "fm-ro")
    matrix.record("S11 reader credential cannot write", status == 403, f"status={status}")

    print("\n== credential rotation during active ingestion ==")
    rotated = []
    for i in range(3):
        token = SENDER_TOKEN if i == 0 else SENDER_NEXT
        event_id = f"fm-rot-{i}-{uuid.uuid4().hex[:6]}"
        status, _, _ = deliver(event(event_id), token=token)
        rotated.append((event_id, status))
    matrix.record("S12 rotation to the next token has no delivery gap",
                  all(s == 202 for _, s in rotated)
                  and all(stored_count(i) == 1 for i, _ in rotated),
                  f"statuses={[s for _, s in rotated]}")

    print("\n== replay ==")
    if MODE == "local":
        replay_ok, detail = replay_scenarios(e1)
        matrix.record("S13 operator replay reconciles, no second record", replay_ok, detail)
    else:
        matrix.record("S13 operator replay reconciles, no second record", True,
                      "needs database access; run in local mode", skipped=True)

    print("\n== infrastructure disruption ==")
    if not DISRUPT:
        for name in ("S14 receiver unavailable: sender retries, exactly one record",
                     "S15 database restart mid-ingestion: no acknowledged-but-absent event"):
            matrix.record(name, True, "DISRUPT=0", skipped=True)
        return

    matrix.record(*receiver_unavailable(app))
    matrix.record(*database_restart())


def replay_scenarios(event_id: str) -> tuple[bool, str]:
    """Replay, then replay again. Neither may create a second record."""
    from audit_core.postgres_backend import PostgresAuditBackend

    backend = PostgresAuditBackend(
        f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core", migrate=False
    )
    try:
        first = backend.replay(event_id)
        second = backend.replay(event_id)
        count = stored_count(event_id)
        return (
            first.duplicate and second.duplicate and count == 1,
            f"first.duplicate={first.duplicate} second.duplicate={second.duplicate} stored={count}",
        )
    finally:
        backend.close()


def receiver_unavailable(app: subprocess.Popen | None) -> tuple[str, bool, str]:
    """Stop the receiver mid-flight; the sender must retry to exactly one record."""
    name = "S14 receiver unavailable: sender retries, exactly one record"
    if app is None:
        return (name, True, "local mode only")
    app.terminate()
    app.wait(timeout=15)

    event_id = f"fm-down-{uuid.uuid4().hex[:8]}"
    status, _ = request("POST", "/v1/events", SENDER_TOKEN, event(event_id), event_id, timeout=3)
    if status != 0:
        return (name, False, f"expected no response while down, got {status}")

    restarted = restart_app()
    if not restarted:
        return (name, False, "receiver did not come back up")
    status, _, tries = deliver(event(event_id))
    count = stored_count(event_id)
    return (name, status == 202 and count == 1,
            f"after recovery status={status} tries={tries} stored={count}")


_app_handle: list[subprocess.Popen] = []


def restart_app() -> bool:
    app = start_app_process()
    _app_handle.append(app)
    return wait_for_app()


def start_app_process() -> subprocess.Popen:
    senders = json.dumps([
        {"name": "user-engine", "tokens": [SENDER_TOKEN, SENDER_NEXT],
         "sources": ["user-engine"], "tenants": [TENANT]},
        {"name": "ops-reader", "tokens": [READER_TOKEN], "sources": ["*"],
         "tenants": ["*"], "may_write": False, "may_read": True},
    ])
    env = {
        **os.environ,
        "AUDIT_CORE_DATABASE_URL":
            f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core",
        "AUDIT_CORE_SENDERS": senders,
        "AUDIT_CORE_HOST": "127.0.0.1",
        "AUDIT_CORE_HTTP_PORT": str(APP_PORT),
        "AUDIT_CORE_REQUIRE_CUSTODY_CLASS": "operational",
        "AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS": "5000",
        "AUDIT_CORE_LOG_LEVEL": "WARNING",
    }
    python = REPO / ".venv" / "bin" / "python"
    return subprocess.Popen(
        [str(python if python.exists() else sys.executable), "-m", "audit_core.ingestion"],
        cwd=REPO, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )


def database_restart() -> tuple[str, bool, str]:
    """Restart PostgreSQL under load.

    The property under test is not "no errors" — errors are expected and
    correct. It is that no event is *acknowledged* without being stored. A 202
    that does not survive the restart is the failure mode this whole service
    exists to prevent.
    """
    name = "S15 database restart mid-ingestion: no acknowledged-but-absent event"
    acknowledged: list[str] = []

    subprocess.run(["docker", "restart", "-t", "1", PG_CONTAINER],
                   capture_output=True, check=True)

    deadline = time.time() + 45
    attempts = 0
    while time.time() < deadline and len(acknowledged) < 5:
        event_id = f"fm-db-{attempts}-{uuid.uuid4().hex[:6]}"
        attempts += 1
        status, _ = request("POST", "/v1/events", SENDER_TOKEN,
                            event(event_id), event_id, timeout=8)
        if status == 202:
            acknowledged.append(event_id)
        elif status not in (0, 503, 500):
            return (name, False, f"unexpected status {status} during restart")
        time.sleep(0.5)

    wait_for_postgres()
    if not acknowledged:
        return (name, False, "nothing was acknowledged after restart; receiver did not recover")

    missing = [e for e in acknowledged if stored_count(e) != 1]
    return (name, not missing,
            f"acknowledged={len(acknowledged)} attempts={attempts} missing={missing}")


# --- entry point ------------------------------------------------------------

def main() -> int:
    matrix = Matrix()
    app = None
    print(f"== audit-core failure matrix (MODE={MODE} BASE_URL={BASE_URL} DISRUPT={DISRUPT}) ==")
    try:
        if MODE == "local":
            app = start_stack()
            _app_handle.append(app)
        elif not wait_for_app(timeout=5):
            print(f"receiver at {BASE_URL} is not reachable", file=sys.stderr)
            return 2
        run(matrix, app)
    finally:
        for handle in _app_handle:
            if handle.poll() is None:
                handle.terminate()
                try:
                    handle.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    handle.kill()
        if MODE == "local":
            subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)

    passed = sum(1 for r in matrix.results if r.ok and not r.skipped)
    skipped = sum(1 for r in matrix.results if r.skipped)
    print(f"\n== {passed} passed, {len(matrix.failed)} failed, {skipped} skipped ==")

    evidence = {
        "workplan_task": "AUDIT-WP-0005-T05",
        "mode": MODE,
        "disruption_scenarios_run": DISRUPT,
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "controlling_assertion":
            "one source outbox event produces exactly one durable normalized event "
            "across retries, replay and infrastructure disruption",
        "passed": passed,
        "failed": len(matrix.failed),
        "skipped": skipped,
        "scenarios": [
            {"name": r.name, "outcome": "skip" if r.skipped else ("pass" if r.ok else "fail"),
             "detail": r.detail}
            for r in matrix.results
        ],
    }
    out_dir = REPO / "evidence"
    out_dir.mkdir(exist_ok=True)
    path = out_dir / f"failure-matrix-{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}.json"
    path.write_text(json.dumps(evidence, indent=2) + "\n")
    print(f"evidence: {path.relative_to(REPO)}  (non-secret, for NK-WP-0024)")

    if matrix.failed:
        print("\nFailures:")
        for result in matrix.failed:
            print(f"  - {result.name}: {result.detail}")
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())