"""FileSink -- the always-on persistence sink.

Every session, in every mode, installs a FileSink.  It appends each event to
``<session_dir>/events.jsonl`` and atomically rewrites
``<session_dir>/state.json`` (tmp file + ``os.replace``) so status readers
never see a torn state.  All writes happen under an exclusive lock on
``<session_dir>/.lock`` -- the same lock ``spens cancel`` takes for its
read-modify-write -- which is what makes the first-writer-wins terminal
state race-free.  The lock itself is platform-specific but equivalent:
``fcntl.flock`` on POSIX, ``msvcrt.locking`` on Windows (see
:func:`session_lock`).

The state machine from :mod:`spens.events` is enforced here: legal
transitions are applied, illegal ones refused, and an existing terminal
state is never overwritten.
"""

from __future__ import annotations

import errno
import json
import os
import sys
import time
from contextlib import contextmanager
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from spens.events import EVENT_STATES, Event, is_terminal, is_valid_transition

if sys.platform == "win32":
    import msvcrt

    def _lock_handle(fh) -> None:
        """Take the exclusive lock, waiting as long as a POSIX flock would.

        ``msvcrt.locking`` locks a byte range starting at the current file
        position, so every taker locks byte 0.  ``LK_LOCK`` only retries
        internally for ~10 seconds before failing, so contention longer
        than that is retried here.
        """
        fh.seek(0)
        while True:
            try:
                msvcrt.locking(fh.fileno(), msvcrt.LK_LOCK, 1)
                return
            except OSError as exc:
                # EACCES/EDEADLK: still held by someone else -- keep waiting.
                if exc.errno not in (errno.EACCES, errno.EDEADLK):
                    raise
                time.sleep(0.05)

    def _unlock_handle(fh) -> None:
        fh.seek(0)
        msvcrt.locking(fh.fileno(), msvcrt.LK_UNLCK, 1)

else:
    import fcntl

    def _lock_handle(fh) -> None:
        fcntl.flock(fh, fcntl.LOCK_EX)

    def _unlock_handle(fh) -> None:
        fcntl.flock(fh, fcntl.LOCK_UN)


STATE_FILENAME = "state.json"
EVENTS_FILENAME = "events.jsonl"
LOCK_FILENAME = ".lock"


@contextmanager
def session_lock(session_dir: Path):
    """Exclusive lock over one session's state files (events.jsonl/state.json).

    Shared with ``spens cancel`` so external writers and the running session
    serialize their read-modify-write cycles.  ``fcntl.flock`` on POSIX;
    ``msvcrt.locking`` on byte 0 of the lock file on Windows.
    """
    session_dir = Path(session_dir)
    session_dir.mkdir(parents=True, exist_ok=True)
    fh = open(session_dir / LOCK_FILENAME, "a+")  # noqa: SIM115 -- closed manually below
    try:
        _lock_handle(fh)
    except BaseException:
        fh.close()
        raise
    try:
        yield
    finally:
        try:
            _unlock_handle(fh)
        finally:
            fh.close()


def read_state(session_dir: Path) -> dict[str, Any] | None:
    """Read ``state.json``, returning None when missing or malformed."""
    try:
        with open(Path(session_dir) / STATE_FILENAME, encoding="utf-8") as fh:
            return json.load(fh)
    except (OSError, json.JSONDecodeError):
        return None


def _utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


class FileSink:
    def __init__(self, session_dir: Path, base: dict[str, Any] | None = None) -> None:
        self.session_dir = Path(session_dir)
        self.events_path = self.session_dir / EVENTS_FILENAME
        self.state_path = self.session_dir / STATE_FILENAME
        #: Metadata written for a fresh session (session_id, prompt, env,
        #: agent, containers, started_at).  An existing state.json wins over
        #: it, so a background-mode child reusing the session keeps the
        #: parent's started_at instead of clobbering it.
        self.base: dict[str, Any] = dict(base or {})

    # -- Sink interface ------------------------------------------------------

    def on_event(self, event: Event) -> None:
        with session_lock(self.session_dir):
            current = read_state(self.session_dir)
            state = self._merge_state(current, event)
            with open(self.events_path, "a", encoding="utf-8") as fh:
                fh.write(event.to_line() + "\n")
            self._write_state_atomic(state)

    def close(self) -> None:
        pass

    # -- internals -----------------------------------------------------------

    def _merge_state(
        self, current: dict[str, Any] | None, event: Event
    ) -> dict[str, Any]:
        """Merge ``event`` into the on-disk state, enforcing the state machine."""
        state: dict[str, Any] = dict(self.base)
        if current:
            state.update(current)
        data = event.data

        if event.event == "agent_started" and data.get("container"):
            state["agent_container"] = data["container"]
        if event.event in ("agent_exited", "finished") and "exit_code" in data:
            state["exit_code"] = data["exit_code"]

        new_state = EVENT_STATES.get(event.event)
        if new_state is not None:
            old_state = state.get("state")
            if old_state is None or not is_terminal(old_state) and (
                new_state == old_state or is_valid_transition(old_state, new_state)
            ):
                state["state"] = new_state
            # else: illegal transition, or the existing terminal state wins
            # (e.g. an external ``spens cancel``); refuse the overwrite.
        elif state.get("state") is None:
            # An event arrived before ``started`` (e.g. a config-validation
            # warning during session scaffolding): the session is being built.
            state["state"] = "building"

        state["updated_at"] = _utc_now_iso()
        return state

    def _write_state_atomic(self, state: dict[str, Any]) -> None:
        tmp_path = self.state_path.with_name(self.state_path.name + ".tmp")
        with open(tmp_path, "w", encoding="utf-8") as fh:
            json.dump(state, fh)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp_path, self.state_path)
