Improve History's seek startup time and recordings query performance (#24011)

* serve a segment startup ladder so seeks begin playing sooner

nginx-vod was handed one 10s segment per recording file, so every playlist start had to download and decode a full segment before the first frame. Declare real keyframe data per clip and let nginx cut short leading segments from it.

- add vod_bootstrap_segment_durations 1000/2000/4000 so each playlist starts with 1s/2s/4s segments before settling at 10s
- emit real clip-relative keyFrameDurations (plus firstKeyFrameOffset when nonzero) from the recording keyframe index; rows without an index keep the whole-clip declaration, the only safe cut without keyframe knowledge
- drop the manifest's segment_duration field, which was always inert: nginx-vod parses only camelCase segmentDuration
- rebuild the player source at the seek target, quantized to a 10s grid, so the ladder applies to every seek and seek URLs stay repeatable for nginx's mapping and response caches
- route the seek model, in-range checks, and the stale-report guard through the source window rather than the chunk range
- bridge repositioning seeks (>2s from the last played timestamp) through the preview player and hold the release anchor one commit, so neither path paints a stale frame
- clear a pending loading timer before replacing it; an orphaned timer escaped onPlaying's clearTimeout and flashed loading mid-playback

* keep recordings queries on their indexes

Several recordings queries degraded into full scans or large sorts on big databases: the planner ignored index order, or the query shape gave it nothing tight to seek on. Reshape them into bounded seeks and add the composite index the per-stream lookups need.

- index recordings on (camera, stream_type, start_time DESC) and drop the (camera, stream_type) index it supersedes
- walk the recordings summary day by day with EXISTS probes and per-camera MIN/MAX seeks, skipping ahead over empty gaps instead of bucketing every row for the requested cameras
- run the summary endpoint on the event loop rather than the threadpool
- bound the unavailable-recordings query by start_time per camera and merge the results in Python
- bound the expire query's start_time so it seeks the retention window instead of scanning a camera's whole history
- enumerate deleted cameras with one index seek each rather than a camera NOT IN (...) scan
- compute bandwidth with segment_size filtered in a CASE projection; as a WHERE predicate it baited the planner into the (camera, segment_size) index plus a full sort of the camera's history
- fall back to a 1000-segment window when the recent 100 are all zero-size, so an ingest glitch doesn't report zero bandwidth
- limit the needs_refresh count instead of counting every segment
- cover sub-only and sparse calendar days, midnight-spanning day attribution, multi-camera gap merging, deleted-camera expiry, and zero-size segment runs

* fix mypy

frigate/storage.py

@@ -6,7 +6,7 @@ import threading
 from multiprocessing.synchronize import Event as MpEvent
 from pathlib import Path
 
-from peewee import SQL, fn
+from peewee import SQL, Case, fn
 
 from frigate.config import FrigateConfig
 from frigate.const import (
@@ -35,6 +35,35 @@ class StorageMaintainer(threading.Thread):
         self.stop_event = stop_event
         self.camera_storage_stats: dict[str, dict] = {}
 
+    def _recent_stream_bandwidth(
+        self, camera: str, stream_type: str, window: int
+    ) -> float | None:
+        """Average MB/s over a stream's most recent rows, or None if no sample.
+
+        Zero-size rows are excluded inside the projection, not the WHERE
+        clause: a segment_size predicate baits the planner into the
+        (camera, segment_size) index plus a full sort of the camera's
+        history instead of the time-ordered index.
+        """
+        recent = (
+            Recordings.select(
+                Case(
+                    None,
+                    [(Recordings.segment_size > 0, bandwidth_equation)],
+                    None,
+                ).alias("bw")
+            )
+            .where(
+                Recordings.camera == camera,
+                Recordings.stream_type == stream_type,
+            )
+            .order_by(Recordings.start_time.desc())
+            .limit(window)
+            .alias("recent")
+        )
+        avg: float | None = Recordings.select(fn.AVG(SQL("bw"))).from_(recent).scalar()
+        return avg
+
     def calculate_camera_bandwidth(self) -> None:
         """Calculate an average MB/hr for each camera."""
         for camera in self.config.cameras.keys():
@@ -47,9 +76,10 @@ class StorageMaintainer(threading.Thread):
             if self.camera_storage_stats.get(camera, {}).get("needs_refresh", True):
                 self.camera_storage_stats[camera] = {
                     "needs_refresh": (
-                        Recordings.select(fn.COUNT("*"))
+                        Recordings.select(Recordings.id)
                         .where(Recordings.camera == camera, Recordings.segment_size > 0)
-                        .scalar()
+                        .limit(50)
+                        .count()
                         < 50
                     )
                 }
@@ -58,31 +88,18 @@ class StorageMaintainer(threading.Thread):
                 # type and sum the rates; mixing streams would average small
                 # sub segments against large main segments and underestimate
                 # the true write rate
-                bandwidth = 0
+                bandwidth = 0.0
                 for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB):
-                    try:
-                        # Subquery to get last 100 segments, then average their bandwidth
-                        last_100 = (
-                            Recordings.select(bandwidth_equation.alias("bw"))
-                            .where(
-                                Recordings.camera == camera,
-                                Recordings.segment_size > 0,
-                                Recordings.stream_type == stream_type,
-                            )
-                            .order_by(Recordings.start_time.desc())
-                            .limit(100)
-                            .alias("recent")
+                    avg_bw = self._recent_stream_bandwidth(camera, stream_type, 100)
+                    if avg_bw is None:
+                        # the recent window can be all zero-size ingest
+                        # glitches; look further back before concluding
+                        # the stream writes nothing
+                        avg_bw = self._recent_stream_bandwidth(
+                            camera, stream_type, 1000
                         )
-
-                        bandwidth += round(
-                            Recordings.select(fn.AVG(SQL("bw")))
-                            .from_(last_100)
-                            .scalar()
-                            * 3600,
-                            2,
-                        )
-                    except TypeError:
-                        pass
+                    if avg_bw is not None:
+                        bandwidth += round(avg_bw * 3600, 2)
 
                 bandwidth = round(bandwidth, 2)