"""HUX-01 activity events: the per-conversation append-only log and its readers.

``emit`` is the one write path every family uses. It runs the redaction
pipeline, allocates ``seq`` under the conversation lock, honours idempotency
keys and keeps a checkpoint document per conversation. Readers page by
``after_seq`` or follow an SSE stream whose ``id:`` is the seq, and every
served record passes serve-time redaction for the caller's surface.
"""

from __future__ import annotations

import json
import threading
import time
from pathlib import Path
from typing import Any
from collections.abc import Iterator

from hux import contracts, redaction
from hux.errors import Invalid, NotFound, TooLarge
from hux.http import Request, Response, Router, page
from hux.identity import Identity
from hux.store import TenantStore, new_id, now_iso

FAMILY = "events"
SEQ_FAMILY = "events_seq"
IDEM_FAMILY = "events_idem"
PAGE_MAX = 200
PAGE_DEFAULT = 100
STREAM_MAX_POLLS = 900
STREAM_POLL_SECONDS = 1.0
SCHEMAS = contracts.load_all()
KINDS = frozenset(contracts.load_schema("event.schema.json")["properties"]["kind"]["enum"])
USER_KINDS = frozenset({"message.user", "approval.resolved", "run.cancelled", "memory.forgotten", "suggestion.dismissed"})
_streams: dict[str, int] = {}
_streams_guard = threading.Lock()


def _seq_id(conversation_id: str) -> str:
    return f"seq_{conversation_id}"


def is_private(store: TenantStore, conversation_id: str) -> bool:
    """True when the conversation document (HUX-03) says the mode is private (SO-28)."""
    try:
        return store.get("conversations", conversation_id).get("mode") == "private"
    except NotFound:
        return False


def _ledger_path(store: TenantStore, conversation_id: str) -> Path:
    return store.root / FAMILY / f"{conversation_id}.jsonl"


def conversation_known(store: TenantStore, conversation_id: str) -> bool:
    """Ownership check (SO-18): the conversation document, its seq checkpoint or its event ledger exists in this subject's tree."""
    return (
        store.exists("conversations", conversation_id)
        or store.exists(SEQ_FAMILY, _seq_id(conversation_id))
        or _ledger_path(store, conversation_id).is_file()
    )


def last_seq(store: TenantStore, conversation_id: str) -> int:
    """The seq of the newest complete ledger line, read from the file tail (0 when the ledger is empty or absent).

    A crash between the ledger append and the checkpoint write leaves the
    checkpoint behind the ledger; ``emit`` takes the larger of the two so a seq
    is never handed out twice (F5). Only the last ``LINE_CAP_BYTES`` window is
    read, and a torn trailing line is skipped.
    """
    path = _ledger_path(store, conversation_id)
    if not path.is_file():
        return 0
    with open(path, "rb") as handle:
        handle.seek(0, 2)
        size = handle.tell()
        handle.seek(max(0, size - redaction.LINE_CAP_BYTES - 2))
        tail = handle.read()
    for line in reversed(tail.split(b"\n")):
        if not line:
            continue
        try:
            return int(json.loads(line).get("seq", 0))
        except (json.JSONDecodeError, AttributeError, ValueError):
            continue
    return 0


def _actor(identity: Identity, kind: str) -> dict[str, str]:
    if kind in USER_KINDS and identity.trust != "worker":
        return {"type": "user", "id": identity.subject}
    if identity.trust == "worker":
        return {"type": "system", "id": "hux-worker"}
    return {"type": "assistant", "id": "hermes"}


def _find_replay(store: TenantStore, conversation_id: str, key: str) -> dict[str, Any] | None:
    for row in store.read(IDEM_FAMILY, conversation_id):
        if row.get("idempotency_key") == key:
            for event in store.read(FAMILY, conversation_id):
                if event.get("id") == row.get("event_id"):
                    return event
    return None


def build(identity: Identity, conversation_id: str, kind: str, summary: str, detail: dict | None, evidence: list | None,
          sensitivity: str, run_id: str | None, turn: int | None, correlation_id: str | None, idempotency_key: str | None) -> dict[str, Any]:
    """Run the redaction pipeline and shape an unsequenced ``hux.event.v1`` record."""
    if kind not in KINDS:
        raise Invalid(f"unknown event kind {kind!r}")
    if sensitivity not in ("public", "personal", "sensitive", "restricted"):
        raise Invalid("unknown sensitivity")
    hits: list[str] = []
    summary_clean = redaction.scrub_value(str(summary or "").strip(), hits)[: redaction.SUMMARY_MAX] or "[empty]"
    detail_clean = redaction.scrub_value(redaction.filter_detail(kind, detail), hits)
    detail_clean, truncated = redaction.cap_detail(detail_clean)
    evidence_clean = redaction.scrub_value(redaction.filter_evidence(evidence), hits)
    stamp = now_iso()
    record: dict[str, Any] = {
        "schema": "hux.event.v1",
        "id": new_id("evt"),
        "seq": 0,
        "ts": stamp,
        "conversation_id": conversation_id,
        "kind": kind,
        "summary": summary_clean,
        "provenance": {"surface": identity.surface, "actor": _actor(identity, kind), "recorded_at": stamp, "conversation_id": conversation_id},
        "sensitivity": sensitivity,
        "redaction": redaction.derive_level(sensitivity, hits, truncated),
        "turn": max(0, int(turn or 0)),
        "identity": identity.record(),
    }
    if detail_clean:
        record["detail"] = detail_clean
    if evidence_clean:
        record["evidence"] = evidence_clean
    if run_id:
        record["run_id"] = str(run_id)[:120]
        record["provenance"]["run_id"] = record["run_id"]
    if correlation_id:
        record["correlation_id"] = str(correlation_id)[:120]
    if idempotency_key:
        record["idempotency_key"] = idempotency_key
    return record


