"""Thread read policy for Matrix conversation cache.

Read-side invariants:

1. Every thread read through this policy first waits for the room and same-thread write queue to drain
   (``wait_for_thread_idle``), so a read never observes cache state older than mutations already queued
   when the read began.
   Startup prewarm bypasses this policy and relies on the backend replacement guard without occupying
   the live write coordinator during its bulk room scan.

2. Dispatch-safe modes (``DISPATCH_SNAPSHOT``, ``DISPATCH_FULL``) bound the whole wait-plus-fetch by one
   shared timeout and return an explicitly degraded empty result
   (``THREAD_HISTORY_SOURCE_DEGRADED``) instead of blocking dispatch; consumers must treat that result
   as unusable for caching and root proofs.

3. Each cache-miss refill is single-flight per full read contract, so one leader runs the homeserver
   scan and concurrent matching readers share it. There is no admission gate, failure backoff, or cooldown.
   ``ADVISORY_FULL``, ``STRICT_FULL``, and ``STRICT_SOURCE_REFRESH`` have no dispatch timeout; strict
   modes are intentionally not dispatch-safe because they may block for authoritative post-lock model
   context or a direct source refresh.

4. A stale-cache thread tail is never used for MSC3440 latest-event fallback:
   ``get_latest_thread_event_id_if_needed`` falls back to the thread root instead.
"""

from __future__ import annotations

import asyncio
import time
import typing
from enum import Enum, auto
from typing import TYPE_CHECKING

from mindroom.constants import runtime_dispatch_thread_read_timeout_seconds
from mindroom.matrix.cache.thread_cache_helpers import latest_visible_thread_event_id
from mindroom.matrix.cache.thread_history_result import ThreadHistoryResult, thread_history_result
from mindroom.matrix.thread_diagnostics import (
    THREAD_HISTORY_DEGRADED_DIAGNOSTIC,
    THREAD_HISTORY_ERROR_DIAGNOSTIC,
    THREAD_HISTORY_SOURCE_DEGRADED,
    THREAD_HISTORY_SOURCE_DIAGNOSTIC,
    THREAD_HISTORY_SOURCE_STALE_CACHE,
)
from mindroom.timing import elapsed_ms_since

if TYPE_CHECKING:
    import structlog

    from mindroom.bot_runtime_view import BotRuntimeView
    from mindroom.matrix.cache.write_coordinator import EventCacheWriteCoordinator


_CACHE_COORDINATOR_TIMEOUT = "cache_coordinator_timeout"
_DISPATCH_READ_TIMEOUT = "dispatch_read_timeout"


def _remaining_dispatch_thread_read_seconds(queue_wait_started: float, timeout_seconds: float) -> float:
    """Return the remaining dispatch-safe read budget."""
    deadline = queue_wait_started + timeout_seconds
    return max(0.0, deadline - time.perf_counter())


class ThreadReadMode(Enum):
    """Named thread-read policies for cache coordination and source freshness."""

    ADVISORY_FULL = auto()
    DISPATCH_SNAPSHOT = auto()
    DISPATCH_FULL = auto()
    STRICT_FULL = auto()
    STRICT_SOURCE_REFRESH = auto()

    @property
    def dispatch_safe(self) -> bool:
        """Return whether this mode is on the live dispatch fail-open path."""
        # STRICT_FULL intentionally stays false: it may block for authoritative post-lock model context.
        return self in {
            ThreadReadMode.DISPATCH_SNAPSHOT,
            ThreadReadMode.DISPATCH_FULL,
        }


class _ThreadHistoryFetcher(typing.Protocol):
    """Client-backed thread-history fetcher with refresh diagnostics metadata."""

    def __call__(
        self,
        room_id: str,
        thread_id: str,
        *,
        caller_label: str,
        coordinator_queue_wait_ms: float,
    ) -> typing.Awaitable[ThreadHistoryResult]:
        """Fetch one thread from cache/source and attach refresh diagnostics."""


