feat: smart gap detection with coverage windows - track exact time ranges captured per stream

local_worker/worker.py

@@ -328,6 +328,51 @@ def notify_stream_end():
     except Exception as e:
         print(f"[Sync] Failed to send stream-end: {e}")
 
+# Coverage window tracking
+coverage_id = None
+coverage_stop_flag = threading.Event()
+
+def start_coverage(stream_id, source='live'):
+    """Register a new coverage window with the backend"""
+    global coverage_id
+    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
+    try:
+        res = requests.post(f"{API_URL}/api/log/coverage/start",
+                            json={"streamId": stream_id, "source": source},
+                            headers=headers, timeout=5)
+        if res.status_code == 200:
+            coverage_id = res.json().get("coverageId")
+            print(f"[Coverage] Started window #{coverage_id} for stream {stream_id}")
+    except Exception as e:
+        print(f"[Coverage] Error starting coverage: {e}")
+
+def stop_coverage():
+    """Finalize the current coverage window"""
+    global coverage_id
+    if not coverage_id:
+        return
+    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
+    try:
+        requests.post(f"{API_URL}/api/log/coverage/end",
+                      json={"coverageId": coverage_id},
+                      headers=headers, timeout=5)
+        print(f"[Coverage] Ended window #{coverage_id}")
+    except Exception as e:
+        print(f"[Coverage] Error ending coverage: {e}")
+    coverage_id = None
+
+def coverage_heartbeat_loop():
+    """Background thread: ping coverage heartbeat every 2 minutes"""
+    headers = {"x-api-key": API_KEY}
+    while not coverage_stop_flag.is_set():
+        time.sleep(120)
+        if coverage_id:
+            try:
+                requests.patch(f"{API_URL}/api/log/coverage/{coverage_id}/heartbeat",
+                               headers=headers, timeout=5)
+            except Exception:
+                pass
+
 # =========================================================================
 # CAPTURE METHODS
 # =========================================================================
@@ -389,76 +434,63 @@ def run_stream_capture(model):
 
                         start_epoch = parse_time(start_time)
                         
-                        # -- CHAT BACKFILL TASKS --
-                        chat_tasks = []
-                        if not min_msg_time:
-                            # 0 messages, download everything
-                            chat_tasks.append((0, None))
-                        else:
-                            min_msg_epoch = parse_time(min_msg_time)
-                            max_msg_epoch = parse_time(max_msg_time)
-                            
-                            min_msg_offset = int(min_msg_epoch - start_epoch) if (min_msg_epoch and start_epoch) else 0
-                            max_msg_offset = int(max_msg_epoch - start_epoch) if (max_msg_epoch and start_epoch) else 0
-                            
-                            # If first message is > 60s from start, download the beginning gap
-                            if min_msg_offset > 60:
-                                chat_tasks.append((0, min_msg_offset))
-                            # Download from last message to end
-                            chat_tasks.append((max_msg_offset, None))
-
-                        # -- VOICE BACKFILL TASKS --
-                        voice_tasks = []
-                        if not min_voice_time:
-                            # 0 words, transcribe everything
-                            voice_tasks.append((0, None))
-                        else:
-                            min_voice_epoch = parse_time(min_voice_time)
-                            max_voice_epoch = parse_time(max_voice_time)
-                            
-                            min_voice_offset = int(min_voice_epoch - start_epoch) if (min_voice_epoch and start_epoch) else 0
-                            max_voice_offset = int(max_voice_epoch - start_epoch) if (max_voice_epoch and start_epoch) else 0
-                            
-                            # If first voice word is > 60s from start, transcribe the beginning gap
-                            if min_voice_offset > 60:
-                                voice_tasks.append((0, min_voice_offset))
-                            # Transcribe from last word (with 10s safety overlap) to end
-                            voice_tasks.append((max(0, max_voice_offset - 10), None))
-
-                        print(f"\n=========================================================")
-                        print(f"[Backfill] Processing: {title}")
-                        print(f"[Backfill] VOD ID:       {vod_id}")
-                        print(f"[Backfill] Stream ID:   {stream_id}")
-                        print(f"[Backfill] Chat Tasks:  {chat_tasks}")
-                        print(f"[Backfill] Voice Tasks: {voice_tasks}")
-                        print(f"=========================================================")
-                        
-                        # 1. Download and upload chat segments
-                        for start_off, end_off in chat_tasks:
-                            print(f"[Backfill] Downloading chat from {start_off}s to {'end' if end_off is None else str(end_off) + 's'}...")
-                            chat_comments = download_vod_chat(vod_id, start_offset=start_off, end_offset=end_off)
+                        # Notify start of coverage for this backfill session
+                        start_coverage(stream_id, source='backfill')
+
+                        gaps = target.get("gaps", [])
+                        if not gaps:
+                            print(f"[Backfill] No gaps found for stream {stream_id}, marking complete.")
+                            try:
+                                requests.post(f"{API_URL}/api/streams/mark-backfilled",
+                                              json={"streamId": stream_id}, headers=headers_get, timeout=10)
+                            except Exception as e:
+                                print(f"[Backfill] Error marking complete: {e}")
+                            stop_coverage()
+                            continue
+
+                        print(f"[Backfill] Found {len(gaps)} gap(s) to fill:")
+                        for g in gaps:
+                            to_str = str(g['to_offset']) + 's' if g['to_offset'] is not None else 'end'
+                            print(f"  Gap: {g['from_offset']}s → {to_str}")
+
+                        # Download and upload each gap in order
+                        for gap_idx, gap in enumerate(gaps, 1):
+                            from_off = gap.get('from_offset', 0)
+                            to_off   = gap.get('to_offset', None)
+                            to_str   = str(to_off) + 's' if to_off is not None else 'end'
+                            print(f"\n[Backfill] Gap {gap_idx}/{len(gaps)}: chat {from_off}s → {to_str}")
+
+                            chat_comments = download_vod_chat(vod_id, start_offset=from_off, end_offset=to_off)
                             if chat_comments:
