"""Private durable addresses and content-free delivery projections for check-ins."""

from __future__ import annotations

import errno
import json
import os
import re
import secrets
import stat
import time
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Final, cast, final

import fcntl


_MODE: Final[int] = 0o600
_DIR_MODE: Final[int] = 0o700
_MAX_STATE_BYTES: Final[int] = 256 * 1024
_FILE_FLAGS: Final[int] = os.O_NOFOLLOW | os.O_CLOEXEC
_DIRECTORY_FLAGS: Final[int] = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
_TELEGRAM_ID_MAX: Final[int] = (1 << 63) - 1
_DOMAIN_CURSOR_SESSION_ID: Final[re.Pattern[str]] = re.compile(r"^[a-f0-9]{32}$")
_DOMAIN_CURSOR_STEPS: Final[frozenset[str]] = frozenset({
    "launch", "bodyweight", "sleep_duration", "sleep_quality", "condition", "pain",
    "calories", "macros", "meals", "water", "digestion", "appetite_stress",
    "training_plan", "weight_change_percent", "fatigue", "exercise_feasibility",
    "meal_deviation", "wc", "ft", "ef", "md", "optional_note", "summary",
    "edit_menu", "completion", "training_summary", "workout_quality", "done",
    "performance", "intensity", "operator_note", "safety_ack", "Q-SLEEP-CAUSE",
    "Q-SLEEP-ADJUST", "Q-COND-SYMPTOM", "Q-COND-INTENSITY", "Q-PERF-REASON",
    "Q-PERF-NEXT",
})


class BindingStoreError(RuntimeError):
    """The private binding projection store could not be used safely."""


class BindingStoreCorruption(BindingStoreError):
    """Persisted binding or projection state is malformed or unsafe."""


class BindingStoreConflict(BindingStoreError):
    """An ingress or cursor would make the projection state ambiguous."""


class IngressKind(str, Enum):
    """The opaque Telegram ingress surface that owns one projection."""

    CALLBACK = "callback"
    TEXT = "text"


class ProjectionPhase(str, Enum):
    """Durable state of one local domain-to-Telegram publication attempt."""

    PREPARED = "prepared"
    DOMAIN_COMMITTED = "domain_committed"
    SEND_STARTED = "send_started"
    DELIVERED = "delivered"
    DELIVERY_FAILED = "delivery_failed"
    DELIVERY_UNCERTAIN = "delivery_uncertain"


_NONTERMINAL_PHASES: Final[frozenset[ProjectionPhase]] = frozenset({
    ProjectionPhase.PREPARED,
    ProjectionPhase.DOMAIN_COMMITTED,
    ProjectionPhase.SEND_STARTED,
})
_NEXT_PHASES: Final[dict[ProjectionPhase, frozenset[ProjectionPhase]]] = {
    ProjectionPhase.PREPARED: frozenset({ProjectionPhase.DOMAIN_COMMITTED}),
    ProjectionPhase.DOMAIN_COMMITTED: frozenset({
        ProjectionPhase.SEND_STARTED,
        ProjectionPhase.DELIVERY_FAILED,
        ProjectionPhase.DELIVERY_UNCERTAIN,
    }),
    ProjectionPhase.SEND_STARTED: frozenset({
        ProjectionPhase.DELIVERED,
        ProjectionPhase.DELIVERY_FAILED,
        ProjectionPhase.DELIVERY_UNCERTAIN,
    }),
}


def _object_dict(value: object) -> dict[str, object] | None:
    if not isinstance(value, dict):
        return None
    return cast(dict[str, object], value)


def _object_list(value: object) -> list[object] | None:
    if not isinstance(value, list):
        return None
    return cast(list[object], value)


def _legacy_nonempty_string(value: object) -> str | None:
    return value if isinstance(value, str) and value and len(value) <= 256 else None


def _bounded_int(value: object, minimum: int) -> int | None:
    return value if type(value) is int and minimum <= value <= _TELEGRAM_ID_MAX else None


def _nonnegative_int(value: object) -> int | None:
    return _bounded_int(value, 0)


def _positive_int(value: object) -> int | None:
    return _bounded_int(value, 1)


def _telegram_chat_id(value: object) -> int | None:
    return value if type(value) is int and -_TELEGRAM_ID_MAX <= value <= _TELEGRAM_ID_MAX and value != 0 else None


def _domain_session_id(value: object) -> str | None:
    return value if isinstance(value, str) and _DOMAIN_CURSOR_SESSION_ID.fullmatch(value) else None


