"""Thread-safe circuit breaker whose lock guards ONLY state transitions.

The lock never wraps the wrapped I/O or the state-change callback, so calls run concurrently
(pybreaker serializes; this does not). Half-open admits exactly one probe via a non-blocking
BoundedSemaphore decided atomically under the lock (no TOCTOU). Callbacks fire AFTER the lock is
released; under concurrency they may arrive slightly out of order, so each carries its own
(old, new) pair.
"""

import functools
import logging
import threading
import time
from enum import Enum, auto
from typing import Callable, TypeVar

_log = logging.getLogger(__name__)

T = TypeVar('T')
Transition = tuple[str, str, str]  # (name, old_state, new_state)


class State(Enum):
    """Circuit states."""

    CLOSED = 'closed'
    OPEN = 'open'
    HALF_OPEN = 'halfopen'


class _Admit(Enum):
    """Internal _admit decision: run the call, run it as the single half-open probe, or reject."""

    RUN = auto()
    PROBE = auto()
    REJECT = auto()


class CircuitBreakerError(Exception):
    """Raised when a call is rejected because the circuit is open."""

    def __init__(self, name: str) -> None:
        """Build the error for the named circuit."""
        super().__init__(f"circuit '{name}' is open")
        self.name = name


class CircuitBreaker:
    """A breaker that trips after consecutive failures and admits a single half-open probe."""

    def __init__(
        self,
        name: str,
        fail_max: int,
        reset_timeout: float,
        success_threshold: int,
        count_failure: Callable[[BaseException], bool],
        on_state_change: Callable[[str, str, str], None],
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        """Configure the breaker; on_state_change fires OUTSIDE the lock and may block / do I/O."""
        self.name = name
        self._fail_max = fail_max
        self._reset_timeout = reset_timeout
        self._success_threshold = success_threshold
        self._count_failure = count_failure
        self._on_state_change = on_state_change
        self._clock = clock
        self._lock = threading.Lock()
        # Bounded so an accidental over-release surfaces as an error instead of silently
        # inflating the gate past one permit.
        self._half_open = threading.BoundedSemaphore(1)
        self._state = State.CLOSED
        self._fails = 0
        self._succ = 0
        self._opened_at: float | None = None

    @property
    def state(self) -> str:
        """Current state as a string ('closed' | 'open' | 'halfopen')."""
        return self._state.value

    def _transition(self, new: State) -> Transition:
        """Swap state under the lock and return the change as (name, old, new).

        The caller fires the callback with this tuple after releasing the lock.
        """
        old, self._state = self._state, new
        return (self.name, old.value, new.value)

    def _fire(self, transition: Transition | None) -> None:
        """Run the state-change callback outside the lock.

        A throwing callback must never corrupt the state machine or strand the gate -- but it
        must not vanish either, since on_state_change is the breaker's only observability seam.
        """
        if transition is None:
            return
        try:
            self._on_state_change(*transition)
        except Exception:
            _log.exception("circuit '%s' on_state_change callback failed", self.name)

    def _admit(self) -> tuple[_Admit, Transition | None]:
        with self._lock:
            if self._state is State.CLOSED:
                return _Admit.RUN, None
            if self._state is State.OPEN:
                opened_at = self._opened_at
                if (
                    opened_at is not None
                    and self._clock() - opened_at < self._reset_timeout
                ):
                    return _Admit.REJECT, None
                if self._half_open.acquire(blocking=False):
                    transition = self._transition(State.HALF_OPEN)
                    self._succ = 0
                    return _Admit.PROBE, transition
                return _Admit.REJECT, None
            # HALF_OPEN: admit one probe at a time toward success_threshold
            if self._half_open.acquire(blocking=False):
                return _Admit.PROBE, None
            return _Admit.REJECT, None

    def call(self, fn: Callable[[], T]) -> T:
        """Run fn through the breaker; fast-fail when open, admit one probe when half-open."""
        admit, transition = self._admit()
        self._fire(transition)  # callback OUTSIDE the lock
        if admit is _Admit.REJECT:
            raise CircuitBreakerError(self.name)
        try:
            result = fn()  # I/O OUTSIDE the lock
        except Exception as e:  # control-flow exceptions (KeyboardInterrupt/SystemExit) aren't failures
            try:
                failed = self._count_failure(e)
            except Exception:
                # A buggy classifier must not mask the real downstream error: count it and move on.
                _log.exception(
                    "circuit '%s' count_failure classifier raised", self.name
                )
                failed = True
            if failed:
                self._fire(self._on_failure())
            raise
        else:
            self._fire(self._on_success())
            return result
        finally:
            if admit is _Admit.PROBE:
                self._half_open.release()  # MANDATORY: else the gate leaks -> stuck open

    def __call__(self, fn: Callable[..., T]) -> Callable[..., T]:
        """Wrap `fn` (decorator form) so every call runs through `call`.

        Equivalent to wrapping the body in `breaker.call(...)`; use `.call(fn)` directly when the
        guarded unit is a one-off closure rather than a whole function.
        """

        @functools.wraps(fn)
        def wrapper(*args: object, **kwargs: object) -> T:
            return self.call(lambda: fn(*args, **kwargs))

        return wrapper

    def _on_failure(self) -> Transition | None:
        with self._lock:
            if self._state is State.OPEN:
                # Stragglers that fail after the breaker already tripped must not re-open it
                # (which would slide _opened_at forward and starve the half-open probe).
                return None
            if self._state is State.HALF_OPEN:
                transition = self._transition(State.OPEN)
                self._opened_at = self._clock()
                self._fails = self._succ = 0
                return transition
            self._fails += 1
            if self._fails >= self._fail_max:
                transition = self._transition(State.OPEN)
                self._opened_at = self._clock()
                self._fails = 0
                return transition
            return None

    def _on_success(self) -> Transition | None:
        with self._lock:
            if self._state is State.HALF_OPEN:
                self._succ += 1
                if self._succ >= self._success_threshold:
                    transition = self._transition(State.CLOSED)
                    self._fails = self._succ = 0
                    return transition
            else:
                self._fails = 0
            return None