-                                print(f"[Backfill] Uploading {len(chat_comments)} chat comments...")
+                                print(f"[Backfill] Uploading {len(chat_comments)} messages...")
                                 batch_size = 100
                                 headers_post = {"x-api-key": API_KEY, "Content-Type": "application/json"}
                                 for i in range(0, len(chat_comments), batch_size):
                                     batch = chat_comments[i:i+batch_size]
                                     try:
-                                        requests.post(f"{API_URL}/api/log/messages", json={"messages": batch}, headers=headers_post, timeout=10)
+                                        requests.post(f"{API_URL}/api/log/messages",
+                                                      json={"messages": batch},
+                                                      headers=headers_post, timeout=10)
                                     except Exception as e:
                                         print(f"[Backfill] Chat batch upload error: {e}")
-                        
-                        # 2. Download and transcribe audio segments
-                        print(f"[Backfill] Starting Whisper audio transcription...")
-                        for start_off, end_off in voice_tasks:
-                            print(f"[Backfill] Transcribing audio from {start_off}s to {'end' if end_off is None else str(end_off) + 's'}...")
+                            else:
+                                print(f"[Backfill] No chat in this gap.")
+
+                        # Transcribe audio gaps
+                        print(f"[Backfill] Starting Whisper audio transcription for gaps...")
+                        for gap_idx, gap in enumerate(gaps, 1):
+                            from_off = gap.get('from_offset', 0)
+                            to_off   = gap.get('to_offset', None)
+                            to_str   = str(to_off) + 's' if to_off is not None else 'end'
+                            print(f"[Backfill] Audio gap {gap_idx}/{len(gaps)}: {from_off}s → {to_str}")
                             try:
                                 start_time_iso = start_time.replace("+00:00", "").replace("Z", "") + "Z"
-                                transcribe_vod_audio(vod_id, start_time_iso, model, start_offset=start_off, end_offset=end_off)
+                                transcribe_vod_audio(vod_id, start_time_iso, model,
+                                                     start_offset=from_off, end_offset=to_off)
                             except Exception as e:
                                 print(f"[Backfill] Transcription error: {e}")
                         
-                        # 3. Mark completed
+                        # 3. Mark completed and end coverage
                         print(f"[Backfill] Marking stream {stream_id} as backfilled...")
                         try:
                             res_mark = requests.post(f"{API_URL}/api/streams/mark-backfilled", json={"streamId": stream_id}, headers=headers_get, timeout=10)
@@ -468,6 +500,7 @@ def run_stream_capture(model):
                                 print(f"[Backfill] Failed to mark stream: {res_mark.status_code} - {res_mark.text}")
                         except Exception as e:
                             print(f"[Backfill] Network error marking stream: {e}")
+                        stop_coverage()
                         
                         # Loop again immediately to process next VOD or check if live
                         continue
@@ -600,13 +633,28 @@ if __name__ == "__main__":
     # 2. Start Twitch Chat background threads
     t_chat = threading.Thread(target=twitch_chat_listener, daemon=True)
     t_sender = threading.Thread(target=chat_sender, daemon=True)
+    t_heartbeat = threading.Thread(target=coverage_heartbeat_loop, daemon=True)
     
     t_chat.start()
     t_sender.start()
