"""SQLite-backed, conduit-opaque affect store (Bifrost affect plane, v1).

Contract: docs/contracts/bifrost_affect_provider.contract.md

The store persists Worldtree's Tier-3 affect snapshots verbatim and round-trips
them. It runs NO affect logic: it reads only the two addressing keys
(`agent_id`, `end_user_id`) and treats `pad`/`valence`/`persona_baselines`/
`emitted_at` as an opaque blob (INV-001). Idempotency is replay-or-conflict,
keyed by (actor-from-ctx, idempotency_key) (INV-008), and snapshots are
last-write-wins by arrival across distinct keys (INV-002).
"""
from __future__ import annotations

import hashlib
import json
import sqlite3
import time
from typing import Any

from bifrost.affect import AffectIdempotencyConflict, AffectInvalidArguments
from bifrost.consumer import ConsumerRegistration, build_affect_app
from bifrost.reference_server import JwtVerifier
from starlette.requests import Request
from starlette.responses import JSONResponse

_SHORT_RETRY_TTL_SECONDS = 300


def _ctx_actor(ctx: Any) -> str:
    """Mirror bifrost reference `_ctx_actor`: idempotency actor = JWT `sub`.

    The real DispatchContext exposes the `sub` claim as `session_id`; test
    contexts set `jwt_sub`. (INV-006: the actor comes from ctx, never the
    snapshot.)
    """
    return str(getattr(ctx, "jwt_sub", None) or getattr(ctx, "session_id", ""))


class RatatoskrAffectStore:
    """The affect `MemoryDataStore`-shaped store handed to `build_affect_app`."""

    affect_supported = True

    def __init__(self, conn: sqlite3.Connection):
        self._conn = conn

    async def emit(
        self,
        snapshot: dict,
        *,
        idempotency_key: str,
        ctx: Any,
        idempotency_class: str | None = None,
    ) -> dict:
        del idempotency_class  # reserved; affect.* uses a single short-retry class

        # INV-001: read ONLY the two addressing keys; everything else is opaque.
        agent_id = snapshot.get("agent_id")
        end_user_id = snapshot.get("end_user_id")
        if not (
            isinstance(agent_id, str)
            and agent_id
            and isinstance(end_user_id, str)
            and end_user_id
        ):
            raise AffectInvalidArguments("snapshot missing agent_id / end_user_id")
        if not (isinstance(idempotency_key, str) and idempotency_key):
            raise AffectInvalidArguments("idempotency_key required")

        # Whole-blob serialize + hash is opacity-safe (not a field read).
        blob = json.dumps(snapshot, sort_keys=True, separators=(",", ":"))
        digest = hashlib.sha256(blob.encode()).hexdigest()
        idempotency_id = f"affect.emit|{_ctx_actor(ctx)}|{idempotency_key}"

        # INV-008: replay (same digest) -> no-op; reuse with different digest -> conflict.
        cached = self._conn.execute(
            "SELECT digest FROM affect_idempotency WHERE idempotency_id = ?",
            (idempotency_id,),
        ).fetchone()
        if cached is not None:
            if cached[0] == digest:
                return {"stored": True}
            raise AffectIdempotencyConflict(
                "idempotency key reused with different payload"
            )

        # INV-002 + INV-007: LWW snapshot upsert + idempotency record, one transaction.
        now = time.time()
        with self._conn:
            self._conn.execute(
                "INSERT INTO affect_snapshots (agent_id, end_user_id, snapshot_json, arrived_at) "
                "VALUES (?, ?, ?, ?) "
                "ON CONFLICT(agent_id, end_user_id) DO UPDATE SET "
                "snapshot_json = excluded.snapshot_json, arrived_at = excluded.arrived_at",
                (agent_id, end_user_id, blob, str(now)),
            )
            self._conn.execute(
                "INSERT INTO affect_idempotency (idempotency_id, digest, expires_at) "
                "VALUES (?, ?, ?)",
                (idempotency_id, digest, now + _SHORT_RETRY_TTL_SECONDS),
            )
        return {"stored": True}

    def get(self, agent_id: str, end_user_id: str) -> dict | None:
        """Sync read-back seam returning the verbatim stored snapshot (or None).

        The async wire verb `fetch` wraps this; tests, the D2 read route, and
        rehydrate-seed also call it directly.
        """
        row = self._conn.execute(
            "SELECT snapshot_json FROM affect_snapshots WHERE agent_id = ? AND end_user_id = ?",
            (agent_id, end_user_id),
        ).fetchone()
        return json.loads(row[0]) if row is not None else None

    async def fetch(self, agent_id: str, end_user_id: str) -> dict:
        """Async affect.fetch handler — return the stored snapshot in bifrost's
        {found, snapshot} shape, conduit-opaque.

        INV-010 (strong-or-absent): bifrost >=0.10.0 gates EVERY affect op on the
        store advertising affect_supported + emit + fetch (`_supports_affect_plane`),
        so this method MUST exist for the affect capability to dispatch at all —
        an emit-only store 400s. Mirrors the reference InMemoryAffectStore.fetch;
        returns the whole blob opaque (INV-001 — never reads pad/valence).
        """
        if not (
            isinstance(agent_id, str)
            and agent_id
            and isinstance(end_user_id, str)
            and end_user_id
        ):
            raise AffectInvalidArguments("fetch missing agent_id / end_user_id")
        snap = self.get(agent_id, end_user_id)
        if snap is None:
            return {"found": False}
        return {"found": True, "snapshot": snap}