def emit_with_status(store: TenantStore, identity: Identity, conversation_id: str, kind: str, summary: str, detail: dict | None = None,
                     evidence: list | None = None, sensitivity: str = "personal", run_id: str | None = None, turn: int | None = None,
                     correlation_id: str | None = None, idempotency_key: str | None = None) -> tuple[dict[str, Any] | None, bool]:
    """Like ``emit`` but also says whether the record was an idempotent replay. None means private mode (nothing written)."""
    if is_private(store, conversation_id):
        return None, False
    record = build(identity, conversation_id, kind, summary, detail, evidence, sensitivity, run_id, turn, correlation_id, idempotency_key)
    with store.lock(f"{FAMILY}:{conversation_id}"):
        if idempotency_key:
            existing = _find_replay(store, conversation_id, idempotency_key)
            if existing is not None:
                return existing, True
        seq_id = _seq_id(conversation_id)
        checkpoint = store.get(SEQ_FAMILY, seq_id) if store.exists(SEQ_FAMILY, seq_id) else {"id": seq_id, "next_seq": 1, "last_event_id": ""}
        record["seq"] = max(int(checkpoint["next_seq"]), last_seq(store, conversation_id) + 1)
        problems = contracts.validate_record(record, SCHEMAS)
        if problems:
            raise Invalid("event failed contract validation", problems)
        line = json.dumps(record, sort_keys=True, separators=(",", ":")).encode()
        if len(line) > redaction.LINE_CAP_BYTES:
            raise TooLarge("event line exceeds 64 KiB")
        store.append(FAMILY, conversation_id, record)
        store.put(SEQ_FAMILY, {**checkpoint, "next_seq": record["seq"] + 1, "last_event_id": record["id"], "checkpointed_at": now_iso()})
        if idempotency_key:
            store.append(IDEM_FAMILY, conversation_id, {"idempotency_key": idempotency_key, "event_id": record["id"], "seq": record["seq"], "at": record["ts"]})
    return record, False


def emit(store: TenantStore, identity: Identity, conversation_id: str, kind: str, summary: str, detail: dict | None = None,
         evidence: list | None = None, sensitivity: str = "personal", run_id: str | None = None, turn: int | None = None,
         correlation_id: str | None = None, idempotency_key: str | None = None) -> dict[str, Any] | None:
    """Append one event and return the stored record (None when the conversation is private)."""
    record, _ = emit_with_status(store, identity, conversation_id, kind, summary, detail, evidence, sensitivity, run_id, turn, correlation_id, idempotency_key)
    return record


def read_after(store: TenantStore, conversation_id: str, after_seq: int, limit: int) -> list[dict[str, Any]]:
    """Stored events with ``seq > after_seq``, oldest first, at most ``limit``."""
    rows = [row for row in store.read(FAMILY, conversation_id) if int(row.get("seq", 0)) > after_seq]
    return rows[:limit]


def rewrite_full(store: TenantStore, conversation_id: str, summary: str, reason: str, match=None) -> int:
    """Rewrite matching events of one conversation to ``redaction.level: full`` (SO-24, forget, decay). Returns the count."""
    with store.lock(f"{FAMILY}:{conversation_id}"):
        rows = store.read(FAMILY, conversation_id)
        changed = 0
        out: list[dict[str, Any]] = []
        for row in rows:
            if row.get("redaction", {}).get("level") != "full" and (match is None or match(row)):
                row = redaction.full_redaction(row, summary, reason)
                changed += 1
            out.append(row)
        if changed:
            store.rewrite(FAMILY, conversation_id, out)
    return changed


def redact_memory_references(store: TenantStore, memory_id: str) -> int:
    """Fully redact every event, in any conversation, that names a forgotten memory id (SO-24)."""

    def references(row: dict[str, Any]) -> bool:
        if row.get("detail", {}).get("memory_id") == memory_id:
            return True
        return any(ref.get("kind") == "memory" and ref.get("id") == memory_id for ref in row.get("evidence", []))

    return sum(rewrite_full(store, name, "[forgotten memory]", "memory forgotten", references) for name in store.ledgers(FAMILY))