def _domain_step(value: object) -> str | None:
    return value if isinstance(value, str) and value in _DOMAIN_CURSOR_STEPS else None


def _domain_cursor_version(value: object) -> int | None:
    return value if type(value) is int and 0 <= value <= 1_000_000 else None


@dataclass(frozen=True, slots=True)
class CursorIdentity:
    """One value-free wizard cursor before or after an ingress mutation."""

    session_id: str
    step: str
    version: int

    def __post_init__(self) -> None:
        if (
            _domain_session_id(self.session_id) is None
            or _domain_step(self.step) is None
            or _domain_cursor_version(self.version) is None
        ):
            raise ValueError("invalid Telegram projection cursor")

    def to_dict(self) -> dict[str, str | int]:
        return {"session_id": self.session_id, "step": self.step, "version": self.version}

    @classmethod
    def from_dict(cls, value: object) -> CursorIdentity | None:
        record = _object_dict(value)
        if record is None or set(record) != {"session_id", "step", "version"}:
            return None
        session_id = _domain_session_id(record["session_id"])
        step = _domain_step(record["step"])
        version = _domain_cursor_version(record["version"])
        if session_id is None or step is None or version is None:
            return None
        return cls(session_id, step, version)


@dataclass(frozen=True, slots=True)
class IngressIdentity:
    """Opaque Telegram update/message identity, never callback or answer content."""

    update_id: int
    kind: IngressKind
    message_id: int
    actor_id: int
    chat_id: int
    topic_id: int

    def __post_init__(self) -> None:
        if (
            _nonnegative_int(self.update_id) is None
            or type(self.kind) is not IngressKind
            or _positive_int(self.message_id) is None
            or _positive_int(self.actor_id) is None
            or _telegram_chat_id(self.chat_id) is None
            or _nonnegative_int(self.topic_id) is None
        ):
            raise ValueError("invalid Telegram projection ingress")

    def to_dict(self) -> dict[str, str | int]:
        return {
            "update_id": self.update_id,
            "kind": self.kind.value,
            "message_id": self.message_id,
            "actor_id": self.actor_id,
            "chat_id": self.chat_id,
            "topic_id": self.topic_id,
        }

    @classmethod
    def from_dict(cls, value: object) -> IngressIdentity | None:
        expected = {"update_id", "kind", "message_id", "actor_id", "chat_id", "topic_id"}
        record = _object_dict(value)
        if record is None or set(record) != expected:
            return None
        update_id = _nonnegative_int(record["update_id"])
        kind = record["kind"]
        message_id = _positive_int(record["message_id"])
        actor_id = _positive_int(record["actor_id"])
        chat_id = _telegram_chat_id(record["chat_id"])
        topic_id = _nonnegative_int(record["topic_id"])
        if (
            update_id is None
            or not isinstance(kind, str)
            or message_id is None
            or actor_id is None
            or chat_id is None
            or topic_id is None
        ):
            return None
        try:
            return cls(update_id, IngressKind(kind), message_id, actor_id, chat_id, topic_id)
        except ValueError:
            return None


@dataclass(frozen=True, slots=True)
class TelegramProjection:
    """A content-free, repairable local projection of a Telegram publication."""

    ingress: IngressIdentity
    source: CursorIdentity
    target: CursorIdentity
    phase: ProjectionPhase
    expires_at: int
    receipt_message_id: int | None = None

    def __post_init__(self) -> None:
        receipt = self.receipt_message_id
        if (
            type(self.ingress) is not IngressIdentity
            or type(self.source) is not CursorIdentity
            or type(self.target) is not CursorIdentity
            or type(self.phase) is not ProjectionPhase
            or _positive_int(self.expires_at) is None
            or self.source.session_id != self.target.session_id
            or self.target.version != self.source.version + 1
            or (self.phase is ProjectionPhase.DELIVERED) != (receipt is not None)
            or (receipt is not None and _positive_int(receipt) is None)
        ):
            raise ValueError("invalid Telegram projection")

    @property
    def session_id(self) -> str:
        return self.source.session_id

    def to_dict(self) -> dict[str, object]:
        return {
            "ingress": self.ingress.to_dict(),
            "source_cursor": self.source.to_dict(),
            "target_cursor": self.target.to_dict(),
            "phase": self.phase.value,
            "expires_at": self.expires_at,
            "receipt_message_id": self.receipt_message_id,
        }

    @classmethod
    def from_dict(cls, value: object) -> TelegramProjection | None:
        expected = {
            "ingress", "source_cursor", "target_cursor", "phase", "expires_at", "receipt_message_id",
        }
        record = _object_dict(value)
        if record is None or set(record) != expected:
            return None
        ingress = IngressIdentity.from_dict(record["ingress"])
        source = CursorIdentity.from_dict(record["source_cursor"])
        target = CursorIdentity.from_dict(record["target_cursor"])
        phase = record["phase"]
        expires_at = _positive_int(record["expires_at"])
        raw_receipt = record["receipt_message_id"]
        receipt = None if raw_receipt is None else _positive_int(raw_receipt)
        if (
            ingress is None
            or source is None
            or target is None
            or not isinstance(phase, str)
            or expires_at is None
            or raw_receipt is not None and receipt is None
        ):
            return None
        try:
            return cls(ingress, source, target, ProjectionPhase(phase), expires_at, receipt)
        except ValueError:
            return None