+    t_heartbeat.start()
     
-    # Notify backend that stream has started
+    # Notify backend that stream has started and get stream id for coverage
     notify_stream_start()
     
+    # Start coverage window for the active stream
+    try:
+        import requests as _req
+        _headers = {"x-api-key": API_KEY}
+        _stream_res = _req.get(f"{API_URL}/api/streams/active", headers=_headers, timeout=5)
+        if _stream_res.status_code == 200:
+            _active = _stream_res.json()
+            _sid = _active.get("stream", {}).get("id") or _active.get("id")
+            if _sid:
+                start_coverage(_sid, source='live')
+    except Exception as _e:
+        print(f"[Coverage] Could not get active stream for coverage: {_e}")
+    
     # 3. Start Audio Capture (Blocks main thread)
     try:
         if CAPTURE_METHOD == "stream":
@@ -618,7 +666,9 @@ if __name__ == "__main__":
     except KeyboardInterrupt:
         print("\n[Shutting Down] Gracefully stopping threads...")
     finally:
-        # Notify backend that stream has ended
+        # End coverage window and notify stream ended
+        stop_coverage()
+        coverage_stop_flag.set()
         notify_stream_end()
         stop_flag.set()
         time.sleep(1)

server/db.js

@@ -1573,6 +1573,160 @@ export async function getModeratorProfilesData(streamId = null) {
   return profiles.sort((a, b) => b.total_actions - a.total_actions);
 }
 
+// ============================================================
+// COVERAGE WINDOWS — track which time ranges were captured
+// ============================================================
+
+/**
+ * Start a new coverage window (worker session started)
+ */
+export async function startCoverageWindow(streamId, source = 'live') {
+  const now = new Date().toISOString();
+  if (dbMode === 'sqlite') {
+    const result = sqliteDb.prepare(
+      'INSERT INTO stream_coverage (stream_id, covered_from, source) VALUES (?, ?, ?)'
+    ).run(streamId, now, source);
+    return result.lastInsertRowid;
+  } else {
+    const { data, error } = await supabase
+      .from('stream_coverage')
+      .insert({ stream_id: streamId, covered_from: now, source })
+      .select('id')
+      .single();
+    if (error) console.error('[Supabase] startCoverageWindow error:', error.message);
+    return data?.id || null;
+  }
+}
+
+/**
+ * Update coverage heartbeat (worker still running)
+ */
+export async function updateCoverageHeartbeat(coverageId) {
+  const now = new Date().toISOString();
+  if (dbMode === 'sqlite') {
+    sqliteDb.prepare('UPDATE stream_coverage SET covered_to = ? WHERE id = ?').run(now, coverageId);
+  } else {
+    await supabase.from('stream_coverage').update({ covered_to: now }).eq('id', coverageId);
+  }
+}
+
+/**
+ * End a coverage window (worker session stopped)
+ */
+export async function endCoverageWindow(coverageId) {
+  const now = new Date().toISOString();
+  if (dbMode === 'sqlite') {
+    sqliteDb.prepare('UPDATE stream_coverage SET covered_to = ? WHERE id = ?').run(now, coverageId);
+  } else {
+    await supabase.from('stream_coverage').update({ covered_to: now }).eq('id', coverageId);
+  }
+}
+
+/**
+ * Fetch VOD duration in seconds via Twitch GQL
+ */
+async function getVodDurationSeconds(vodId) {
+  try {
+    const res = await fetch('https://gql.twitch.tv/gql', {
+      method: 'POST',
+      headers: {
+        'Client-Id': 'kimne78kx3ncx6brgo4mv6wki5h1ko',
+        'Content-Type': 'application/json'
+      },
+      body: JSON.stringify({
+        query: `query { video(id: "${vodId}") { lengthSeconds } }`
+      })
+    });
+    const data = await res.json();
+    return data?.data?.video?.lengthSeconds || null;
+  } catch (e) {
+    console.error('[GQL] getVodDurationSeconds error:', e.message);
+    return null;
+  }
+}
+
+/**
+ * Compute uncovered gaps for a stream given its coverage windows.
+ * Returns array of { from_offset, to_offset } in seconds from stream start.
+ */
+export async function getCoverageGaps(streamId, vodId, streamStartTime) {
+  const startEpoch = new Date(streamStartTime).getTime() / 1000;
+
+  // 1. Get VOD duration
+  const durationSeconds = vodId ? await getVodDurationSeconds(vodId) : null;
+  const vodEnd = durationSeconds != null ? startEpoch + durationSeconds : null;
+
+  // 2. Get all coverage windows for this stream
+  let windows = [];
+  if (dbMode === 'sqlite') {
+    windows = sqliteDb.prepare(
+      'SELECT covered_from, covered_to FROM stream_coverage WHERE stream_id = ? AND covered_to IS NOT NULL ORDER BY covered_from ASC'
+    ).all(streamId);
+  } else {
+    const { data } = await supabase
+      .from('stream_coverage')
+      .select('covered_from, covered_to')
+      .eq('stream_id', streamId)
+      .not('covered_to', 'is', null)
+      .order('covered_from', { ascending: true });
+    windows = data || [];
+  }
+
+  if (windows.length === 0) {
+    // No coverage at all — backfill entire VOD
+    if (vodEnd) return [{ from_offset: 0, to_offset: durationSeconds }];
+    return [{ from_offset: 0, to_offset: null }];
+  }
+
+  // 3. Convert to [start_offset, end_offset] pairs (seconds from stream start)
+  const intervals = windows.map(w => ({
+    s: Math.max(0, Math.round(new Date(w.covered_from).getTime() / 1000 - startEpoch)),
+    e: Math.round(new Date(w.covered_to).getTime() / 1000 - startEpoch)
+  })).filter(w => w.e > w.s);
+
+  // 4. Sort and merge overlapping intervals
+  intervals.sort((a, b) => a.s - b.s);
+  const merged = [];
+  for (const iv of intervals) {
+    if (merged.length === 0 || iv.s > merged[merged.length - 1].e + 30) {
+      merged.push({ ...iv });
+    } else {
+      merged[merged.length - 1].e = Math.max(merged[merged.length - 1].e, iv.e);
+    }
+  }
+
+  // 5. Find gaps
+  const gaps = [];
+  const THRESHOLD = 60; // ignore gaps < 60 seconds
+
+  // Gap before first coverage window
+  if (merged[0].s > THRESHOLD) {
+    gaps.push({ from_offset: 0, to_offset: merged[0].s });
+  }
+
+  // Gaps between windows
+  for (let i = 0; i < merged.length - 1; i++) {
+    const gapSize = merged[i + 1].s - merged[i].e;
+    if (gapSize > THRESHOLD) {
+      gaps.push({ from_offset: merged[i].e, to_offset: merged[i + 1].s });
+    }
+  }
+
+  // Gap after last coverage window (to end of VOD)
+  const lastEnd = merged[merged.length - 1].e;
+  if (vodEnd) {
+    const remainingSeconds = durationSeconds - lastEnd;
+    if (remainingSeconds > THRESHOLD) {
+      gaps.push({ from_offset: lastEnd, to_offset: durationSeconds });
+    }
+  } else {
+    // Unknown VOD duration — backfill from last known point to end
+    gaps.push({ from_offset: lastEnd, to_offset: null });
+  }
+
+  return gaps;
+}
+
 /**
  * Get list of VOD streams waiting for backfill
  */
