Merge pull request 'Click-to-listen on the live wall + HTTP Range on /media (#233)' (#253) from feat/233-wall-audio into main

drb-c2-core/app/internal/storage.py

@@ -121,6 +121,67 @@ async def download_audio(gcs_uri: str) -> Optional[bytes]:
         return None
 
 
+# ── Ranged reads for /media playback (server-26#233) ────────────────────────
+# A 600 s FLAC call is ~13 MB. download_audio() above pulls the whole object
+# into memory, so /media used to make the browser wait for all of it before
+# the first sample played. These two read only what a request asks for: the
+# object size (one metadata GET) and then an inclusive byte span, which the
+# router streams out in chunks. One cached client, because the router issues
+# several span reads per playback and a fresh Client() per read re-resolves
+# credentials every time.
+_gcs_client_cache = None
+
+
+def _gcs_client():
+    global _gcs_client_cache
+    if _gcs_client_cache is None:
+        from google.cloud import storage
+        if settings.gcp_credentials_path:
+            _gcs_client_cache = storage.Client.from_service_account_json(settings.gcp_credentials_path)
+        else:
+            _gcs_client_cache = storage.Client()
+    return _gcs_client_cache
+
+
+async def audio_object_size(gcs_uri: str) -> Optional[int]:
+    """Size in bytes of a stored audio object, or None if it is missing/unreadable."""
+    bucket_name, blob_path = split_gcs_uri(gcs_uri)
+    if not bucket_name:
+        return None
+
+    def _size() -> Optional[int]:
+        blob = _gcs_client().bucket(bucket_name).blob(blob_path)
+        blob.reload()
+        return blob.size
+
+    try:
+        size = await asyncio.to_thread(_size)
+    except Exception as e:
+        logger.warning(f"GCS metadata read failed for {gcs_uri}: {e}")
+        return None
+    return int(size) if size is not None else None
+
+
+async def download_audio_range(gcs_uri: str, start: int, end: int) -> Optional[bytes]:
+    """Bytes ``start``..``end`` INCLUSIVE of a stored object (GCS's own convention).
+
+    Reads only that span from GCS — never the whole object.
+    """
+    bucket_name, blob_path = split_gcs_uri(gcs_uri)
+    if not bucket_name:
+        return None
+
+    def _read() -> bytes:
+        blob = _gcs_client().bucket(bucket_name).blob(blob_path)
+        return blob.download_as_bytes(start=start, end=end)
+
+    try:
+        return await asyncio.to_thread(_read)
+    except Exception as e:
+        logger.warning(f"GCS ranged read failed for {gcs_uri} [{start}-{end}]: {e}")
+        return None
+
+
 def split_gcs_uri(gcs_uri: str) -> Tuple[Optional[str], Optional[str]]:
     """``gs://bucket/path/to.mp3`` → ``("bucket", "path/to.mp3")``."""
     if not gcs_uri or not gcs_uri.startswith("gs://"):

drb-c2-core/app/routers/media.py

@@ -10,19 +10,100 @@ happens inline below, in the same spirit as routers/enrollment.py.
 The bucket stays fully private and c2-core reads the object server-side with
 Application Default Credentials, so no GCS signed URL — and therefore no
 service-account private key on the VM — is involved anywhere in this path.
+
+HTTP Range (server-26#233): the body is streamed from GCS in fixed-size span
+reads instead of downloaded whole first. A 600 s FLAC call is ~13 MB, and the
+old full-body response (``Accept-Ranges: none``) made the browser wait for all
+of it before playback started and made seeking re-download it. Now playback
+starts after the first span and a seek is a ``206`` for just the bytes needed.
+Only a single range is honoured; a multi-range request gets the full body,
+which RFC 9110 §14.2 permits.
 """
-from fastapi import APIRouter, HTTPException, Query, Response
+import re
+from typing import AsyncIterator, Optional, Tuple
+
+from fastapi import APIRouter, Header, HTTPException, Query, Response
+from fastapi.responses import StreamingResponse
+
 from app.internal import firestore as fstore
-from app.internal.storage import verify_audio_link, gcs_uri_for_call, download_audio, content_type_for
+from app.internal.logger import logger
+from app.internal.storage import (
+    audio_object_size,
+    content_type_for,
+    download_audio_range,
+    gcs_uri_for_call,
+    verify_audio_link,
+)
 
 router = APIRouter(prefix="/media", tags=["media"])
 
+# Bytes per GCS span read. ~25 s of FLAC at ~1.3 MB/min, so the first span —
+# the only one the browser waits on — is one small GCS round-trip.
+STREAM_CHUNK_BYTES = 512 * 1024
+
+# Immutable content, but the URL expires — cache privately only.
+_CACHE_CONTROL = "private, max-age=3600"
+
+_RANGE_RE = re.compile(r"^\s*bytes\s*=\s*(\d*)\s*-\s*(\d*)\s*$", re.IGNORECASE)
+
+
+class _Unsatisfiable(Exception):
+    pass
+
+
+def parse_range(header: Optional[str], size: int) -> Optional[Tuple[int, int]]:
+    """Resolve a ``Range`` header against an object of ``size`` bytes.
+
+    Returns an inclusive ``(start, end)``, or None when the header is absent or
+    should be ignored (not ``bytes``, multi-range, malformed — RFC 9110 lets a
+    server answer those with the full 200). Raises _Unsatisfiable for a
+    well-formed range that selects nothing (-> 416).
+    """
+    if not header:
+        return None
+    m = _RANGE_RE.match(header)
+    if not m:
+        return None  # other unit, multi-range ("a-b,c-d") or garbage
+    first, last = m.group(1), m.group(2)
+    if first == "" and last == "":
+        return None
+    if first == "":
+        # Suffix range: the final N bytes.
+        n = int(last)
+        if n == 0 or size == 0:
+            raise _Unsatisfiable()
+        return max(0, size - n), size - 1
+    start = int(first)
+    if last != "" and int(last) < start:
+        return None  # invalid spec -> ignore the header
+    if start >= size:
+        raise _Unsatisfiable()
+    end = size - 1 if last == "" else min(int(last), size - 1)
+    return start, end
+
+
+async def _stream(gcs_uri: str, first_chunk: bytes, start: int, end: int) -> AsyncIterator[bytes]:
+    yield first_chunk
+    pos = start + len(first_chunk)
+    while pos <= end:
+        chunk_end = min(pos + STREAM_CHUNK_BYTES - 1, end)
+        data = await download_audio_range(gcs_uri, pos, chunk_end)
+        if not data:
+            # Headers are already sent, so the status can't change; ending the
+            # body short makes the client see a truncated response rather than
+            # silently wrong audio.
+            logger.warning(f"/media stream cut short at byte {pos} of {gcs_uri}")
+            return
+        yield data
+        pos += len(data)
+
 
 @router.get("/calls/{call_id}/audio")
 async def get_call_audio(
     call_id: str,
     exp: int = Query(..., description="Link expiry, unix seconds."),
     sig: str = Query(..., description="HMAC over call_id and expiry."),
+    range_header: Optional[str] = Header(None, alias="Range"),
 ):
     # Verify before touching Firestore so an invalid link costs nothing.
     if not verify_audio_link(call_id, exp, sig):
@@ -36,26 +117,48 @@ async def get_call_audio(
     if not gcs_uri:
         raise HTTPException(404, "No audio for this call.")
 
-    data = await download_audio(gcs_uri)
-    if not data:
+    size = await audio_object_size(gcs_uri)
+    if size is None:
         raise HTTPException(404, "Audio object missing from storage.")
 
-    return Response(
-        content=data,
-        # Was hardcoded audio/mpeg. The node now uploads FLAC, and a browser
-        # will not play a FLAC body labelled audio/mpeg. Derived from the stored
-        # object's extension so old .mp3 recordings keep working unchanged.
-        media_type=content_type_for(gcs_uri),
-        headers={
-            "Content-Length": str(len(data)),
-            # Whole body at once, no Range support. This was comfortable at
-            # 16 kbps mono (~60 KB for a 30 s call); FLAC is ~1.3 MB/min, so a
-            # long call is now tens of MB and the browser must download all of
-            # it before playback starts. Acceptable for typical few-second
-            # transmissions, but this is the change that makes Range support
-            # actually matter — see DEFERRED.md.
-            "Accept-Ranges": "none",
-            # Immutable content, but the URL expires — cache privately only.
-            "Cache-Control": "private, max-age=3600",
-        },
+    # Was hardcoded audio/mpeg. The node now uploads FLAC, and a browser will
+    # not play a FLAC body labelled audio/mpeg. Derived from the stored
+    # object's extension so old .mp3 recordings keep working unchanged.
+    media_type = content_type_for(gcs_uri)
+
+    try:
+        span = parse_range(range_header, size)
+    except _Unsatisfiable:
+        return Response(
+            status_code=416,
+            headers={"Content-Range": f"bytes */{size}", "Accept-Ranges": "bytes"},
+        )
+
+    if size == 0:
+        return Response(
+            content=b"",
+            media_type=media_type,
+            headers={"Accept-Ranges": "bytes", "Cache-Control": _CACHE_CONTROL},
+        )
+
+    start, end = span if span else (0, size - 1)
+    # Fetched before the response starts so a read failure is still a clean
+    # 404 rather than a stream that dies after a 200/206 has gone out.
+    first = await download_audio_range(gcs_uri, start, min(start + STREAM_CHUNK_BYTES - 1, end))
+    if not first:
+        raise HTTPException(404, "Audio object missing from storage.")
+
+    headers = {
+        "Content-Length": str(end - start + 1),
+        "Accept-Ranges": "bytes",
+        "Cache-Control": _CACHE_CONTROL,
+    }
+    if span:
+        headers["Content-Range"] = f"bytes {start}-{end}/{size}"
+
+    return StreamingResponse(
+        _stream(gcs_uri, first, start, end),
+        status_code=206 if span else 200,
+        media_type=media_type,
+        headers=headers,
     )

drb-c2-core/tests/test_media_range.py

@@ -0,0 +1,220 @@
+"""
+HTTP Range on /media/calls/{id}/audio (server-26#233).
+
+The GCS reads are replaced by an in-memory object: ``audio_object_size`` and
+``download_audio_range`` are patched where routers/media.py looks them up, and
+every span read is recorded so the tests can assert the route never asks GCS
+for more than the requested range (plus at most one STREAM_CHUNK_BYTES span).
+"""
+import time
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.internal import storage
+from app.routers import media
+
+app = FastAPI()
+app.include_router(media.router)
+client = TestClient(app)
+
+CALL_ID = "call-abc"
+GCS_URI = "gs://bucket/calls/call-abc.flac"
+
+
+class FakeObject:
+    """A stored object plus a log of every (start, end) span read from it."""
+
+    def __init__(self, data: bytes):
+        self.data = data
+        self.reads: list[tuple[int, int]] = []
+        self.size_calls = 0
+
+    async def size(self, gcs_uri):
+        self.size_calls += 1
+        return len(self.data)
+
+    async def read(self, gcs_uri, start, end):
+        self.reads.append((start, end))
+        return self.data[start:end + 1]
+
+
+@pytest.fixture
+def signed(monkeypatch):
+    monkeypatch.setattr(storage.settings, "service_key", "test-key", raising=False)
+    exp = int(time.time()) + 600
+    return {"exp": exp, "sig": storage.sign_audio_link(CALL_ID, exp)}
+
+
+def _run(obj: FakeObject, params, headers=None, doc=None):
+    doc_get = AsyncMock(return_value=doc if doc is not None else {"call_id": CALL_ID, "audio_gcs_uri": GCS_URI})
+    with patch.object(media.fstore, "doc_get", doc_get), \
+            patch.object(media, "audio_object_size", obj.size), \
+            patch.object(media, "download_audio_range", obj.read):
+        resp = client.get(f"/media/calls/{CALL_ID}/audio", params=params, headers=headers or {})
+    return resp, doc_get
+
+
+def test_bounded_range_returns_206_with_exact_bytes(signed):
+    obj = FakeObject(bytes(range(256)) * 40)  # 10240 bytes
+    resp, _ = _run(obj, signed, {"Range": "bytes=100-199"})
+    assert resp.status_code == 206
+    assert resp.content == obj.data[100:200]
+    assert resp.headers["content-range"] == f"bytes 100-199/{len(obj.data)}"
+    assert resp.headers["content-length"] == "100"
+    assert resp.headers["accept-ranges"] == "bytes"
+    assert resp.headers["content-type"] == "audio/flac"
+    assert resp.headers["cache-control"] == "private, max-age=3600"
+    # Only the requested span was read from GCS — not the whole object.
+    assert obj.reads == [(100, 199)]
+
+
+def test_open_ended_range(signed):
+    obj = FakeObject(b"x" * 5000 + b"tail")
+    resp, _ = _run(obj, signed, {"Range": "bytes=5000-"})
+    assert resp.status_code == 206
+    assert resp.content == b"tail"
+    assert resp.headers["content-range"] == "bytes 5000-5003/5004"
+
+
+def test_suffix_range(signed):
+    obj = FakeObject(b"0123456789")
+    resp, _ = _run(obj, signed, {"Range": "bytes=-3"})
+    assert resp.status_code == 206
+    assert resp.content == b"789"
+    assert resp.headers["content-range"] == "bytes 7-9/10"
+
+
+def test_suffix_longer_than_object_returns_whole_object(signed):
+    obj = FakeObject(b"0123456789")
+    resp, _ = _run(obj, signed, {"Range": "bytes=-500"})
+    assert resp.status_code == 206
+    assert resp.content == obj.data
+    assert resp.headers["content-range"] == "bytes 0-9/10"
+
+
+def test_end_past_object_is_clamped(signed):
+    obj = FakeObject(b"0123456789")
+    resp, _ = _run(obj, signed, {"Range": "bytes=8-999"})
+    assert resp.status_code == 206
+    assert resp.content == b"89"
+    assert resp.headers["content-range"] == "bytes 8-9/10"
+
+
+@pytest.mark.parametrize("rng", ["bytes=10-", "bytes=50-60", "bytes=-0"])
+def test_unsatisfiable_range_is_416(signed, rng):
+    obj = FakeObject(b"0123456789")
+    resp, _ = _run(obj, signed, {"Range": rng})
+    assert resp.status_code == 416
+    assert resp.headers["content-range"] == "bytes */10"
+    assert obj.reads == []
+
+
+def test_no_range_is_200_full_body_and_advertises_ranges(signed):
+    obj = FakeObject(b"abcdef" * 100)
+    resp, _ = _run(obj, signed)
+    assert resp.status_code == 200
+    assert resp.content == obj.data
+    assert resp.headers["accept-ranges"] == "bytes"
+    assert resp.headers["content-length"] == str(len(obj.data))
+    assert "content-range" not in resp.headers
+
+
+@pytest.mark.parametrize("rng", ["bytes=0-1,4-5", "items=0-5", "bytes=5-2", "garbage"])
+def test_ignored_range_headers_fall_back_to_200(signed, rng):
+    obj = FakeObject(b"0123456789")
+    resp, _ = _run(obj, signed, {"Range": rng})
+    assert resp.status_code == 200
+    assert resp.content == obj.data
+
+
+def test_long_object_is_streamed_in_bounded_spans(signed):
+    # A ~13 MB "600 s FLAC": a seek near the end reads only from there on, in
+    # spans no larger than STREAM_CHUNK_BYTES, and the first span is all the
+    # route waits on before the response starts.
+    size = 13 * 1024 * 1024
+    obj = FakeObject(bytes(size))
+    start = size - 1_500_000
+    resp, _ = _run(obj, signed, {"Range": f"bytes={start}-"})
+    assert resp.status_code == 206
+    assert len(resp.content) == 1_500_000
+    assert obj.reads[0][0] == start
+    assert all(e - s + 1 <= media.STREAM_CHUNK_BYTES for s, e in obj.reads)
+    assert sum(e - s + 1 for s, e in obj.reads) == 1_500_000
+
+
+def test_bad_signature_is_403_before_any_lookup(signed):
+    obj = FakeObject(b"0123456789")
+    resp, doc_get = _run(obj, {"exp": signed["exp"], "sig": "0" * 64}, {"Range": "bytes=0-3"})
+    assert resp.status_code == 403
+    doc_get.assert_not_called()
+    assert obj.size_calls == 0
+    assert obj.reads == []
+
+
+def test_expired_link_is_403(signed, monkeypatch):
+    exp = int(time.time()) - 5
+    obj = FakeObject(b"0123456789")
+    resp, doc_get = _run(obj, {"exp": exp, "sig": storage.sign_audio_link(CALL_ID, exp)})
+    assert resp.status_code == 403
+    doc_get.assert_not_called()
+
+
+def test_missing_object_is_404(signed):
+    obj = FakeObject(b"")
+
+    async def no_size(gcs_uri):
+        return None
+
+    with patch.object(media.fstore, "doc_get", AsyncMock(return_value={"call_id": CALL_ID, "audio_gcs_uri": GCS_URI})), \
+            patch.object(media, "audio_object_size", no_size), \
+            patch.object(media, "download_audio_range", obj.read):
+        resp = client.get(f"/media/calls/{CALL_ID}/audio", params=signed)
+    assert resp.status_code == 404
+
+
+def test_legacy_mp3_keeps_its_content_type(signed):
+    obj = FakeObject(b"ID3" + b"\0" * 50)
+    resp, _ = _run(obj, signed, {"Range": "bytes=0-2"}, doc={"call_id": CALL_ID, "audio_gcs_uri": "gs://b/calls/x.mp3"})
+    assert resp.status_code == 206
+    assert resp.headers["content-type"] == "audio/mpeg"
+    assert resp.content == b"ID3"
+
+
+# ── storage.py: the ranged reads never pull the whole object ────────────────
+def _fake_client(monkeypatch, size=1234, payload=b"span"):
+    from unittest.mock import MagicMock
+    client = MagicMock()
+    blob = client.bucket.return_value.blob.return_value
+    blob.size = size
+    blob.download_as_bytes.return_value = payload
+    monkeypatch.setattr(storage, "_gcs_client_cache", client)
+    return client, blob
+
+
+def test_storage_size_reloads_metadata_only(monkeypatch):
+    import asyncio
+    client, blob = _fake_client(monkeypatch, size=13_000_000)
+    assert asyncio.run(storage.audio_object_size(GCS_URI)) == 13_000_000
+    client.bucket.assert_called_with("bucket")
+    client.bucket.return_value.blob.assert_called_with("calls/call-abc.flac")
+    blob.reload.assert_called_once()
+    blob.download_as_bytes.assert_not_called()
+
+
+def test_storage_range_read_passes_inclusive_bounds(monkeypatch):
+    import asyncio
+    _, blob = _fake_client(monkeypatch)
+    assert asyncio.run(storage.download_audio_range(GCS_URI, 100, 199)) == b"span"
+    blob.download_as_bytes.assert_called_once_with(start=100, end=199)
+
+
+def test_storage_range_read_failure_returns_none(monkeypatch):
+    import asyncio
+    _, blob = _fake_client(monkeypatch)
+    blob.download_as_bytes.side_effect = RuntimeError("gcs down")
+    assert asyncio.run(storage.download_audio_range(GCS_URI, 0, 10)) is None
+    blob.reload.side_effect = RuntimeError("404")
+    assert asyncio.run(storage.audio_object_size(GCS_URI)) is None