from __future__ import annotations

import ctypes
import hashlib
import os
import secrets
from pathlib import Path
from typing import Any, Callable, Mapping

from .constants import PENDING_SCHEMA
from .strict_json import canonical_bytes, loads


class DurabilityError(RuntimeError):
    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code


def fsync_directory(path: Path) -> None:
    if os.name != "nt":
        fd = os.open(path, os.O_RDONLY)
        try:
            os.fsync(fd)
        finally:
            os.close(fd)
        return
    create_file = ctypes.windll.kernel32.CreateFileW
    handle = create_file(str(path), 0x40000000, 0x7, None, 3, 0x02000000, None)
    if handle == ctypes.c_void_p(-1).value:
        raise OSError(ctypes.get_last_error(), "CreateFileW directory failed")
    try:
        if not ctypes.windll.kernel32.FlushFileBuffers(handle):
            raise OSError(ctypes.get_last_error(), "FlushFileBuffers directory failed")
    finally:
        ctypes.windll.kernel32.CloseHandle(handle)


class PendingStore:
    def __init__(self, path: Path, boundary_hook: Callable[[str], None] | None = None) -> None:
        self.path = path
        self._hook = boundary_hook or (lambda _phase: None)

    def _boundary(self, name: str) -> None:
        self._hook(name)

    def load(self) -> dict[str, Any] | None:
        if not self.path.exists():
            return None
        try:
            value = loads(self.path.read_bytes())
        except Exception as exc:
            raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending state is unreadable") from exc
        required = {
            "schema", "run_id", "request_id", "phase", "action_budget_consumed", "may_have_dispatched",
            "refresh_count", "retry_count", "permit_id", "permit_payload_sha256", "deadline_at",
        }
        if not isinstance(value, dict) or set(value) != required or value["schema"] != PENDING_SCHEMA:
            raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending state schema differs")
        return value

    def write(self, value: Mapping[str, Any]) -> None:
        payload = canonical_bytes(dict(value), newline=True)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        partial = self.path.parent / f".{self.path.name}.{secrets.token_hex(8)}.partial"
        self._boundary("before_open")
        binary = getattr(os, "O_BINARY", 0)
        fd = os.open(partial, os.O_CREAT | os.O_EXCL | os.O_WRONLY | binary, 0o600)
        try:
            self._boundary("after_open")
            view = memoryview(payload)
            while view:
                written = os.write(fd, view)
                if written <= 0:
                    raise OSError("short pending write")
                view = view[written:]
            self._boundary("after_write")
            os.fsync(fd)
            self._boundary("after_file_fsync")
        finally:
            os.close(fd)
        os.replace(partial, self.path)
        self._boundary("after_replace")
        if self.path.read_bytes() != payload:
            raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending readback differs")
        self._boundary("after_readback")
        fsync_directory(self.path.parent)
        self._boundary("after_directory_fsync")
        if self.path.read_bytes() != payload:
            raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending second readback differs")
        self._boundary("after_second_readback")

    def initialize(self, *, run_id: str, request_id: str, deadline_at: str) -> dict[str, Any]:
        value = {
            "schema": PENDING_SCHEMA,
            "run_id": run_id,
            "request_id": request_id,
            "phase": "PREPARED",
            "action_budget_consumed": False,
            "may_have_dispatched": False,
            "refresh_count": 0,
            "retry_count": 0,
            "permit_id": None,
            "permit_payload_sha256": None,
            "deadline_at": deadline_at,
        }
        self.write(value)
        return value

    def consume(self, value: Mapping[str, Any], *, permit_id: str, permit_payload: bytes) -> dict[str, Any]:
        if value.get("phase") != "PREPARED" or value.get("action_budget_consumed") is not False:
            raise DurabilityError("E_DISPATCH_REPLAY", "action budget is not available")
        updated = dict(value)
        updated.update(
            phase="ACTION_BUDGET_CONSUMED",
            action_budget_consumed=True,
            may_have_dispatched=True,
            refresh_count=1,
            retry_count=0,
            permit_id=permit_id,
            permit_payload_sha256=hashlib.sha256(permit_payload).hexdigest(),
        )
        self.write(updated)
        return updated

    def recovery_projection(self) -> dict[str, Any]:
        try:
            value = self.load()
        except DurabilityError:
            return {"error_code": "E_DISPATCH_DURABILITY_AMBIGUOUS", "refresh_count": 1, "retry_count": 0, "may_have_dispatched": True}
        if value is None or value["action_budget_consumed"] is False:
            return {"error_code": "E_PRE_PERMIT_STOP", "refresh_count": 0, "retry_count": 0, "may_have_dispatched": False}
        return {"error_code": "E_POST_PERMIT_UNCERTAIN", "refresh_count": 1, "retry_count": 0, "may_have_dispatched": True}