from __future__ import annotations

import asyncio
import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any

from .models import EventEnvelope


class EventStoreCorrupt(RuntimeError):
    pass


class EventStore:
    """Append-only JSONL event store with monotonic sequencing and live replay.

    The store is deliberately small and local. It has no network client and never
    downloads anything. A process lock serializes appends and makes subscription
    registration race-free relative to replay.
    """

    def __init__(self, path: Path, *, max_replay_events: int = 10_000) -> None:
        self.path = Path(path)
        self.max_replay_events = max_replay_events
        self._lock = asyncio.Lock()
        self._next_seq: int | None = None
        self._subscribers: set[asyncio.Queue[EventEnvelope]] = set()

    async def initialize(self) -> None:
        async with self._lock:
            if self._next_seq is not None:
                return
            events = await asyncio.to_thread(self._read_sync)
            self._next_seq = (events[-1].seq + 1) if events else 1

    async def append(
        self,
        event_type: str,
        *,
        session_id: str,
        turn_id: str,
        operation_id: str,
        trace_id: str,
        source: str,
        payload: dict[str, Any] | None = None,
        causation_id: str | None = None,
    ) -> EventEnvelope:
        await self.initialize()
        async with self._lock:
            assert self._next_seq is not None
            event = EventEnvelope(
                type=event_type,
                seq=self._next_seq,
                session_id=session_id,
                turn_id=turn_id,
                operation_id=operation_id,
                trace_id=trace_id,
                causation_id=causation_id,
                source=source,
                payload=payload or {},
            )
            await asyncio.to_thread(self._append_sync, event)
            self._next_seq += 1
            for queue in tuple(self._subscribers):
                queue.put_nowait(event)
            return event

    async def replay(
        self,
        *,
        after_seq: int = 0,
        session_id: str | None = None,
        operation_id: str | None = None,
        limit: int | None = None,
    ) -> list[EventEnvelope]:
        await self.initialize()
        effective_limit = self.max_replay_events if limit is None else min(
            max(limit, 0), self.max_replay_events
        )
        async with self._lock:
            events = await asyncio.to_thread(self._read_sync)
        result = [
            event
            for event in events
            if event.seq > after_seq
            and (session_id is None or event.session_id == session_id)
            and (operation_id is None or event.operation_id == operation_id)
        ]
        return result[:effective_limit]

    @asynccontextmanager
    async def subscribe(
        self,
        *,
        after_seq: int = 0,
        session_id: str | None = None,
        operation_id: str | None = None,
    ) -> AsyncIterator[AsyncIterator[EventEnvelope]]:
        """Yield replay first, then live events without a replay/live race."""

        await self.initialize()
        queue: asyncio.Queue[EventEnvelope] = asyncio.Queue()
        async with self._lock:
            existing = await asyncio.to_thread(self._read_sync)
            replay = [
                event
                for event in existing
                if event.seq > after_seq
                and (session_id is None or event.session_id == session_id)
                and (operation_id is None or event.operation_id == operation_id)
            ][: self.max_replay_events]
            self._subscribers.add(queue)

        async def iterator() -> AsyncIterator[EventEnvelope]:
            for event in replay:
                yield event
            while True:
                event = await queue.get()
                if session_id is not None and event.session_id != session_id:
                    continue
                if operation_id is not None and event.operation_id != operation_id:
                    continue
                yield event

        try:
            yield iterator()
        finally:
            self._subscribers.discard(queue)

    def _read_sync(self) -> list[EventEnvelope]:
        if not self.path.exists():
            return []
        events: list[EventEnvelope] = []
        with self.path.open("r", encoding="utf-8") as handle:
            for line_number, line in enumerate(handle, start=1):
                if not line.strip():
                    continue
                try:
                    events.append(EventEnvelope.model_validate_json(line))
                except Exception as exc:
                    raise EventStoreCorrupt(
                        f"Invalid event log record at {self.path}:{line_number}"
                    ) from exc
        previous = 0
        for event in events:
            if event.seq <= previous:
                raise EventStoreCorrupt("Event sequence must be strictly increasing")
            previous = event.seq
        return events

    def _append_sync(self, event: EventEnvelope) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        record = event.model_dump_json(exclude_none=False)
        with self.path.open("a", encoding="utf-8", newline="\n") as handle:
            handle.write(record)
            handle.write("\n")
            handle.flush()


def event_to_sse(event: EventEnvelope) -> str:
    data = event.model_dump_json(exclude_none=False)
    return f"id: {event.seq}\nevent: {event.type}\ndata: {data}\n\n"