coverage_reservations

class CoverageReservations:
    """Hands each user a distinct location on a lease, recycling it when it lapses.

    Construction claims ``scope``: it is stamped with this class's persisted
    format version, and any state written in another shape — including by an
    unrelated consumer — is cleared and rebuilt rather than misread
    ([ensure_format][capturegraph.scheduling.organize.scope.ensure_format]).

    Args:
        scope: Where the batch and reservations persist, isolated to this
            scheduler's use (e.g. ``state.scope("coverage")``).
        distance: A distance between two ``cg.Location``s — the same metric used
            to select the batch (``cgsh.distance.location(sigma_m=...)``). It both
            picks the location closest to a requester and decides when a capture
            covers a reservation.
        threshold: A reservation is released once a capture lies within this
            distance of it, in the units ``distance`` returns (e.g. ``1.0`` is one
            ``sigma_m``). Defaults to ``1.0``.
        lease: How long a handout is held before, absent a covering capture, the
            location returns to the batch for reallocation. Defaults to 30
            minutes.
    """

    def __init__(
        self,
        scope: Scope,
        *,
        distance: Callable[[cg.Location, cg.Location], float],
        threshold: float = 1.0,
        lease: timedelta = timedelta(minutes=30),
    ) -> None:
        """Claim ``scope``, resetting it if its stamped format has changed."""
        ensure_format(scope, _FORMAT)
        self._scope = scope
        self._held = scope.scope("held")
        self._distance = distance
        self._threshold = threshold
        self._lease = lease

    @property
    def exhausted(self) -> bool:
        """Whether the batch has no locations left to hand out."""
        return not self._scope.get("batch")

    def refill(self, locations: Iterable[cg.Location]) -> None:
        """Replace the batch with ``locations``, kept in the order given.

        Hand-out order is decided per request by proximity to the asker
        (see [reserve][]), so the batch itself needs no ordering.
        """
        self._scope["batch"] = [cg.Location.encode(location) for location in locations]

    def reserve(
        self,
        user_id: str,
        *,
        at: cg.Location | None = None,
        fulfilled_by: Sequence[cg.Location] = (),
        now: datetime | None = None,
    ) -> cg.Location | None:
        """The location this user holds, leasing one from the batch if none.

        Args:
            user_id: The user asking for a location.
            at: The requester's current location, if known. The batch location
                closest to it is leased; without it, the batch is leased in
                insertion order (FIFO).
            fulfilled_by: Captured locations; a reservation within ``threshold``
                of any of these is released first, freeing it for reuse.
            now: The reference time; defaults to ``datetime.now()``. Pass it
                explicitly to make lease expiry deterministic under test.

        Returns:
            The user's reserved location, or ``None`` if the batch is exhausted
            and the user holds nothing.
        """
        now = now if now is not None else datetime.now()
        self._release_covered(fulfilled_by)
        self._reclaim_lapsed(now)

        if user_id in self._held:
            return cg.Location.decode(_as_lease(self._held[user_id])["location"])

        batch = _as_batch(self._scope.get("batch", []))
        if not batch:
            return None

        index = self._closest_index(batch, at) if at is not None else 0
        chosen = batch[index]
        self._scope["batch"] = batch[:index] + batch[index + 1 :]
        self._held[user_id] = self._lease_entry(chosen, now)
        return cg.Location.decode(chosen)

    def release(self, user_id: str) -> None:
        """Release a user's reservation without returning its location to the batch."""
        if user_id in self._held:
            del self._held[user_id]

    def _lease_entry(self, encoded: JSONValue, now: datetime) -> dict[str, JSONValue]:
        """A held-lease record: the encoded location and when its hold expires."""
        return {"location": encoded, "expires": (now + self._lease).isoformat()}

    def _release_covered(self, captures: Sequence[cg.Location]) -> None:
        """Retire reservations a capture has already covered (the spot is done)."""
        if not captures:
            return
        for user_id, entry in self._held.items():
            location = cg.Location.decode(_as_lease(entry)["location"])
            if any(self._distance(capture, location) < self._threshold for capture in captures):
                del self._held[user_id]

    def _reclaim_lapsed(self, now: datetime) -> None:
        """Return locations whose lease has lapsed to the batch for reallocation."""
        reclaimed: list[JSONValue] = []
        for user_id, entry in self._held.items():
            lease = _as_lease(entry)
            expires = lease["expires"]
            if not isinstance(expires, str):
                raise TypeError(
                    f"expected an ISO-8601 expiry in scope, got {type(expires).__name__}"
                )
            if datetime.fromisoformat(expires) <= now:
                reclaimed.append(lease["location"])
                del self._held[user_id]
        if reclaimed:
            self._scope["batch"] = _as_batch(self._scope.get("batch", [])) + reclaimed

    def _closest_index(self, batch: list[JSONValue], at: cg.Location) -> int:
        """Index of the batch entry nearest ``at`` under ``self._distance``."""
        return min(
            range(len(batch)),
            key=lambda index: self._distance(at, cg.Location.decode(batch[index])),
        )