feat: automatic VOD ID resolution for stream sessions and historical pending streams

local_worker/worker.py

@@ -319,13 +319,130 @@ def notify_stream_start():
     except Exception as e:
         print(f"[Sync] Error sending stream-start: {e}")
 
-def notify_stream_end():
-    """Notify backend that the stream has ended"""
+def get_recent_twitch_vods(channel_name):
+    """Fetch recent Twitch VODs of a user using public Twitch GQL API"""
+    url = "https://gql.twitch.tv/gql"
+    headers = {
+        "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko",
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
+    }
+    payload = {
+        "query": """
+        query($login: String!) {
+            user(login: $login) {
+                videos(first: 10, type: ARCHIVE) {
+                    edges {
+                        node {
+                            id
+                            title
+                            createdAt
+                            lengthSeconds
+                        }
+                    }
+                }
+            }
+        }
+        """,
+        "variables": {
+            "login": channel_name
+        }
+    }
+    try:
+        res = requests.post(url, json=payload, headers=headers, timeout=10)
+        if res.status_code == 200:
+            data = res.json()
+            edges = data.get("data", {}).get("user", {}).get("videos", {}).get("edges", [])
+            return [edge.get("node") for edge in edges if edge and edge.get("node")]
+    except Exception as e:
+        print(f"[VOD Resolve] Error fetching Twitch VODs: {e}")
+    return []
+
+def find_matching_vod(stream_start_time_iso, recent_vods):
+    try:
+        from datetime import datetime
+        clean_stream = stream_start_time_iso.replace('Z', '+00:00')
+        stream_dt = datetime.fromisoformat(clean_stream)
+    except Exception as e:
+        print(f"[Match] Error parsing stream start time: {e}")
+        return None
+
+    best_match = None
+    min_diff = float('inf')
+
+    for node in recent_vods:
+        vod_id = node.get("id")
+        created_at_raw = node.get("createdAt")
+        if not vod_id or not created_at_raw:
+            continue
+        try:
+            clean_vod = created_at_raw.replace('Z', '+00:00')
+            vod_dt = datetime.fromisoformat(clean_vod)
+        except Exception:
+            continue
+        
+        diff = abs((stream_dt - vod_dt).total_seconds())
+        # If the start times are within 2.5 hours (9000 seconds)
+        if diff < 9000 and diff < min_diff:
+            min_diff = diff
+            best_match = vod_id
+
+    return best_match
+
+def resolve_missing_vods():
+    """Find and resolve twitch_vod_id for pending streams that are missing it"""
     headers = {"x-api-key": API_KEY}
+    try:
+        # 1. Fetch missing VOD streams from backend
+        res = requests.get(f"{API_URL}/api/streams/missing-vod", headers=headers, timeout=10)
+        if res.status_code != 200:
+            return
+        
+        streams = res.json().get("streams", [])
+        if not streams:
+            return
+            
+        print(f"[VOD Resolve] Found {len(streams)} pending stream(s) lacking VOD ID.")
+        
+        # 2. Get recent VODs from Twitch
+        recent_vods = get_recent_twitch_vods(TWITCH_CHANNEL)
+        if not recent_vods:
+            print("[VOD Resolve] Could not retrieve recent Twitch VODs. Skipping resolve.")
+            return
+            
+        # 3. Match each stream to a Twitch VOD
+        for stream in streams:
+            stream_id = stream.get("id")
+            start_time_str = stream.get("start_time")
+            if not stream_id or not start_time_str:
+                continue
+                
+            matched_vod_id = find_matching_vod(start_time_str, recent_vods)
+            if matched_vod_id:
+                print(f"[VOD Resolve] Stream ID {stream_id} ({start_time_str}) matched with Twitch VOD {matched_vod_id}.")
+                # Send update to server
+                up_res = requests.post(f"{API_URL}/api/streams/{stream_id}/resolve-vod", 
+                                       json={"twitchVodId": matched_vod_id}, 
+                                       headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
+                                       timeout=10)
+                if up_res.status_code == 200:
+                    print(f"[VOD Resolve] Successfully updated VOD ID for stream {stream_id}!")
+                else:
+                    print(f"[VOD Resolve] Failed to update VOD ID: {up_res.status_code}")
+            else:
+                print(f"[VOD Resolve] No matching VOD found for stream ID {stream_id} ({start_time_str}) within timeframe.")
+    except Exception as e:
+        print(f"[VOD Resolve] Exception during resolve loop: {e}")
+
+def notify_stream_end(vod_id=None):
+    """Notify backend that the stream has ended"""
+    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
     url = f"{API_URL}/api/log/stream-end"
+    payload = {}
+    if vod_id:
+        payload["twitchVodId"] = vod_id
     try:
-        requests.post(url, headers=headers, timeout=5)
-        print("[Sync] Sent stream-end signal.")
+        requests.post(url, json=payload, headers=headers, timeout=5)
+        print(f"[Sync] Sent stream-end signal. VOD ID resolved: {vod_id}")
     except Exception as e:
         print(f"[Sync] Failed to send stream-end: {e}")
 
@@ -415,6 +532,8 @@ def run_stream_capture(model):
     
     while not stop_flag.is_set():
         if not check_stream_live():
