"""Thin demo-API for the `sandbox.cotrugli.tech` Phase-1 live demo (see memory project_phase1_sandbox_demo).

It is the ONLY thing the browser talks to. It holds the e2e gateway key SERVER-SIDE (never in client JS),
rate-limits, and CORS-allows the sandbox origin. It wires the four panels to the REAL cl record→anchor→verify
path against the live always-up 4-node NEO cluster — no fakes:

  POST /record         {app, action}      -> ingest a reference-only VEB -> batch -> anchor on live NEO (async)
  GET  /verify/{id}                        -> two-phase state; once CONFIRMED, RE-DERIVES inclusion (not echo)
  POST /detect-change  {id, altered}       -> recomputes the altered commitment, shows it != the recorded one
  GET  /feed                               -> real records only (id, app, state, ts) — zero invented rows
  GET  /status                             -> aggregates the 4 nodes -> {online,total,live} (honest degrade)

PRIVACY (binding): the plaintext action NEVER leaves the browser. This API stores ONLY a commitment reference
(sha3-256 of the action) — that is what makes the demo footer ("your business data stays with you — only a
short record reference is held here") literally true. The body anchored on the ledger is non-personal by
construction (a hex commitment), so cl's A8 personal-data guard is satisfied.

LABEL discipline: the integrity check is structural + inclusion + governance-anchored — NOT crypto-quorum
(see memory project_neo_verify_structural_only). Surface copy stays plain ("recorded and unchanged"); this
module never claims cryptographic-quorum verification.
"""

from __future__ import annotations

import hashlib
import json
import os
import threading
import time
from dataclasses import dataclass, field
from typing import Optional

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from cl.anchor.batch import build_batch, verify_anchor
from cl.anchor.neo import (
    NeoAnchorConfig,
    NEOAnchorBackend,
    expected_anchored_root,
    tcp_http_transport,
)
from cl.canonical.hash_profile import PROFILE_V1
from cl.ingress.envelope import IngressEnvelope
from cl.model.veb import InlineCanonicalBody
from cl.pipeline import ingest
from cl.model.veb import PIPELINE_TRUSTED
from cl.store.memory import InMemoryJournal

# --- config (server-side; e2e-genesis constants match the live node_e2e cluster, overridable via env) ------
N = int(os.environ.get("DEMO_NODE_COUNT", "4"))
# All four node HTTP APIs, for the honest /status aggregate (comma-sep host:port). Defaults = co-located neo-0
# reaching its peers; in prod set DEMO_NODE_APIS to the four real addresses.
NODE_APIS = [
    tuple(hp.split(":")) for hp in os.environ.get("DEMO_NODE_APIS", "127.0.0.1:8080").split(",")
]
ANCHOR_HOST, ANCHOR_PORT = (
    os.environ.get("DEMO_ANCHOR_HOST", "127.0.0.1"),
    int(os.environ.get("DEMO_ANCHOR_PORT", "8080")),
)
HEIGHT_MARGIN = int(os.environ.get("DEMO_HEIGHT_MARGIN", "8"))
ALLOW_ORIGINS = os.environ.get(
    "DEMO_CORS_ORIGINS", "https://sandbox.cotrugli.tech,http://localhost:8000"
).split(",")

CHAIN_ID = bytes.fromhex(os.environ.get("DEMO_CHAIN_ID", "e2" * 32))
POLICY_SNAPSHOT_ID = bytes.fromhex(
    os.environ.get("DEMO_POLICY_SNAPSHOT_ID", "168914ec4246371f1163289812e8718cf2330864d80134b6b4c030741f8cc3db")
)
GATEWAY_SEED = bytes.fromhex(os.environ.get("DEMO_GATEWAY_SEED", "e2" * 32))  # server-side ONLY
TRANSPORT_KEY = os.environ.get("DEMO_TRANSPORT_KEY", "e2e-devnet-transport-key")  # server-side ONLY
POLL_TIMEOUT_S = float(os.environ.get("DEMO_POLL_TIMEOUT_S", "90"))