@dataclass(slots=True)
class WizardBinding:
    """No health values: only the exact Telegram address of a wizard prompt."""

    session_id: str
    owner_id: str
    chat_id: str
    topic_id: str
    step: str
    version: int
    message_id: str
    expires_at: int
    awaiting_text: bool = False

    @classmethod
    def from_dict(cls, value: object) -> WizardBinding | None:
        record = _object_dict(value)
        if record is None:
            return None
        required = {"session_id", "owner_id", "chat_id", "topic_id", "step", "version", "message_id", "expires_at"}
        allowed = required | {"awaiting_text"}
        if not required.issubset(record) or set(record) - allowed:
            return None
        session_id = _legacy_nonempty_string(record["session_id"])
        owner_id = _legacy_nonempty_string(record["owner_id"])
        chat_id = _legacy_nonempty_string(record["chat_id"])
        topic_id = _legacy_nonempty_string(record["topic_id"])
        step = _legacy_nonempty_string(record["step"])
        awaiting_text = record.get("awaiting_text", False)
        if record["message_id"] == "" and step == "launch" and awaiting_text is False:
            message_id = ""
        else:
            message_id = _legacy_nonempty_string(record["message_id"])
        version = _nonnegative_int(record["version"])
        expires_at = _positive_int(record["expires_at"])
        if (
            session_id is None
            or owner_id is None
            or chat_id is None
            or topic_id is None
            or step is None
            or message_id is None
            or version is None
            or expires_at is None
            or type(awaiting_text) is not bool
        ):
            return None
        return cls(
            session_id, owner_id, chat_id, topic_id, step, version, message_id, expires_at, awaiting_text,
        )

    def to_dict(self) -> dict[str, str | int | bool]:
        return {
            "session_id": self.session_id, "owner_id": self.owner_id, "chat_id": self.chat_id,
            "topic_id": self.topic_id, "step": self.step, "version": self.version,
            "message_id": self.message_id, "expires_at": self.expires_at,
            "awaiting_text": self.awaiting_text,
        }


@dataclass(frozen=True, slots=True)
class _BindingState:
    bindings: tuple[WizardBinding, ...]
    active_session_id: str | None
    projections: tuple[TelegramProjection, ...]