# -- routes ------------------------------------------------------------------

def _int_query(request: Request, name: str, default: int, floor: int = 0) -> int:
    raw = request.query.get(name, "")
    if raw == "":
        return default
    if not raw.lstrip("-").isdigit():
        raise Invalid(f"{name} must be an integer")
    return max(floor, int(raw))


def _require_conversation(request: Request) -> str:
    conversation_id = request.params["id"]
    if not conversation_known(request.store, conversation_id):
        raise NotFound("conversation not found")
    return conversation_id


def list_events(request: Request) -> Response:
    """``GET /hux/v1/conversations/{id}/events?after_seq=&limit=``: one page, ``next`` is the last seq served."""
    conversation_id = _require_conversation(request)
    after_seq = _int_query(request, "after_seq", 0)
    limit = min(_int_query(request, "limit", PAGE_DEFAULT, 1), PAGE_MAX)
    rows = read_after(request.store, conversation_id, after_seq, limit)
    items = [redaction.redact_record(row, request.identity.surface) for row in rows]
    request.audit("events.list", conversation_id)
    return page(items, items[-1]["seq"] if len(items) == limit else None)


def append_event(request: Request) -> Response:
    """``POST /hux/v1/conversations/{id}/events``: append from the trusted hop; ids and seq are server-assigned (SO-15)."""
    conversation_id = _require_conversation(request)
    body = request.body if isinstance(request.body, dict) else None
    if body is None:
        raise Invalid("body must be an object")
    if "id" in body or "seq" in body:
        raise Invalid("id and seq are server-assigned")
    if not isinstance(body.get("kind"), str) or not isinstance(body.get("summary"), str):
        raise Invalid("kind and summary are required")
    detail = body.get("detail") if isinstance(body.get("detail"), dict) else None
    evidence = body.get("evidence") if isinstance(body.get("evidence"), list) else None
    turn = body.get("turn") if isinstance(body.get("turn"), int) else None
    record, replayed = emit_with_status(
        request.store, request.identity, conversation_id, body["kind"], body["summary"], detail, evidence,
        body.get("sensitivity", "personal"), body.get("run_id"), turn, body.get("correlation_id"), request.idempotency_key() or None,
    )
    if record is None:
        request.audit("events.append", conversation_id, "allow", "private_mode")
        return Response(204)
    request.audit("events.append", f"{conversation_id}/{record['id']}", "allow", "replayed" if replayed else "")
    served = redaction.redact_record(record, request.identity.surface)
    return Response(200 if replayed else 201, served, {"HUX-Replayed": "true"} if replayed else {})


def _sse(record: dict[str, Any]) -> bytes:
    return f"id: {record['seq']}\nevent: {record['kind']}\ndata: {json.dumps(record, sort_keys=True)}\n\n".encode()


def stream_events(request: Request) -> Response:
    """``GET /hux/v1/conversations/{id}/events/stream``: SSE replay from ``Last-Event-ID`` (or ``after_seq``), then a bounded live poll.

    ``max_polls`` and ``poll_ms`` (query) bound the live phase so a client, or a
    test, decides how long to wait; the server caps them at the 15 minute idle
    limit (SO-17). A second stream for the same (subject, conversation) closes
    the first.
    """
    conversation_id = _require_conversation(request)
    last_id = request.header("Last-Event-ID")
    cursor = int(last_id) if last_id.isdigit() else _int_query(request, "after_seq", 0)
    max_polls = min(_int_query(request, "max_polls", STREAM_MAX_POLLS), STREAM_MAX_POLLS)
    poll_seconds = min(_int_query(request, "poll_ms", int(STREAM_POLL_SECONDS * 1000)), 60000) / 1000
    store, surface = request.store, request.identity.surface
    key = f"{request.identity.subject}:{conversation_id}"
    with _streams_guard:
        token = _streams[key] = _streams.get(key, 0) + 1
    request.audit("events.stream", conversation_id)

    def generate() -> Iterator[bytes]:
        position = cursor
        yield b"retry: 2000\n\n"
        polls = 0
        while True:
            for record in read_after(store, conversation_id, position, PAGE_MAX):
                position = record["seq"]
                yield _sse(redaction.redact_record(record, surface))
            if polls >= max_polls or _streams.get(key) != token:
                break
            polls += 1
            yield b": keepalive\n\n"
            time.sleep(poll_seconds)

    return Response(200, stream=generate)


def register(router: Router) -> None:
    """Attach HUX-01 routes."""
    router.add("GET", "/hux/v1/conversations/{id}/events", "HUX-01", "events.list", list_events)
    router.add("POST", "/hux/v1/conversations/{id}/events", "HUX-01", "events.append", append_event)
    router.add("GET", "/hux/v1/conversations/{id}/events/stream", "HUX-01", "events.stream", stream_events)