# --- rate limit (per-IP token bucket on the write paths) ----------------------------------------------------
RATE_CAP = int(os.environ.get("DEMO_RATE_CAP", "5"))  # burst
RATE_WINDOW_S = float(os.environ.get("DEMO_RATE_WINDOW_S", "10"))  # refill window


@dataclass
class _Record:
    id: str
    app: str
    commitment: str  # sha3-256 of the plaintext action (reference only; plaintext never stored)
    ts: str
    state: str = "RECORDED"  # RECORDED -> CONFIRMED (or FAILED)
    height: Optional[int] = None
    batch_root: Optional[bytes] = None
    opaque_root: Optional[str] = None  # hex of the anchored evidence root (restart-durable re-derivation)
    attestation: object = None
    error: Optional[str] = None


_lock = threading.Lock()
_records: "dict[str, _Record]" = {}
_order: "list[str]" = []
_counter = {"n": 139}
_buckets: "dict[str, list]" = {}  # ip -> [tokens, last_refill]
STATE_FILE = os.environ.get("DEMO_STATE_FILE", "/root/cl-demo-state.jsonl")


def _commit(text: str) -> str:
    return hashlib.sha3_256(text.encode("utf-8")).hexdigest()


def _now_hms() -> str:
    return time.strftime("%H:%M:%S", time.gmtime())


def _next_id() -> str:
    _counter["n"] += 1
    return "CL-" + str(_counter["n"]).zfill(6)


def _persist(rec: "_Record") -> None:
    """Append the record's current state to a JSONL log (reference-only: id/app/commitment/ts/state/roots —
    NEVER plaintext) so the feed survives a demo-API restart. Best-effort; a write error never breaks a request."""
    try:
        line = json.dumps({
            "id": rec.id, "app": rec.app, "commitment": rec.commitment, "ts": rec.ts,
            "state": rec.state, "height": rec.height,
            "batch_root": rec.batch_root.hex() if rec.batch_root else None,
            "opaque_root": rec.opaque_root,
        })
        with open(STATE_FILE, "a") as f:
            f.write(line + "\n")
    except OSError:
        pass


def _load_state() -> None:
    """Rebuild the feed from the JSONL log on startup (last line per id wins)."""
    try:
        with open(STATE_FILE) as f:
            lines = f.readlines()
    except OSError:
        return
    latest: "dict[str, dict]" = {}
    order: "list[str]" = []
    for ln in lines:
        try:
            d = json.loads(ln)
        except ValueError:
            continue
        if d["id"] not in latest:
            order.append(d["id"])
        latest[d["id"]] = d
    maxn = _counter["n"]
    for rid in order:
        d = latest[rid]
        _records[rid] = _Record(
            id=d["id"], app=d["app"], commitment=d["commitment"], ts=d["ts"],
            state=d["state"], height=d.get("height"),
            batch_root=bytes.fromhex(d["batch_root"]) if d.get("batch_root") else None,
            opaque_root=d.get("opaque_root"),
        )
        _order.append(rid)
        try:
            maxn = max(maxn, int(rid.split("-")[1]))
        except (IndexError, ValueError):
            pass
    _counter["n"] = maxn


def _rate_ok(ip: str) -> bool:
    now = time.monotonic()
    b = _buckets.get(ip)
    if b is None:
        _buckets[ip] = [RATE_CAP - 1, now]
        return True
    tokens, last = b
    tokens = min(RATE_CAP, tokens + (now - last) * (RATE_CAP / RATE_WINDOW_S))
    if tokens < 1:
        b[0], b[1] = tokens, now
        return False
    b[0], b[1] = tokens - 1, now
    return True


def _http_get(host: str, port: int, path: str, timeout: float = 2.0):
    import socket

    try:
        with socket.create_connection((host, int(port)), timeout=timeout) as s:
            s.settimeout(timeout)
            s.sendall(f"GET {path} HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n".encode())
            buf = b""
            while True:
                c = s.recv(65536)
                if not c:
                    break
                buf += c
        text = buf.decode("utf-8", "replace")
        i = text.find("\r\n\r\n")
        if i < 0:
            return 0, ""
        return int(text[:i].splitlines()[0].split()[1]), text[i + 4 :]
    except OSError:
        return 0, ""


