"""Caller-owned registration for every bootstrap resource allocation."""

from __future__ import annotations

import errno
import fcntl
import os
from collections.abc import Callable
from contextlib import ExitStack
from threading import RLock
from types import TracebackType
from typing import Literal, ParamSpec, TypeVar, final

_P = ParamSpec("_P")
_T = TypeVar("_T")
_FD_LIFECYCLE_LOCK = RLock()
FD_LIFECYCLE_LOCK = _FD_LIFECYCLE_LOCK
_PendingFailure = tuple[BaseException, TracebackType | None]


@final
class _CloseOnce:
    __slots__ = ("_close", "_closed", "_pending_failure", "_retry")

    def __init__(
        self, close: Callable[[], None], *, retry: bool = False,
        pending_failure: Callable[[], _PendingFailure | None] | None = None,
    ) -> None:
        self._close = close
        self._closed = False
        self._pending_failure = pending_failure
        self._retry = retry

    def __call__(self) -> None:
        if self._closed:
            return
        pending: tuple[BaseException, TracebackType | None] | None = None
        while not self._closed:
            try:
                self._close()
                self._closed = True
            except BaseException as error:
                if not self._retry:
                    raise
                if pending is None:
                    pending = error, error.__traceback__
                if self._pending_failure is not None:
                    authoritative = self._pending_failure()
                    if authoritative is not None:
                        pending = authoritative
                        self._closed = True
        if pending is not None:
            raise pending[0].with_traceback(pending[1]) from None


@final
class _OwnedFd:
    __slots__ = ("fd", "_close_descriptor", "_pending", "_state")

    def __init__(
        self, close_descriptor: Callable[[int], None], *,
        initial_state: Literal["open", "uncertain"] = "open",
    ) -> None:
        self.fd = -1
        self._close_descriptor = close_descriptor
        self._pending: _PendingFailure | None = None
        self._state: Literal["open", "uncertain", "closed"] = initial_state

    def close(self) -> None:
        # Every caller enters the process-wide lifecycle lock before dispatching
        # this callback, so a closed number cannot be registrar-reallocated.
        if self._state == "open" and self.fd < 0:
            self._state = "closed"
        while self._state != "closed":
            try:
                if self._state == "uncertain":
                    if self.fd < 0:
                        self._state = "closed"
                        continue
                    try:
                        _ = fcntl.fcntl(self.fd, fcntl.F_GETFD)
                    except OSError as probe_error:
                        if probe_error.errno == errno.EBADF:
                            self.fd = -1
                            self._state = "closed"
                            continue
                        raise
                    self._state = "open"
                self._state = "uncertain"
                self._close_descriptor(self.fd)
                self.fd = -1
                self._state = "closed"
            except BaseException as error:
                if self._pending is None:
                    self._pending = error, error.__traceback__
        if self._pending is not None:
            raise self._pending[0].with_traceback(self._pending[1]) from None

    def take_pending_failure(self) -> _PendingFailure | None:
        if self._state != "closed":
            return None
        pending = self._pending
        self._pending = None
        return pending


CloseGuard = _CloseOnce
OwnedDescriptor = _OwnedFd


@final
class ResourceRegistrar:
    """One idempotent reverse-order owner shared by all nested factories."""

    __slots__ = ("_fds", "_resources")

    def __init__(self) -> None:
        self._resources = ExitStack()
        self._fds: dict[int, _OwnedFd] = {}

    def _register_guard(self, guard: _CloseOnce) -> None:
        registered = False
        try:
            _ = self._resources.callback(guard)
            registered = True
        finally:
            if not registered:
                guard()

    def callback(
        self, close: Callable[_P, None], *args: _P.args, **kwargs: _P.kwargs
    ) -> None:
        guard = _CloseOnce(lambda: close(*args, **kwargs))
        self._register_guard(guard)

    def adopt(
        self, factory: Callable[[], _T], close: Callable[[_T], None]
    ) -> _T:
        resources: list[_T] = []
        guard: _CloseOnce | None = None
        try:
            resources.append(factory())
            guard = _CloseOnce(lambda: close(resources[0]))
            self.callback(guard)
        except BaseException:
            if guard is not None:
                guard()
            elif resources:
                close(resources[0])
            raise
        return resources[0]

    def open_fd(self, opener: Callable[[], int]) -> int:
        owned = _OwnedFd(os.close)
        with _FD_LIFECYCLE_LOCK:
            try:
                owned.fd = opener()
                guard = _CloseOnce(
                    owned.close, retry=True,
                    pending_failure=owned.take_pending_failure,
                )
                self._register_guard(guard)
                self._fds[owned.fd] = owned
            except BaseException:
                owned.close()
                raise
        return owned.fd

    def close_fd(self, descriptor: int) -> None:
        with _FD_LIFECYCLE_LOCK:
            owned = self._fds.pop(descriptor, None)
            if owned is not None:
                owned.close()

    def close(self) -> None:
        with _FD_LIFECYCLE_LOCK:
            self._resources.close()
            self._fds.clear()