@@ -1580,15 +1734,7 @@ export async function getPendingBackfillStreams() {
   if (dbMode === 'sqlite') {
     const streams = sqliteDb.prepare("SELECT * FROM streams WHERE backfill_status = 'pending' AND twitch_vod_id IS NOT NULL ORDER BY start_time ASC").all();
     for (const stream of streams) {
-      const msgMaxRow = sqliteDb.prepare("SELECT max(timestamp) as max_time FROM messages WHERE stream_id = ?").get(stream.id);
-      const msgMinRow = sqliteDb.prepare("SELECT min(timestamp) as min_time FROM messages WHERE stream_id = ?").get(stream.id);
-      const voiceMaxRow = sqliteDb.prepare("SELECT max(timestamp) as max_time FROM voice_words WHERE stream_id = ?").get(stream.id);
-      const voiceMinRow = sqliteDb.prepare("SELECT min(timestamp) as min_time FROM voice_words WHERE stream_id = ?").get(stream.id);
-
-      stream.max_message_time = msgMaxRow ? msgMaxRow.max_time : null;
-      stream.min_message_time = msgMinRow ? msgMinRow.min_time : null;
-      stream.max_voice_time = voiceMaxRow ? voiceMaxRow.max_time : null;
-      stream.min_voice_time = voiceMinRow ? voiceMinRow.min_time : null;
+      stream.gaps = await getCoverageGaps(stream.id, stream.twitch_vod_id, stream.start_time);
     }
     return streams;
   } else {
@@ -1606,42 +1752,7 @@ export async function getPendingBackfillStreams() {
 
     if (streams) {
       for (const stream of streams) {
-        // Query max message timestamp in Supabase
-        const { data: msgMaxData } = await supabase
-          .from('messages')
-          .select('timestamp')
-          .eq('stream_id', stream.id)
-          .order('timestamp', { ascending: false })
-          .limit(1);
-
-        // Query min message timestamp in Supabase
-        const { data: msgMinData } = await supabase
-          .from('messages')
-          .select('timestamp')
-          .eq('stream_id', stream.id)
-          .order('timestamp', { ascending: true })
-          .limit(1);
-        
-        // Query max voice timestamp in Supabase
-        const { data: voiceMaxData } = await supabase
-          .from('voice_words')
-          .select('timestamp')
-          .eq('stream_id', stream.id)
-          .order('timestamp', { ascending: false })
-          .limit(1);
-
-        // Query min voice timestamp in Supabase
-        const { data: voiceMinData } = await supabase
-          .from('voice_words')
-          .select('timestamp')
-          .eq('stream_id', stream.id)
-          .order('timestamp', { ascending: true })
-          .limit(1);
-
-        stream.max_message_time = msgMaxData && msgMaxData.length > 0 ? msgMaxData[0].timestamp : null;
-        stream.min_message_time = msgMinData && msgMinData.length > 0 ? msgMinData[0].timestamp : null;
-        stream.max_voice_time = voiceMaxData && voiceMaxData.length > 0 ? voiceMaxData[0].timestamp : null;
-        stream.min_voice_time = voiceMinData && voiceMinData.length > 0 ? voiceMinData[0].timestamp : null;
+        stream.gaps = await getCoverageGaps(stream.id, stream.twitch_vod_id, stream.start_time);
       }
     }
     return streams || [];

server/server.js

@@ -34,7 +34,10 @@ import {
   deleteStream,
   updateStreamMetadata,
   getSystemStats,
-  cleanupGhostStreams
+  cleanupGhostStreams,
+  startCoverageWindow,
+  updateCoverageHeartbeat,
+  endCoverageWindow
 } from './db.js';
 // import { initializeEventSub } from './eventsub.js';
 import { cache, cacheMiddleware } from './cache.js';
@@ -277,6 +280,44 @@ app.get('/api/streams/pending-backfill', authenticateWorker, async (req, res) =>
   }
 });
 
+// POST start coverage window
+app.post('/api/log/coverage/start', authenticateWorker, async (req, res) => {
+  const { streamId, source } = req.body;
+  if (!streamId) return res.status(400).json({ error: 'Missing streamId' });
+  try {
+    const coverageId = await startCoverageWindow(streamId, source || 'live');
+    res.json({ success: true, coverageId });
+  } catch (err) {
+    console.error('[API Error] /api/log/coverage/start:', err);
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+});
+
+// PATCH heartbeat coverage window
+app.patch('/api/log/coverage/:id/heartbeat', authenticateWorker, async (req, res) => {
+  const coverageId = parseInt(req.params.id);
+  try {
+    await updateCoverageHeartbeat(coverageId);
+    res.json({ success: true });
+  } catch (err) {
+    console.error('[API Error] /api/log/coverage/heartbeat:', err);
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+});
+
+// POST end coverage window
+app.post('/api/log/coverage/end', authenticateWorker, async (req, res) => {
+  const { coverageId } = req.body;
+  if (!coverageId) return res.status(400).json({ error: 'Missing coverageId' });
+  try {
+    await endCoverageWindow(coverageId);
+    res.json({ success: true });
+  } catch (err) {
+    console.error('[API Error] /api/log/coverage/end:', err);
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+});
+
 // POST mark stream as backfilled
 app.post('/api/streams/mark-backfilled', authenticateWorker, async (req, res) => {
   const { streamId, twitchStreamId } = req.body;

server/supabase_migration_coverage.sql

@@ -0,0 +1,14 @@
+-- Migration: Add stream_coverage table for smart gap detection
+-- Run this in Supabase SQL editor
+
+CREATE TABLE IF NOT EXISTS stream_coverage (
+  id BIGSERIAL PRIMARY KEY,
+  stream_id INTEGER REFERENCES streams(id) ON DELETE CASCADE,
+  covered_from TIMESTAMPTZ NOT NULL,
+  covered_to   TIMESTAMPTZ,
+  source TEXT NOT NULL DEFAULT 'live',  -- 'live' | 'backfill' | 'vod_backfiller'
+  created_at TIMESTAMPTZ DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_stream_coverage_stream_id ON stream_coverage(stream_id);
+CREATE INDEX IF NOT EXISTS idx_stream_coverage_covered_from ON stream_coverage(covered_from);