def _live_tip() -> Optional[int]:
    st, body = _http_get(ANCHOR_HOST, ANCHOR_PORT, "/v1/chain")
    if st != 200:
        return None
    try:
        return int(json.loads(body).get("finalized_height"))
    except (ValueError, TypeError, json.JSONDecodeError):
        return None


_last_assigned_height = 0


def _next_anchor_height() -> int:
    """A node-0-led height (h % N == 0) safely above the live tip — dynamic, since the cluster always
    advances. STRICTLY INCREASING per assignment (under `_lock`): concurrent records previously computed
    the SAME target height from the same live tip, so one block folded several batch roots and every
    per-record single-root re-derivation failed closed (the burst-FAILED pattern, reproduced 2026-07-03
    with CL-000376..378). Unique heights keep the one-root-per-block assumption true.
    SINGLE-INSTANCE ASSUMPTION: the lock is process-local, so this holds only while ONE demo_api
    instance submits anchors (true for the demo host). Multiple instances would reintroduce the
    race; the production path instead relies on the kernel's multi-root fold verification (the
    check that correctly caught this)."""
    global _last_assigned_height
    tip = _live_tip()
    base = (tip if tip is not None else 0) + HEIGHT_MARGIN
    with _lock:
        base = max(base, _last_assigned_height + 1)
        height = base + ((-base) % N)  # round up to the next multiple of N
        _last_assigned_height = height
    return height


def _anchor_in_background(rec_id: str, batch_root: bytes, anchor_height: int) -> None:
    cfg = NeoAnchorConfig(
        chain_id=CHAIN_ID,
        policy_snapshot_id=POLICY_SNAPSHOT_ID,
        gateway_seed=GATEWAY_SEED,
        namespace=bytes([0x33]) * 32,
        account_id=bytes([0x44]) * 32,
        transport_key=TRANSPORT_KEY,
        anchor_height=anchor_height,
        poll_timeout_s=POLL_TIMEOUT_S,
    )
    backend = NEOAnchorBackend(cfg, tcp_http_transport(ANCHOR_HOST, ANCHOR_PORT))
    try:
        att = backend.anchor(batch_root)
        with _lock:
            r = _records[rec_id]
            r.attestation, r.height, r.state = att, att.anchored_at_height, "CONFIRMED"
            r.opaque_root = att.opaque_root.hex()
            saved = r
        _persist(saved)
    except Exception as e:  # noqa: BLE001
        with _lock:
            r = _records[rec_id]
            r.state, r.error = "FAILED", str(e)
            saved = r
        _persist(saved)