@final
class BindingStore:
    """Atomic owner-only persistence for prompt addresses and UI projections."""

    def __init__(self, path: Path) -> None:
        self._path: Path = Path(path)
        self._lock_path: Path = self._path.with_suffix(self._path.suffix + ".lock")

    def load(
        self, owner_id: str, chat_id: str, topic_id: str, now_epoch: int,
    ) -> tuple[dict[str, WizardBinding], str | None]:
        with self._locked() as directory_fd:
            state = self._read_unlocked(directory_fd)
        bindings = {
            binding.session_id: binding
            for binding in state.bindings
            if binding.expires_at > now_epoch
            and (binding.owner_id, binding.chat_id, binding.topic_id) == (owner_id, chat_id, topic_id)
        }
        active_id = state.active_session_id if state.active_session_id in bindings else None
        return bindings, active_id

    def load_projections(self, now_epoch: int) -> tuple[TelegramProjection, ...]:
        """Return unexpired delivery state without upgrading or rewriting legacy data."""
        if _nonnegative_int(now_epoch) is None:
            raise ValueError("invalid projection read time")
        with self._locked() as directory_fd:
            state = self._read_unlocked(directory_fd)
        return tuple(item for item in state.projections if item.expires_at > now_epoch)

    def save(self, bindings: dict[str, WizardBinding], active_session_id: str | None) -> None:
        """Persist a real binding mutation in v2 while retaining live projections."""
        if type(bindings) is not dict:
            raise ValueError("invalid bindings")
        for session_id, binding in bindings.items():
            if type(binding) is not WizardBinding or session_id != binding.session_id:
                raise ValueError("invalid binding")
        if active_session_id is not None and (
            not active_session_id or active_session_id not in bindings
        ):
            raise ValueError("invalid active binding")
        with self._locked() as directory_fd:
            current = self._read_unlocked(directory_fd)
            validated: list[WizardBinding] = []
            for binding in bindings.values():
                parsed = WizardBinding.from_dict(binding.to_dict())
                if parsed is None:
                    raise ValueError("invalid binding")
                validated.append(parsed)
            retained = _unexpired(current.projections, int(time.time()))
            self._write_unlocked(
                directory_fd,
                _BindingState(tuple(validated), active_session_id, retained),
            )

    def record_projection(
        self, projection: TelegramProjection, *, now_epoch: int | None = None,
    ) -> bool:
        """Atomically deduplicate and advance one local content-free projection."""
        if type(projection) is not TelegramProjection:
            raise ValueError("invalid projection")
        now = int(time.time()) if now_epoch is None else now_epoch
        if _nonnegative_int(now) is None:
            raise ValueError("invalid projection write time")
        with self._locked() as directory_fd:
            state = self._read_unlocked(directory_fd)
            projections = list(_unexpired(state.projections, now))
            same_update = next(
                (item for item in projections if item.ingress.update_id == projection.ingress.update_id),
                None,
            )
            if same_update is not None:
                if same_update.ingress != projection.ingress:
                    raise BindingStoreConflict("Telegram ingress identity conflicts")
                if same_update == projection:
                    return False
                if (
                    same_update.source != projection.source
                    or same_update.target != projection.target
                    or same_update.expires_at != projection.expires_at
                    or projection.phase not in _NEXT_PHASES.get(same_update.phase, frozenset())
                ):
                    raise BindingStoreConflict("Telegram ingress cursor conflicts")
                projections[projections.index(same_update)] = projection
            else:
                session_items = [item for item in projections if item.session_id == projection.session_id]
                if any(item.phase in _NONTERMINAL_PHASES for item in session_items):
                    raise BindingStoreConflict("session already has a nonterminal projection")
                if session_items and projection.source != session_items[-1].target:
                    raise BindingStoreConflict("projection cursor is stale")
                projections.append(projection)
            updated = _BindingState(state.bindings, state.active_session_id, tuple(projections))
            _validate_projection_set(updated.projections)
            self._write_unlocked(directory_fd, updated)
            return True

    @contextmanager
    def _locked(self) -> Generator[int, None, None]:
        directory_fd = self._open_private_parent()
        lock_fd: int | None = None
        try:
            lock_fd = _open_private_file(directory_fd, self._lock_path.name, writable=True, create=True)
            fcntl.flock(lock_fd, fcntl.LOCK_EX)
            _verify_private_file(directory_fd, self._lock_path.name, os.fstat(lock_fd))
            yield directory_fd
        except BindingStoreError:
            raise
        except OSError as exc:
            raise BindingStoreError("binding store lock is unavailable") from exc
        finally:
            if lock_fd is not None:
                try:
                    fcntl.flock(lock_fd, fcntl.LOCK_UN)
                finally:
                    os.close(lock_fd)
            os.close(directory_fd)

    def _open_private_parent(self) -> int:
        directory_fd = _open_private_parent_directory(self._path.parent)
        try:
            info = os.fstat(directory_fd)
            if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
                raise BindingStoreError("binding store parent is unsafe")
            os.fchmod(directory_fd, _DIR_MODE)
            if stat.S_IMODE(os.fstat(directory_fd).st_mode) != _DIR_MODE:
                raise BindingStoreError("binding store parent is unsafe")
            return directory_fd
        except BindingStoreError:
            os.close(directory_fd)
            raise
        except OSError as exc:
            os.close(directory_fd)
            raise BindingStoreError("binding store parent is unavailable") from exc
        except BaseException:
            os.close(directory_fd)
            raise

    def _read_unlocked(self, directory_fd: int) -> _BindingState:
        try:
            descriptor = _open_private_file(
                directory_fd, self._path.name, writable=False, create=False,
            )
        except FileNotFoundError:
            return _BindingState((), None, ())
        try:
            info = os.fstat(descriptor)
            _verify_private_file(directory_fd, self._path.name, info)
            if info.st_size > _MAX_STATE_BYTES:
                raise BindingStoreCorruption("binding store is too large")
            chunks: list[bytes] = []
            remaining = _MAX_STATE_BYTES
            while remaining:
                chunk = os.read(descriptor, remaining)
                if not chunk:
                    break
                chunks.append(chunk)
                remaining -= len(chunk)
            final_info = os.fstat(descriptor)
            _verify_private_file(directory_fd, self._path.name, final_info)
            if final_info.st_size > _MAX_STATE_BYTES:
                raise BindingStoreCorruption("binding store is too large")
        except BindingStoreError:
            raise
        except OSError as exc:
            raise BindingStoreError("binding store is unreadable") from exc
        finally:
            os.close(descriptor)
        try:
            decoded = b"".join(chunks).decode("utf-8")
            raw = cast(object, json.loads(decoded))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise BindingStoreCorruption("binding store is malformed") from exc
        return _parse_state(raw)

    def _write_unlocked(self, directory_fd: int, state: _BindingState) -> None:
        payload = json.dumps(_state_to_v2(state), separators=(",", ":"), sort_keys=True).encode("utf-8")
        temporary_name = f".{self._path.name}.{secrets.token_hex(16)}.tmp"
        descriptor: int | None = None
        replaced = False
        try:
            descriptor = _open_private_file(
                directory_fd, temporary_name, writable=True, create=True,
            )
            _write_all(descriptor, payload)
            os.fsync(descriptor)
            os.close(descriptor)
            descriptor = None
            os.replace(
                temporary_name,
                self._path.name,
                src_dir_fd=directory_fd,
                dst_dir_fd=directory_fd,
            )
            replaced = True
            os.fsync(directory_fd)
        except BindingStoreError:
            raise
        except OSError as exc:
            raise BindingStoreError("binding store atomic write failed") from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            if not replaced:
                try:
                    os.unlink(temporary_name, dir_fd=directory_fd)
                except FileNotFoundError:
                    pass
                except OSError as exc:
                    raise BindingStoreError("binding store temporary cleanup failed") from exc