class ThreadReadPolicy:
    """Own thread-history reads for one cache facade."""

    def __init__(
        self,
        *,
        logger_getter: typing.Callable[[], structlog.stdlib.BoundLogger],
        runtime: BotRuntimeView,
        fetch_thread_history_from_client: _ThreadHistoryFetcher,
        fetch_dispatch_thread_history_from_client: _ThreadHistoryFetcher,
        fetch_dispatch_thread_snapshot_from_client: _ThreadHistoryFetcher,
        refresh_thread_history_from_source: _ThreadHistoryFetcher,
    ) -> None:
        self._logger_getter = logger_getter
        self.runtime = runtime
        self.fetch_thread_history_from_client = fetch_thread_history_from_client
        self.fetch_dispatch_thread_history_from_client = fetch_dispatch_thread_history_from_client
        self.fetch_dispatch_thread_snapshot_from_client = fetch_dispatch_thread_snapshot_from_client
        self.refresh_thread_history_from_source = refresh_thread_history_from_source

    @property
    def logger(self) -> structlog.stdlib.BoundLogger:
        """Return the facade-bound logger so collaborator rebinding stays visible."""
        return self._logger_getter()

    def _coordinator(self) -> EventCacheWriteCoordinator | None:
        return self.runtime.event_cache_write_coordinator

    def _dispatch_thread_read_timeout_seconds(self) -> float:
        return runtime_dispatch_thread_read_timeout_seconds(self.runtime.runtime_paths)

    def _fetcher_for_mode(self, mode: ThreadReadMode) -> _ThreadHistoryFetcher:
        """Return the client fetcher matching one named read policy."""
        return {
            ThreadReadMode.ADVISORY_FULL: self.fetch_thread_history_from_client,
            ThreadReadMode.DISPATCH_SNAPSHOT: self.fetch_dispatch_thread_snapshot_from_client,
            ThreadReadMode.DISPATCH_FULL: self.fetch_dispatch_thread_history_from_client,
            ThreadReadMode.STRICT_FULL: self.fetch_dispatch_thread_history_from_client,
            ThreadReadMode.STRICT_SOURCE_REFRESH: self.refresh_thread_history_from_source,
        }[mode]

    async def _wait_for_pending_thread_cache_updates(self, room_id: str, thread_id: str) -> None:
        coordinator = self._coordinator()
        if coordinator is None:
            return
        await coordinator.wait_for_thread_idle(
            room_id,
            thread_id,
            ignore_cancelled_room_fences=True,
            coordination_scope=self.runtime.event_cache.principal_id,
        )

    def _degraded_dispatch_timeout_result(
        self,
        *,
        room_id: str,
        thread_id: str,
        caller_label: str,
        queue_wait_started: float,
        error_code: str,
        dispatch_timeout_seconds: float,
        fetch_started: float | None = None,
    ) -> ThreadHistoryResult:
        coordinator_queue_wait_ms = elapsed_ms_since(queue_wait_started, clock=time.perf_counter)
        dispatch_fetch_wait_ms = (
            elapsed_ms_since(fetch_started, clock=time.perf_counter) if fetch_started is not None else None
        )
        diagnostics = {
            THREAD_HISTORY_SOURCE_DIAGNOSTIC: THREAD_HISTORY_SOURCE_DEGRADED,
            THREAD_HISTORY_DEGRADED_DIAGNOSTIC: True,
            THREAD_HISTORY_ERROR_DIAGNOSTIC: error_code,
            "coordinator_queue_wait_ms": coordinator_queue_wait_ms,
            "caller_label": caller_label,
            "dispatch_thread_read_timeout_seconds": dispatch_timeout_seconds,
        }
        if dispatch_fetch_wait_ms is not None:
            diagnostics["dispatch_fetch_wait_ms"] = dispatch_fetch_wait_ms
        log_fields = {
            "room_id": room_id,
            "thread_id": thread_id,
            "caller_label": caller_label,
            "thread_read_degraded": True,
            "thread_read_error": error_code,
            "coordinator_queue_wait_ms": coordinator_queue_wait_ms,
            "dispatch_thread_read_timeout_seconds": dispatch_timeout_seconds,
        }
        if dispatch_fetch_wait_ms is not None:
            log_fields["dispatch_fetch_wait_ms"] = dispatch_fetch_wait_ms
        self.logger.warning(
            "matrix_cache_thread_read_degraded",
            **log_fields,
        )
        return thread_history_result(
            [],
            is_full_history=False,
            diagnostics=diagnostics,
        )

    async def _load_thread_read(
        self,
        room_id: str,
        thread_id: str,
        *,
        fetcher: _ThreadHistoryFetcher,
        caller_label: str,
        queue_wait_started: float,
    ) -> ThreadHistoryResult:
        """Load one read and attach its coordinator queue wait."""
        coordinator_queue_wait_ms = elapsed_ms_since(queue_wait_started, clock=time.perf_counter)
        return await fetcher(
            room_id,
            thread_id,
            caller_label=caller_label,
            coordinator_queue_wait_ms=coordinator_queue_wait_ms,
        )

    async def _load_dispatch_thread_read(
        self,
        room_id: str,
        thread_id: str,
        *,
        fetcher: _ThreadHistoryFetcher,
        caller_label: str,
        queue_wait_started: float,
        dispatch_timeout_seconds: float,
    ) -> ThreadHistoryResult:
        fetch_started = time.perf_counter()
        remaining_timeout = _remaining_dispatch_thread_read_seconds(queue_wait_started, dispatch_timeout_seconds)
        if remaining_timeout <= 0:
            return self._degraded_dispatch_timeout_result(
                room_id=room_id,
                thread_id=thread_id,
                caller_label=caller_label,
                queue_wait_started=queue_wait_started,
                error_code=_DISPATCH_READ_TIMEOUT,
                dispatch_timeout_seconds=dispatch_timeout_seconds,
                fetch_started=fetch_started,
            )
        try:
            # Dispatch read-through fetches are bounded live reads. Cancelling them on timeout
            # is intentional; cache mutation tasks are protected by the write coordinator.
            return await asyncio.wait_for(
                self._load_thread_read(
                    room_id,
                    thread_id,
                    fetcher=fetcher,
                    caller_label=caller_label,
                    queue_wait_started=queue_wait_started,
                ),
                timeout=remaining_timeout,
            )
        except TimeoutError:
            return self._degraded_dispatch_timeout_result(
                room_id=room_id,
                thread_id=thread_id,
                caller_label=caller_label,
                queue_wait_started=queue_wait_started,
                error_code=_DISPATCH_READ_TIMEOUT,
                dispatch_timeout_seconds=dispatch_timeout_seconds,
                fetch_started=fetch_started,
            )

    async def read_thread(
        self,
        room_id: str,
        thread_id: str,
        *,
        mode: ThreadReadMode,
        caller_label: str,
    ) -> ThreadHistoryResult:
        """Resolve one thread read through the same-thread barrier and fetch selection path."""
        queue_wait_started = time.perf_counter()
        fetcher = self._fetcher_for_mode(mode)
        if mode.dispatch_safe:
            dispatch_timeout_seconds = self._dispatch_thread_read_timeout_seconds()
            try:
                remaining_timeout = _remaining_dispatch_thread_read_seconds(
                    queue_wait_started,
                    dispatch_timeout_seconds,
                )
                await asyncio.wait_for(
                    self._wait_for_pending_thread_cache_updates(room_id, thread_id),
                    timeout=remaining_timeout,
                )
            except TimeoutError:
                return self._degraded_dispatch_timeout_result(
                    room_id=room_id,
                    thread_id=thread_id,
                    caller_label=caller_label,
                    queue_wait_started=queue_wait_started,
                    error_code=_CACHE_COORDINATOR_TIMEOUT,
                    dispatch_timeout_seconds=dispatch_timeout_seconds,
                )
            return await self._load_dispatch_thread_read(
                room_id,
                thread_id,
                fetcher=fetcher,
                caller_label=caller_label,
                queue_wait_started=queue_wait_started,
                dispatch_timeout_seconds=dispatch_timeout_seconds,
            )
        await self._wait_for_pending_thread_cache_updates(room_id, thread_id)
        return await self._load_thread_read(
            room_id,
            thread_id,
            fetcher=fetcher,
            caller_label=caller_label,
            queue_wait_started=queue_wait_started,
        )

    async def get_latest_thread_event_id_if_needed(
        self,
        room_id: str,
        thread_id: str | None,
        reply_to_event_id: str | None = None,
        existing_event_id: str | None = None,
        *,
        caller_label: str = "latest_thread_event_lookup",
    ) -> str | None:
        """Resolve the latest visible thread event when MSC3440 fallback needs it."""
        if thread_id is None or existing_event_id is not None or reply_to_event_id is not None:
            return None
        try:
            thread_history = await self.read_thread(
                room_id,
                thread_id,
                mode=ThreadReadMode.ADVISORY_FULL,
                caller_label=caller_label,
            )
        except Exception as exc:
            self.logger.warning(
                "Failed to refresh latest thread event ID; falling back to thread root",
                room_id=room_id,
                thread_id=thread_id,
                error=str(exc),
            )
            return thread_id
        if thread_history.diagnostics.get(THREAD_HISTORY_SOURCE_DIAGNOSTIC) == THREAD_HISTORY_SOURCE_STALE_CACHE:
            self.logger.warning(
                "Ignoring stale cached thread tail for latest-event lookup; falling back to thread root",
                room_id=room_id,
                thread_id=thread_id,
            )
            return thread_id
        return latest_visible_thread_event_id(thread_history) or thread_id
