From c836986b895eba2a77162e6eb07f9d2fbe7f381d Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 5 Oct 2026 02:33:25 -0400 Subject: [PATCH] Click-to-listen on the live wall + HTTP Range on /media (server-26#233) Backend: /media/calls/{id}/audio now honours a single HTTP Range (bytes=a-b, a-, -n) with 206 + Content-Range, 416 for unsatisfiable, and a 200 full body (Accept-Ranges: bytes) otherwise. The body is streamed from GCS in 512 KiB span reads (storage.audio_object_size + download_audio_range) instead of downloading the whole object first, so a 600 s FLAC starts after one span and seeks fetch only what they need. HMAC check and Firestore lookup unchanged and still first. Frontend: InlinePlayer moved to components/InlinePlayer.tsx and now drives lib/audioQueue.ts, a useSyncExternalStore store over ONE audio element (one call plays at a time). Wall spine rows get play buttons and a "Play incident" queue (oldest first) with the playing row highlighted; map path-stop popups get a play button and the playing stop gets an accent ring; rail cards get "play latest". Playback is gated by canPlayAudio() (BUSINESS_MODEL section 3 / Gate A hook). Nothing plays until clicked. Spine truncation is now stated ("showing latest 30 of M", show-all toggle, count outside the loaded window) for server-26#239. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/storage.py | 61 ++++++ drb-c2-core/app/routers/media.py | 147 +++++++++++-- drb-c2-core/tests/test_media_range.py | 220 ++++++++++++++++++++ drb-frontend/components/CallSpineEntry.tsx | 98 +-------- drb-frontend/components/InlinePlayer.tsx | 89 ++++++++ drb-frontend/components/MapView.tsx | 81 ++++--- drb-frontend/components/live/WallPanels.tsx | 60 +++++- drb-frontend/lib/audioQueue.ts | 176 ++++++++++++++++ 8 files changed, 787 insertions(+), 145 deletions(-) create mode 100644 drb-c2-core/tests/test_media_range.py create mode 100644 drb-frontend/components/InlinePlayer.tsx create mode 100644 drb-frontend/lib/audioQueue.ts diff --git a/drb-c2-core/app/internal/storage.py b/drb-c2-core/app/internal/storage.py index 18af633..deea51f 100644 --- a/drb-c2-core/app/internal/storage.py +++ b/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://"): diff --git a/drb-c2-core/app/routers/media.py b/drb-c2-core/app/routers/media.py index 67a35ad..e01b1ae 100644 --- a/drb-c2-core/app/routers/media.py +++ b/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, ) diff --git a/drb-c2-core/tests/test_media_range.py b/drb-c2-core/tests/test_media_range.py new file mode 100644 index 0000000..609399f --- /dev/null +++ b/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 diff --git a/drb-frontend/components/CallSpineEntry.tsx b/drb-frontend/components/CallSpineEntry.tsx index d939c1b..b2635af 100644 --- a/drb-frontend/components/CallSpineEntry.tsx +++ b/drb-frontend/components/CallSpineEntry.tsx @@ -8,105 +8,15 @@ * memorable thing per §2.4: "where was he when he said that" is answered * by looking, not cross-referencing timestamps. */ -import { useEffect, useRef, useState } from "react"; +import { useState } from "react"; import type { CallRecord } from "@/lib/types"; -import { c2api } from "@/lib/c2api"; +import { InlinePlayer } from "@/components/InlinePlayer"; +import { hasAudio as callHasAudio } from "@/lib/audioQueue"; import { isKnownSeverity, SEVERITY_COLORS } from "@/lib/severity"; function fmtTime(iso: string): string { return new Date(iso).toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", second: "2-digit" }); } -function fmtClock(s: number): string { - if (!Number.isFinite(s)) return "0:00"; - const m = Math.floor(s / 60); - const r = Math.floor(s % 60); - return `${m}:${r.toString().padStart(2, "0")}`; -} - -function InlinePlayer({ callId }: { callId: string }) { - const [url, setUrl] = useState(null); - const [error, setError] = useState(false); - const [loading, setLoading] = useState(false); - const [playing, setPlaying] = useState(false); - const [current, setCurrent] = useState(0); - const [duration, setDuration] = useState(0); - const audioRef = useRef(null); - - async function ensureUrl() { - if (url || loading) return; - setLoading(true); - try { - const full = await c2api.getCall(callId); - setUrl(full.audio_url ?? null); - if (!full.audio_url) setError(true); - } catch { - setError(true); - } finally { - setLoading(false); - } - } - - async function toggle() { - if (!url) { - await ensureUrl(); - return; - } - const el = audioRef.current; - if (!el) return; - if (playing) el.pause(); - else el.play(); - } - - // Once the URL lands, autoplay (the click that fetched it was the play intent). - useEffect(() => { - if (url && audioRef.current) audioRef.current.play().catch(() => {}); - }, [url]); - - if (error) { - return Audio unavailable; - } - - return ( -
e.stopPropagation()}> - - { - const t = Number(e.target.value); - if (audioRef.current) audioRef.current.currentTime = t; - setCurrent(t); - }} - className="flex-1 h-1 accent-accent" - disabled={!url} - /> - - {fmtClock(current)} - - {url && ( -
- ); -} function StopMarker({ stopNumber, color }: { stopNumber?: number; color: string }) { if (!stopNumber) { @@ -132,7 +42,7 @@ export function CallSpineEntry({ isAdmin?: boolean; }) { const [expanded, setExpanded] = useState(false); - const hasAudio = !!(call.audio_gcs_uri || call.audio_url); + const hasAudio = callHasAudio(call); const transcript = call.transcript_corrected || call.transcript; const color = isKnownSeverity(call.severity) ? SEVERITY_COLORS[call.severity] : "var(--ink-muted)"; const substantive = !!transcript || (call.units && call.units.length > 0) || !!call.location_coords; diff --git a/drb-frontend/components/InlinePlayer.tsx b/drb-frontend/components/InlinePlayer.tsx new file mode 100644 index 0000000..2a3b88d --- /dev/null +++ b/drb-frontend/components/InlinePlayer.tsx @@ -0,0 +1,89 @@ +"use client"; + +/** + * Click-to-listen control for one call — server-26#233. Moved out of + * CallSpineEntry.tsx so the live wall's spine, the map's path-stop popups and + * rail cards share it. It no longer owns an