"""Subtitle tracks for the Videos app: extracted once, whole-file, to WebVTT."""

import asyncio
import logging
import time

import blake3
from meshbay_common import MNP_VERSION
from meshbay_common.protocol import MNP

from meshbay_node.media_probe import probe_video as _probe_video
from meshbay_node.roots import off_disk
from meshbay_node.transport.webrtc.disk import _locate
from meshbay_node.transport.webrtc.media_tools import (
    _extract_subtitle_to_webvtt,
    _subtitle_timeout_for,
)

log = logging.getLogger("meshbay_node.transport.webrtc_server")


class SubtitlesMixin:
    async def _do_subtitle_request(self, msg: dict) -> None:
        """
        One embedded subtitle track, extracted whole-file to WebVTT and served
        back through the ordinary file_req/chunk path — the same indirection
        as `_do_audio_transcode_request`, and cached the same way, so a
        film's subtitles are extracted once in the life of the file rather
        than once per viewing.

        The ordinal is validated against `probe_video`'s *filtered* list and
        then used as the ffmpeg `-map 0:s:<n>` argument, which is only correct
        because `SubtitleTrack.ordinal` counts every subtitle stream including
        the bitmap ones the list omits (see media_probe.py). Checking
        membership rather than range is what makes that hold: a bitmap
        ordinal is in range and is not in the list, and extracting it would
        produce an empty WebVTT — a subtitle track with no subtitles in it,
        which reports no error anywhere.
        """
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return
        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
        if refusal is not None:
            self._send({"type": "error", "detail": refusal})
            return

        media_cache = self._ctx.get("media_cache")
        if media_cache is None:
            self._send({"type": "error", "detail": "Subtitles unavailable"})
            return

        try:
            ordinal = int(msg.get("track", 0) or 0)
        except (TypeError, ValueError):
            ordinal = -1

        # Logged before the cache is consulted, and that ordering is the point:
        # a cached track used to answer without leaving a line, so the journal
        # could not say whether a viewer had asked for subtitles at all. That
        # turned "no request in the log" into evidence it was never entitled
        # to be — the second time in this feature that a silent success was
        # read as an absence.
        log.info("subtitle: req file=%s track=%d size=%.1fMB slots_free=%s",
                 file_id[:12], ordinal, entry.size / 1e6,
                 getattr(self._ctx.get("_transcode_sem"), "_value", "?"))
        t0 = time.monotonic()

        synthetic_id = f"subtitle:{entry.id}:{ordinal}"
        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
        if cached_hash is not None:
            blob = await media_cache.get_thumb(cached_hash)
            if blob is not None:
                log.info("subtitle: served file=%s track=%d from cache, %d bytes",
                         file_id[:12], ordinal, len(blob))
                self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
                            "file_id": file_id, "track": ordinal,
                            "hash": cached_hash, "size": len(blob),
                            "mime": "text/vtt"})
                return
            # Cached hash but the blob was pruned: fall through and extract
            # again, same as a cold cache.

        inflight = self._ctx.setdefault("_subtitle_inflight", {})
        running = inflight.get(synthetic_id)
        if running is not None:
            log.info("subtitle: file=%s track=%d already extracting, waiting",
                     file_id[:12], ordinal)
            # Shielded: this waiter being cancelled must not cancel the
            # extraction every other waiter is also relying on.
            answer = await asyncio.shield(running)
            if answer is None:
                self._send({"type": "error", "detail": "Subtitle extraction failed"})
                return
            subtitle_hash, size = answer
            log.info("subtitle: served file=%s track=%d from a shared extraction, %d bytes",
                     file_id[:12], ordinal, size)
            self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
                        "file_id": file_id, "track": ordinal,
                        "hash": subtitle_hash, "size": size, "mime": "text/vtt"})
            return

        # One extraction per file and track at a time, across every viewer.
        # The cache is only consulted on the way in, so two clicks seconds
        # apart both missed it and both ran — seen in the log as two identical
        # extractions of one 4.3 GB file overlapping, each holding a transcode
        # slot and reading the whole file. Latecomers wait for the answer the
        # first one is already producing.
        #
        # **Registered before the first `await` below**, never after. The
        # first version registered it after the probe, and two requests
        # arriving together both got past the check above while neither had
        # registered yet — so both extracted, which is the race this exists to
        # close. The probe and the slot check sit behind it for that reason.
        fut = asyncio.get_running_loop().create_future()
        inflight[synthetic_id] = fut
        try:
            probe = await _probe_video(str(file_path))
            if not any(tr.ordinal == ordinal for tr in probe.subtitle_tracks):
                # Not a range check — see this method's docstring.
                log.warning("subtitle: file=%s has no text track %d",
                            file_id[:12], ordinal)
                self._send({"type": "error", "detail": "No such subtitle track"})
                return

            sem = self._transcode_semaphore()
            if sem.locked() and sem._value <= 0:
                log.info("subtitle: refused, no transcode slot free")
                self._send({"type": "error", "detail": "Server busy, retry shortly"})
                return

            budget = _subtitle_timeout_for(entry.size)
            async with sem:
                log.info("subtitle: extracting file=%s track=%d (slot taken, up to %.0fs)",
                         file_id[:12], ordinal, budget)
                blob = await _extract_subtitle_to_webvtt(file_path, ordinal, budget)

            subtitle_hash = blake3.blake3(blob).hexdigest()
            await media_cache.put_thumb(subtitle_hash, synthetic_id, blob)
            self._audit("subtitle_extract", f"{entry.name} [{ordinal}]")
            log.info("subtitle: extracted file=%s track=%d in %.1fs, %d bytes",
                     file_id[:12], ordinal, time.monotonic() - t0, len(blob))
            fut.set_result((subtitle_hash, len(blob)))
            self._send({"type": MNP.SUBTITLE_RESP, "v": MNP_VERSION,
                        "file_id": file_id, "track": ordinal,
                        "hash": subtitle_hash, "size": len(blob),
                        "mime": "text/vtt"})
        # BaseException, not Exception: a cancelled task — the peer went away,
        # the session is being torn down — raises CancelledError, which is not
        # an Exception and would otherwise leave this handler with no reply
        # sent and no line in the log. The client is then waiting on something
        # nothing will ever answer, which is the shape nobody can report.
        except BaseException as e:
            log.warning("subtitle: extract failed file=%s track=%d after %.1fs: %r",
                        file_id[:12], ordinal, time.monotonic() - t0, e)
            self._send({"type": "error",
                        "detail": f"Subtitle extraction failed: {e}"})
            if isinstance(e, asyncio.CancelledError):
                raise
        finally:
            # A result either way, never an exception: a future nobody is
            # waiting on yet would be an unretrieved-exception warning, and a
            # cancelled task would otherwise leave every latecomer waiting on
            # a future that is never resolved at all.
            if not fut.done():
                fut.set_result(None)
            inflight.pop(synthetic_id, None)