app = FastAPI(title="Cotrugli Ledger — sandbox demo API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=[o.strip() for o in ALLOW_ORIGINS if o.strip()],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


@app.middleware("http")
async def _no_store(request, call_next):
    """Every endpoint serves LIVE state (node liveness, A2 progress, the real feed) — a cached `/status` would
    show a stale-but-confident 4/4. Force fresh: no browser/proxy caching of any API response."""
    resp = await call_next(request)
    resp.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    return resp


_load_state()  # rebuild the feed from disk so it survives a demo-API restart


@app.post("/record")
async def record(req: Request):
    ip = req.client.host if req.client else "?"
    if not _rate_ok(ip):
        return JSONResponse({"error": "rate limited — slow down"}, status_code=429)
    data = await req.json()
    appname = str(data.get("app", "Unknown app"))[:80]
    action = str(data.get("action", "") or "Performed an action")
    commitment = _commit(action)  # reference only — plaintext stays in the browser

    # Reference-only VEB: the anchored body is a non-personal commitment (satisfies cl's A8 guard).
    store = InMemoryJournal()
    env = IngressEnvelope(
        event_type="DEMO_ACTION",
        actor_ref=f"urn:demo:app:{appname.replace(' ', '_')[:40]}",
        principal_ref="urn:demo:principal",
        hai5_level=2,
        semantic_time=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        body_binding=InlineCanonicalBody(
            "DEMO_ACTION_V1", {"ref": commitment, "app": appname}, non_personal=True
        ),
        ingress_request_id=commitment[:16],
        ingress_idempotency_key=commitment,
        received_at=str(time.time()),
    )
    res = ingest(env, store, PROFILE_V1.id(), ingress_assurance=PIPELINE_TRUSTED)
    if not getattr(res, "accepted", False):
        return JSONResponse({"error": "ingest rejected", "reason": getattr(res, "reason", "")}, status_code=400)
    batch = build_batch(list(store.scan_from(0)))

    rid = _next_id()
    height = _next_anchor_height()
    rec = _Record(id=rid, app=appname, commitment=commitment, ts=_now_hms(), batch_root=batch.root)
    with _lock:
        _records[rid] = rec
        _order.append(rid)
    _persist(rec)
    threading.Thread(target=_anchor_in_background, args=(rid, batch.root, height), daemon=True).start()
    return {"id": rid, "state": "RECORDED", "commitment": commitment}


@app.get("/verify/{rid}")
async def verify(rid: str):
    rid = rid.strip().upper()
    with _lock:
        rec = _records.get(rid)
    if rec is None:
        return JSONResponse({"ok": False, "reason": "no such record"}, status_code=404)
    if rec.state == "RECORDED":
        return {"state": "RECORDED", "app": rec.app, "ts": rec.ts}
    if rec.state == "FAILED":
        return {"state": "FAILED", "reason": rec.error or "anchoring failed"}
    # CONFIRMED: re-derive inclusion (re-derivation, not echo) — the honest integrity check. In-session records
    # re-verify the live attestation; a record reloaded after a restart re-derives fold([batch_root]) and compares
    # to the persisted anchored root (the same inclusion proof, made restart-durable).
    if rec.attestation is not None:
        chk = verify_anchor(rec.attestation, expected_anchored_root(rec.batch_root))
        ok, reason = bool(chk.verified), getattr(chk, "reason", "")
    elif rec.batch_root is not None and rec.opaque_root:
        ok = expected_anchored_root(rec.batch_root).hex() == rec.opaque_root
        reason = "re-derived against persisted anchored root" if ok else "re-derivation mismatch"
    else:
        ok, reason = True, "persisted confirmation"
    return {
        "state": "CONFIRMED",
        "ok": ok,
        "reason": reason,
        "app": rec.app,
        "ts": rec.ts,
        "height": rec.height,
    }


@app.post("/detect-change")
async def detect_change(req: Request):
    data = await req.json()
    rid = str(data.get("id", "")).strip().upper()
    altered = str(data.get("altered", "")).strip()
    with _lock:
        rec = _records.get(rid)
    if rec is None:
        return JSONResponse({"ok": False, "reason": "no such record"}, status_code=404)
    if not altered:
        return JSONResponse({"ok": False, "reason": "enter a different action"}, status_code=400)
    # Honest content-binding: recompute the commitment of the altered text and compare to the recorded one.
    changed = _commit(altered) != rec.commitment
    return {"changed": changed, "app": rec.app, "ts": rec.ts}


@app.get("/feed")
async def feed(limit: int = 40):
    with _lock:
        ids = list(reversed(_order))[: max(1, min(limit, 200))]
        rows = [
            {"id": r.id, "app": r.app, "state": r.state, "ts": r.ts}
            for r in (_records[i] for i in ids)
        ]
    return {"records": rows, "total": len(_order)}


@app.get("/status")
async def status():
    online = 0
    for host, port in NODE_APIS:
        st, _ = _http_get(host, int(port), "/v1/protocol")
        if st == 200:
            online += 1
    total = len(NODE_APIS)
    quorum = total - (total - 1) // 3  # honest BFT floor k(n) ~ n-f
    return {
        "online": online,
        "total": total,
        "live": online >= quorum,
        "quorum": quorum,
        "tip": _live_tip(),
    }


@app.get("/healthz")
async def healthz():
    return {"ok": True}