+            # Try to resolve any missing VOD IDs before checking pending backfills
+            resolve_missing_vods()
             print(f"[Capture] Channel '{TWITCH_CHANNEL}' is offline. Checking for pending VOD backfill tasks...")
             try:
                 headers_get = {"x-api-key": API_KEY}
@@ -620,7 +739,16 @@ def run_stream_capture(model):
                         try: p.kill() 
                         except: pass
             
-            notify_stream_end()
+            # Try to get the latest VOD ID from Twitch GQL to auto-resolve it
+            latest_vod_id = None
+            try:
+                vods = get_recent_twitch_vods(TWITCH_CHANNEL)
+                if vods:
+                    latest_vod_id = vods[0].get("id")
+            except Exception as e:
+                print(f"[Capture] Error pre-fetching latest VOD ID: {e}")
+            
+            notify_stream_end(latest_vod_id)
             print("[Capture] Pipeline stopped. Waiting 30 seconds before checking stream status...")
             time.sleep(30)
 
@@ -733,7 +861,18 @@ if __name__ == "__main__":
         # End coverage window and notify stream ended
         stop_coverage()
         coverage_stop_flag.set()
-        notify_stream_end()
+        
+        # Try to get the latest VOD ID from Twitch GQL to auto-resolve it
+        latest_vod_id = None
+        if CAPTURE_METHOD == "stream":
+            try:
+                vods = get_recent_twitch_vods(TWITCH_CHANNEL)
+                if vods:
+                    latest_vod_id = vods[0].get("id")
+            except Exception as e:
+                print(f"[Capture] Error pre-fetching latest VOD ID: {e}")
+        
+        notify_stream_end(latest_vod_id)
         stop_flag.set()
         time.sleep(1)
         print("[Shutting Down] Done.")

server/db.js

@@ -1759,6 +1759,48 @@ export async function getPendingBackfillStreams() {
   }
 }
 
+/**
+ * Get list of pending streams that are missing their twitch_vod_id
+ */
+export async function getStreamsMissingVod() {
+  if (dbMode === 'sqlite') {
+    return sqliteDb.prepare("SELECT * FROM streams WHERE backfill_status = 'pending' AND twitch_vod_id IS NULL ORDER BY start_time ASC").all();
+  } else {
+    const { data: streams, error } = await supabase
+      .from('streams')
+      .select('*')
+      .eq('backfill_status', 'pending')
+      .is('twitch_vod_id', null)
+      .order('start_time', { ascending: true });
+
+    if (error) {
+      console.error('[Supabase] Error getting streams missing VOD:', error);
+      return [];
+    }
+    return streams || [];
+  }
+}
+
+/**
+ * Update the twitch_vod_id of a stream
+ */
+export async function updateStreamVodId(streamId, twitchVodId) {
+  if (dbMode === 'sqlite') {
+    sqliteDb.prepare("UPDATE streams SET twitch_vod_id = ? WHERE id = ?").run(twitchVodId, streamId);
+    console.log(`[Database] SQLite Stream ID ${streamId} updated twitch_vod_id to ${twitchVodId}`);
+  } else {
+    const { error } = await supabase
+      .from('streams')
+      .update({ twitch_vod_id: twitchVodId })
+      .eq('id', streamId);
+    if (error) {
+      console.error('[Supabase] Error updating twitch_vod_id:', error);
+    } else {
+      console.log(`[Database] Supabase Stream ID ${streamId} updated twitch_vod_id to ${twitchVodId}`);
+    }
+  }
+}
+
 /**
  * Mark a stream as backfilled and completed
  */

server/server.js

@@ -28,6 +28,8 @@ import {
   getModeratorProfilesData,
   getActiveStream,
   getPendingBackfillStreams,
+  getStreamsMissingVod,
+  updateStreamVodId,
   markStreamBackfilled,
   checkIfStreamExists,
   resetStreamBackfill,
@@ -261,7 +263,8 @@ app.post('/api/log/stream-start', authenticateWorker, async (req, res) => {
 // Explicit end stream command
 app.post('/api/log/stream-end', authenticateWorker, async (req, res) => {
   try {
-    await endActiveStream();
+    const { twitchVodId } = req.body || {};
+    await endActiveStream(null, null, twitchVodId);
     res.json({ success: true });
   } catch (err) {
     console.error('[API Error] /api/log/stream-end:', err);
@@ -280,6 +283,30 @@ app.get('/api/streams/pending-backfill', authenticateWorker, async (req, res) =>
   }
 });
 
+// GET list of streams missing twitch_vod_id
+app.get('/api/streams/missing-vod', authenticateWorker, async (req, res) => {
+  try {
+    const streams = await getStreamsMissingVod();
+    res.json({ success: true, streams });
+  } catch (err) {
+    console.error('[API Error] /api/streams/missing-vod:', err);
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+});
+
+// POST to update twitch_vod_id for a stream
+app.post('/api/streams/:id/resolve-vod', authenticateWorker, async (req, res) => {
+  const { id } = req.params;
+  const { twitchVodId } = req.body;
+  try {
+    await updateStreamVodId(id, twitchVodId);
+    res.json({ success: true });
+  } catch (err) {
+    console.error('[API Error] /api/streams/resolve-vod:', err);
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+});
+
 // POST start coverage window
 app.post('/api/log/coverage/start', authenticateWorker, async (req, res) => {
   const { streamId, source, coveredFrom, coveredTo } = req.body;