def _open_private_parent_directory(parent: Path) -> int:
    """Bind each parent component through a no-follow directory descriptor."""
    try:
        if parent.is_absolute():
            directory_fd = os.open(os.sep, _DIRECTORY_FLAGS)
            components = parent.parts[1:]
        else:
            directory_fd = os.open(".", _DIRECTORY_FLAGS)
            components = parent.parts
    except OSError as exc:
        raise BindingStoreError("binding store parent is unavailable") from exc
    try:
        for component in components:
            if component in {"", "."}:
                continue
            if component == "..":
                raise BindingStoreError("binding store parent is unsafe")
            child_fd = _open_or_create_private_directory_component(directory_fd, component)
            os.close(directory_fd)
            directory_fd = child_fd
        return directory_fd
    except BaseException:
        os.close(directory_fd)
        raise


def _open_or_create_private_directory_component(parent_fd: int, name: str) -> int:
    try:
        child_fd = os.open(name, _DIRECTORY_FLAGS, dir_fd=parent_fd)
    except FileNotFoundError:
        try:
            os.mkdir(name, _DIR_MODE, dir_fd=parent_fd)
        except FileExistsError:
            pass
        except OSError as exc:
            raise BindingStoreError("binding store parent is unavailable") from exc
        try:
            child_fd = os.open(name, _DIRECTORY_FLAGS, dir_fd=parent_fd)
        except OSError as exc:
            raise BindingStoreError("binding store parent is unavailable") from exc
    except OSError as exc:
        raise BindingStoreError("binding store parent is unsafe") from exc
    try:
        if not stat.S_ISDIR(os.fstat(child_fd).st_mode):
            raise BindingStoreError("binding store parent is unsafe")
        return child_fd
    except BindingStoreError:
        os.close(child_fd)
        raise
    except OSError as exc:
        os.close(child_fd)
        raise BindingStoreError("binding store parent is unavailable") from exc
    except BaseException:
        os.close(child_fd)
        raise