def open_affect_store(db_path: str) -> RatatoskrAffectStore:
    """Open the SQLite-backed affect store, creating the schema on first use."""
    # check_same_thread=False: the affect provider is an ASGI app; Starlette/uvicorn
    # may run a handler off the connection's creating thread (and TestClient always
    # does). Access stays serialized by the event loop, so this is safe.
    conn = sqlite3.connect(db_path, check_same_thread=False)
    # INV-006: state busy_timeout explicitly rather than lean on sqlite3's timeout=5.0
    # default — a contended write WAITS up to 5s instead of failing SQLITE_BUSY at once
    # (prep for the composite/standalone two-process topology).
    conn.execute("PRAGMA busy_timeout=5000")
    if db_path != ":memory:":
        conn.execute("PRAGMA journal_mode=WAL")
    conn.execute(
        "CREATE TABLE IF NOT EXISTS affect_snapshots ("
        "agent_id TEXT NOT NULL, end_user_id TEXT NOT NULL, "
        "snapshot_json TEXT NOT NULL, arrived_at TEXT, "
        "PRIMARY KEY (agent_id, end_user_id))"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS affect_idempotency ("
        "idempotency_id TEXT PRIMARY KEY, digest TEXT NOT NULL, expires_at REAL)"
    )
    conn.commit()
    return RatatoskrAffectStore(conn)


def add_affect_read_route(app, store: RatatoskrAffectStore) -> None:
    """Mount the non-bifrost PAD read route GET /affect/state/{agent_id} on `app`,
    reading store.get. SHARED by build_affect_provider_app and the combined provider
    (#18 INV-011 / D2 INV-007) — add_route (NOT Mount) keeps /bifrost/* top-level so
    the op-feed path check still matches them and passes this route through untouched.
    No JWT (internal-LAN trust model).
    """
    async def _affect_state_route(request: Request) -> JSONResponse:
        agent_id = request.path_params["agent_id"]
        end_user_id = request.query_params.get("end_user_id")
        if not end_user_id:  # PRE-001: never look up against a None/empty partition
            return JSONResponse({"error_code": "missing_end_user_id"}, status_code=400)
        snap = store.get(agent_id, end_user_id)
        if snap is None:  # INV-003: explicit no-data, never a fabricated zeroed PAD
            return JSONResponse(
                {
                    "error_code": "no_affect_snapshot",
                    "agent_id": agent_id,
                    "end_user_id": end_user_id,
                },
                status_code=404,
            )
        return JSONResponse(snap)

    app.add_route("/affect/state/{agent_id}", _affect_state_route, methods=["GET"])


def build_affect_provider_app(
    store: RatatoskrAffectStore,
    heimdall_key: bytes,
    consumer_id: str = "ratatoskr",
):
    """Wire the JWT verifier + registration and hand the store to bifrost.

    Returns a Starlette ASGI app exposing POST /bifrost/handshake and
    POST /bifrost/affect-call. The library owns the wire; this is the thin glue.
    """
    if getattr(store, "affect_supported", False) is not True:  # INV-005
        raise ValueError("store must advertise affect_supported=True")
    if not (isinstance(heimdall_key, bytes) and heimdall_key):
        raise ValueError("heimdall_key must be non-empty bytes")
    verifier = JwtVerifier(algorithm="HS256", key_bytes=heimdall_key)
    registration = ConsumerRegistration(consumer_id=consumer_id)
    app = build_affect_app(store=store, verifier=verifier, registration=registration)

    # Issue #18 (Deliverable 2): mount the non-bifrost PAD read route. Extracted into
    # add_affect_read_route so the combined provider mounts the SAME one (Deliverable 1,
    # INV-011) over the same affect.db.
    add_affect_read_route(app, store)
    return app
