"""Small, dependency-free persistence primitives for dashboard state."""

from __future__ import annotations

import json
import os
import tempfile
import threading
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator

try:
    import fcntl
except ImportError:  # pragma: no cover - Windows fallback uses the process lock
    fcntl = None


_LOCKS: dict[str, threading.RLock] = {}
_LOCKS_GUARD = threading.Lock()
_METADATA_KEYS = {"observed_at", "source", "stale", "last_error"}


def _thread_lock(path: Path) -> threading.RLock:
    key = str(path.resolve())
    with _LOCKS_GUARD:
        return _LOCKS.setdefault(key, threading.RLock())


@contextmanager
def file_lock(path: str | os.PathLike[str]) -> Iterator[None]:
    """Serialize readers/writers across threads and, on POSIX, processes."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    lock_path = target.with_name(f".{target.name}.lock")
    with _thread_lock(target):
        with lock_path.open("a+b") as lock_file:
            if fcntl is not None:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                if fcntl is not None:
                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)


def atomic_write_json(path: str | os.PathLike[str], data: Any, *, indent: int = 2) -> None:
    """Durably replace a JSON file without exposing a partial document."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    with file_lock(target):
        fd, temporary = tempfile.mkstemp(
            prefix=f".{target.name}.", suffix=".tmp", dir=target.parent
        )
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(data, handle, indent=indent, default=str)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, target)
            try:
                directory_fd = os.open(target.parent, os.O_DIRECTORY)
                try:
                    os.fsync(directory_fd)
                finally:
                    os.close(directory_fd)
            except (AttributeError, OSError):
                pass
        finally:
            try:
                os.unlink(temporary)
            except FileNotFoundError:
                pass


def load_json(path: str | os.PathLike[str], default: Any = None) -> Any:
    target = Path(path)
    if not target.exists():
        return default
    with file_lock(target):
        try:
            with target.open(encoding="utf-8") as handle:
                return json.load(handle)
        except (OSError, ValueError):
            return default


def _tail_bytes(target: Path, *, line_hint: int, chunk_size: int) -> bytes:
    with target.open("rb") as handle:
        handle.seek(0, os.SEEK_END)
        position = handle.tell()
        blocks: list[bytes] = []
        newlines = 0
        while position > 0 and newlines <= line_hint:
            size = min(chunk_size, position)
            position -= size
            handle.seek(position)
            block = handle.read(size)
            blocks.append(block)
            newlines += block.count(b"\n")
        return b"".join(reversed(blocks))


def load_jsonl_tail(
    path: str | os.PathLike[str], *, limit: int = 90, chunk_size: int = 8192
) -> list[dict[str, Any]]:
    """Read only enough of a JSONL file to return its last valid entries."""
    if limit <= 0:
        return []
    target = Path(path)
    if not target.exists():
        return []
    with file_lock(target):
        raw = _tail_bytes(target, line_hint=limit + 8, chunk_size=max(chunk_size, 32))
    entries: list[dict[str, Any]] = []
    for line in raw.splitlines():
        try:
            value = json.loads(line)
        except (UnicodeDecodeError, ValueError):
            continue
        if isinstance(value, dict):
            entries.append(value)
    return entries[-limit:]


def _utc_day(timestamp: Any) -> str | None:
    if not isinstance(timestamp, str):
        return None
    try:
        parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed.astimezone(timezone.utc).date().isoformat()
    except ValueError:
        return None


def append_daily_jsonl(path: str | os.PathLike[str], entry: dict[str, Any]) -> bool:
    """Append at most one record per UTC day, inspecting only the bounded tail."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    entry_day = _utc_day(entry.get("timestamp"))
    if entry_day is None:
        raise ValueError("entry timestamp must be an ISO-8601 datetime")
    with file_lock(target):
        if target.exists():
            raw = _tail_bytes(target, line_hint=8, chunk_size=4096)
            for line in reversed(raw.splitlines()):
                try:
                    previous = json.loads(line)
                except (UnicodeDecodeError, ValueError):
                    continue
                if _utc_day(previous.get("timestamp")) == entry_day:
                    return False
                break
        payload = (json.dumps(entry, default=str) + "\n").encode("utf-8")
        fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
        try:
            os.write(fd, payload)
            os.fsync(fd)
        finally:
            os.close(fd)
    return True


def _has_observation(payload: Any) -> bool:
    if not isinstance(payload, dict):
        return payload is not None
    return any(value is not None for key, value in payload.items() if key not in _METADATA_KEYS)


def merge_observation(
    previous: Any,
    observed: Any,
    *,
    source: str,
    observed_at: str | None = None,
    error: str | None = None,
) -> dict[str, Any]:
    """Annotate a fresh observation or retain the last-known-good value as stale."""
    if _has_observation(observed):
        merged = dict(observed) if isinstance(observed, dict) else {"value": observed}
        merged.update(
            observed_at=observed_at or datetime.now(timezone.utc).isoformat(),
            source=source,
            stale=False,
            last_error=None,
        )
        return merged

    merged = dict(previous) if isinstance(previous, dict) else {}
    merged.update(
        source=merged.get("source") or source,
        stale=True,
        last_error=error or "metric was not observed",
    )
    merged.setdefault("observed_at", None)
    return merged


def onchain_refresh_due(
    timestamp: Any,
    *,
    now: datetime | None = None,
    ttl_seconds: int = 6 * 60 * 60,
) -> bool:
    """Return whether the last successful on-chain observation exceeded its TTL."""
    if not isinstance(timestamp, str) or not timestamp:
        return True
    try:
        observed = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
        if observed.tzinfo is None:
            observed = observed.replace(tzinfo=timezone.utc)
    except ValueError:
        return True
    current = now or datetime.now(timezone.utc)
    if current.tzinfo is None:
        current = current.replace(tzinfo=timezone.utc)
    return (current.astimezone(timezone.utc) - observed.astimezone(timezone.utc)).total_seconds() >= ttl_seconds