def _verify_private_file(directory_fd: int, name: str, opened: os.stat_result) -> None:
    try:
        named = os.stat(name, dir_fd=directory_fd, follow_symlinks=False)
    except OSError as exc:
        raise BindingStoreError("binding store file is unavailable") from exc
    if (
        not stat.S_ISREG(opened.st_mode)
        or opened.st_uid != os.getuid()
        or opened.st_nlink != 1
        or stat.S_IMODE(opened.st_mode) != _MODE
        or (opened.st_dev, opened.st_ino) != (named.st_dev, named.st_ino)
        or not stat.S_ISREG(named.st_mode)
        or named.st_uid != os.getuid()
        or named.st_nlink != 1
        or stat.S_IMODE(named.st_mode) != _MODE
    ):
        raise BindingStoreError("binding store file is unsafe")


def _open_private_file(directory_fd: int, name: str, *, writable: bool, create: bool) -> int:
    flags = (os.O_RDWR if writable else os.O_RDONLY) | _FILE_FLAGS
    if create:
        flags |= os.O_CREAT
    try:
        descriptor = os.open(name, flags, _MODE, dir_fd=directory_fd)
    except OSError as exc:
        if not create and exc.errno == errno.ENOENT:
            raise FileNotFoundError(name) from exc
        raise BindingStoreError("binding store file is unavailable") from exc
    try:
        if create:
            os.fchmod(descriptor, _MODE)
        _verify_private_file(directory_fd, name, os.fstat(descriptor))
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


def _write_all(descriptor: int, payload: bytes) -> None:
    view = memoryview(payload)
    try:
        while view:
            written = os.write(descriptor, view)
            if written <= 0:
                raise OSError("binding store write made no progress")
            view = view[written:]
    except OSError as exc:
        raise BindingStoreError("binding store write failed") from exc


def _parse_state(raw: object) -> _BindingState:
    record = _object_dict(raw)
    if record is None:
        raise BindingStoreCorruption("binding store schema is invalid")
    version = record.get("version")
    if type(version) is not int:
        raise BindingStoreCorruption("binding store schema is invalid")
    expected = {"version", "active_session_id", "bindings"}
    if version == 2:
        expected.add("projections")
    if version not in {1, 2} or set(record) != expected:
        raise BindingStoreCorruption("binding store schema is invalid")
    active = record["active_session_id"]
    values = _object_list(record["bindings"])
    if (active is not None and _legacy_nonempty_string(active) is None) or values is None:
        raise BindingStoreCorruption("binding store schema is invalid")
    bindings: list[WizardBinding] = []
    session_ids: set[str] = set()
    for item in values:
        binding = WizardBinding.from_dict(item)
        if binding is None or binding.session_id in session_ids:
            raise BindingStoreCorruption("binding store bindings are invalid")
        session_ids.add(binding.session_id)
        bindings.append(binding)
    if active is not None and active not in session_ids:
        raise BindingStoreCorruption("binding store active binding is invalid")
    projections: list[TelegramProjection] = []
    if version == 2:
        projection_values = _object_list(record["projections"])
        if projection_values is None:
            raise BindingStoreCorruption("binding store projections are invalid")
        for item in projection_values:
            projection = TelegramProjection.from_dict(item)
            if projection is None:
                raise BindingStoreCorruption("binding store projections are invalid")
            projections.append(projection)
    _validate_projection_set(tuple(projections))
    return _BindingState(tuple(bindings), cast(str | None, active), tuple(projections))


def _validate_projection_set(projections: tuple[TelegramProjection, ...]) -> None:
    update_ids: set[int] = set()
    by_session: dict[str, list[TelegramProjection]] = {}
    for projection in projections:
        if projection.ingress.update_id in update_ids:
            raise BindingStoreCorruption("binding store has duplicate ingress")
        update_ids.add(projection.ingress.update_id)
        by_session.setdefault(projection.session_id, []).append(projection)
    for session_items in by_session.values():
        if sum(item.phase in _NONTERMINAL_PHASES for item in session_items) > 1:
            raise BindingStoreCorruption("binding store has concurrent projections")
        for earlier, later in zip(session_items, session_items[1:]):
            if earlier.phase in _NONTERMINAL_PHASES or later.source != earlier.target:
                raise BindingStoreCorruption("binding store projection cursor conflicts")


def _unexpired(
    projections: tuple[TelegramProjection, ...], now_epoch: int,
) -> tuple[TelegramProjection, ...]:
    return tuple(item for item in projections if item.expires_at > now_epoch)


def _state_to_v2(state: _BindingState) -> dict[str, object]:
    _validate_projection_set(state.projections)
    return {
        "version": 2,
        "active_session_id": state.active_session_id,
        "bindings": [binding.to_dict() for binding in state.bindings],
        "projections": [projection.to_dict() for projection in state.projections],
    }