feat: smart gap detection and user roles sync backfill improvements

local_worker/chat_mod_worker.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timedelta
 import os
 import sys
 import time
@@ -32,6 +33,65 @@ mod_action_queue = queue.Queue()
 # Flag to signal thread termination
 stop_flag = threading.Event()
 
+# Coverage window tracking
+coverage_id = None
+coverage_stop_flag = threading.Event()
+
+def start_coverage(stream_id, source='live', covered_from=None, covered_to=None):
+    """Register a new coverage window with the backend"""
+    global coverage_id
+    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
+    payload = {"streamId": stream_id, "source": source}
+    if covered_from:
+        payload["coveredFrom"] = covered_from
+    if covered_to:
+        payload["coveredTo"] = covered_to
+    try:
+        res = requests.post(f"{API_URL}/api/log/coverage/start",
+                            json=payload,
+                            headers=headers, timeout=5)
+        if res.status_code == 200:
+            cid = res.json().get("coverageId")
+            print(f"[Coverage] Started window #{cid} (source={source}) for stream {stream_id}")
+            if source == 'live':
+                coverage_id = cid
+            return cid
+    except Exception as e:
+        print(f"[Coverage] Error starting coverage: {e}")
+    return None
+
+def stop_coverage(cid=None, covered_to=None):
+    """Finalize a coverage window"""
+    global coverage_id
+    target_id = cid if cid is not None else coverage_id
+    if not target_id:
+        return
+    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
+    payload = {"coverageId": target_id}
+    if covered_to:
+        payload["coveredTo"] = covered_to
+    try:
+        requests.post(f"{API_URL}/api/log/coverage/end",
+                      json=payload,
+                      headers=headers, timeout=5)
+        print(f"[Coverage] Ended window #{target_id}")
+    except Exception as e:
+        print(f"[Coverage] Error ending coverage: {e}")
+    if cid is None or cid == coverage_id:
+        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
+
 def parse_irc_tags(tags_str):
     """Parse IRC v3 tags into a dictionary"""
     tags = {}
@@ -131,6 +191,7 @@ def twitch_irc_listener():
                         is_streamer = (user.lower() == TWITCH_CHANNEL)
                         is_mod = "moderator" in badges or "broadcaster" in badges
                         is_sub = "subscriber" in badges or "founder" in badges
+                        is_vip = "vip" in badges
                         
                         parsed_msg = {
                             "id": msg_id,
@@ -140,7 +201,8 @@ def twitch_irc_listener():
                             "timestamp": new_iso_timestamp(),
                             "isStreamer": is_streamer,
                             "isMod": is_mod,
-                            "isSub": is_sub
+                            "isSub": is_sub,
+                            "isVip": is_vip
                         }
                         
                         chat_queue.put(parsed_msg)
@@ -336,53 +398,29 @@ def run_backfill_loop():
                     title = target.get("title", "Unknown Archive")
                     start_time = target.get("start_time")
                     
-                    min_msg_time = target.get("min_message_time")
-                    max_msg_time = target.get("max_message_time")
-
-                    # Parse ISO strings to epoch timestamps
-                    def parse_time(t_str):
-                        if not t_str:
-                            return None
-                        try:
-                            clean_t = t_str.split(".")[0].replace("Z", "").replace("+00:00", "")
-                            return time.mktime(time.strptime(clean_t, "%Y-%m-%dT%H:%M:%S"))
-                        except Exception as parse_err:
-                            print(f"[Backfill] Error parsing time '{t_str}': {parse_err}")
-                            return None
-
-                    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))
+                    gaps = target.get("gaps", [])
+                    if not gaps:
+                        print(f"[Backfill] No gaps found for stream {stream_id}, skipping.")
+                        save_processed_stream(stream_id)
+                        continue
 
                     print(f"\n=========================================================")
-                    print(f"[Backfill] Processing Chat: {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] Processing Chat Gaps: {title}")
+                    print(f"[Backfill] VOD ID:                {vod_id}")
+                    print(f"[Backfill] Stream ID:            {stream_id}")
+                    print(f"[Backfill] Gaps count:            {len(gaps)}")
                     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)
+                    # 1. Download and upload chat segments for each gap
+                    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):
@@ -391,6 +429,30 @@ def run_backfill_loop():
                                     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}")
+
+                            # Extract and upload roles
+                            roles = [
+                                {
+                                    "username": m["username"],
+                                    "displayName": m.get("displayName", m["username"]),
+                                    "isMod": m.get("isMod", False),
+                                    "isSub": m.get("isSub", False),
+                                    "isVip": m.get("isVip", False),
+                                    "timestamp": m.get("timestamp")
+                                }
+                                      for m in chat_comments
+                                      if m.get("isMod") or m.get("isSub") or m.get("isVip")
+                            ]
+                            if roles:
+                                print(f"[Backfill] Uploading {len(roles)} roles...")
+                                for i in range(0, len(roles), 500):
+                                    batch = roles[i:i+500]
+                                    try:
+                                        requests.post(f"{API_URL}/api/log/roles", json={"roles": batch}, headers=headers_post, timeout=10)
+                                    except Exception as e:
+                                        print(f"[Backfill] Role batch upload error: {e}")
+                        else:
+                            print(f"[Backfill] No chat in this gap.")
                     
                     # 2. Mark locally as processed (does NOT mark completed on server, so worker.py can do voice)
                     save_processed_stream(stream_id)
@@ -424,16 +486,34 @@ if __name__ == "__main__":
     t_irc = threading.Thread(target=twitch_irc_listener, daemon=True)
     t_chat_send = threading.Thread(target=chat_sender, daemon=True)
     t_mod_send = threading.Thread(target=mod_action_sender, daemon=True)
+    t_heartbeat = threading.Thread(target=coverage_heartbeat_loop, daemon=True)
     
     t_irc.start()
     t_chat_send.start()
     t_mod_send.start()
+    t_heartbeat.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}")
     
     # Start VOD backfilling (GQL chat ONLY) in the main thread
     try:
         run_backfill_loop()
     except KeyboardInterrupt:
         print("\n[Shutting Down] Gracefully stopping threads...")
+    finally:
+        stop_coverage()
+        coverage_stop_flag.set()
         stop_flag.set()
         time.sleep(1)
         print("[Shutting Down] Done.")

local_worker/role_backfiller.py

@@ -33,15 +33,64 @@ def get_all_streams():
         print(f"[Error] Не удалось получить список стримов: {e}")
     return []
 
+def find_vod_id_by_stream_id(stream_id):
+    """Try to find a VOD ID for a given Twitch stream ID using GQL"""
+    url = "https://gql.twitch.tv/gql"
+    gql_headers = {
+        "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko",
+        "User-Agent": "Mozilla/5.0"
+    }
+    payload = {
+        "query": f"""
+        query {{
+            channel(name: "{os.getenv('TWITCH_CHANNEL', 'winx_prinx')}") {{
+                videos(first: 30, type: ARCHIVE) {{
+                    edges {{
+                        node {{
+                            id
+                            broadcastType
+                            lengthSeconds
+                            publishedAt
+                            stream {{
+                                id
+                            }}
+                        }}
+                    }}
+                }}
+            }}
+        }}
+        """
+    }
+    try:
+        res = requests.post(url, json=payload, headers=gql_headers, timeout=10)
+        if res.status_code == 200:
+            data = res.json()
+            edges = data.get("data", {}).get("channel", {}).get("videos", {}).get("edges", [])
+            for edge in edges:
+                node = edge.get("node", {})
+                stream_info = node.get("stream") or {}
+                if stream_info.get("id") == str(stream_id):
+                    return node.get("id")
+    except Exception as e:
+        pass
+    return None
+
 def extract_vod_id(stream):
-    """Extract VOD ID from stream's twitch_stream_id field (e.g. 'vod-2154382910' -> '2154382910')"""
-    twitch_id = stream.get("twitch_stream_id", "")
+    """Extract VOD ID from stream record"""
+    twitch_id = str(stream.get("twitch_stream_id", ""))
     if twitch_id.startswith("vod-"):
         return twitch_id[4:]
-    return None
+    # Live stream ID — try to find corresponding VOD
+    print(f"   -> Поиск VOD для live stream ID {twitch_id}...")
+    vod_id = find_vod_id_by_stream_id(twitch_id)
+    if vod_id:
+        print(f"   -> Найден VOD ID: {vod_id}")
+    else:
+        print(f"   -> VOD не найден (стрим мог быть удалён или VOD недоступен)")
+    return vod_id
 
 def download_vod_roles(vod_id):
-    """Download chat log of a past VOD and extract only roles"""
+    """Download chat log of a past VOD and extract only roles, with retry on failure"""
     roles_dict = {}
     url = "https://gql.twitch.tv/gql"
     gql_headers = {
@@ -51,6 +100,7 @@ def download_vod_roles(vod_id):
 
     current_offset = 0
     total_messages = 0
+    MAX_RETRIES = 5
 
     while True:
         payload = {
@@ -67,74 +117,96 @@ def download_vod_roles(vod_id):
             }
         }
 
-        try:
-            res = requests.post(url, json=payload, headers=gql_headers, timeout=10)
-            if res.status_code != 200:
-                break
-
-            data = res.json()
-            if isinstance(data, list):
-                data = data[0]
-
-            video = data.get("data", {}).get("video", {})
-            if not video:
-                break
-
-            edges = (video.get("comments") or {}).get("edges") or []
-            if not edges:
-                break
-
-            for edge in edges:
-                if not edge:
-                    continue
-                node = edge.get("node")
-                if not node:
-                    continue
-                commenter = node.get("commenter")
-                if not commenter:
-                    continue
-                user = commenter.get("login")
-                if not user:
-                    continue
-
-                display_name = commenter.get("displayName", user)
-                message = node.get("message")
-                if not message:
-                    continue
-
-                timestamp = node.get("createdAt")
-                user_badges = message.get("userBadges") or []
-                badges = [b.get("setID") for b in user_badges if b]
-                is_mod = "moderator" in badges or "broadcaster" in badges
-                is_sub = "subscriber" in badges or "founder" in badges
-                is_vip = "vip" in badges
-
-                roles_dict[user.lower()] = {
-                    "username": user,
-                    "displayName": display_name,
-                    "isMod": is_mod,
-                    "isSub": is_sub,
-                    "isVip": is_vip,
-                    "timestamp": timestamp
-                }
-                total_messages += 1
-
-            print(f"   -> Обработано {total_messages} сообщений, найдено {len(roles_dict)} уникальных пользователей...", end="\r")
-
-            last_offset = edges[-1].get("node", {}).get("contentOffsetSeconds")
-            if last_offset is not None:
-                next_offset = last_offset + 1
-                if next_offset <= current_offset:
-                    next_offset = current_offset + 30
-                current_offset = next_offset
-            else:
-                break
-        except Exception as e:
-            print(f"\n   [!] Ошибка при загрузке: {e}")
+        success = False
+        for attempt in range(MAX_RETRIES):
+            try:
+                res = requests.post(url, json=payload, headers=gql_headers, timeout=15)
+                if res.status_code != 200:
+                    break
+
+                data = res.json()
+                if isinstance(data, list):
+                    data = data[0]
+
+                video = data.get("data", {}).get("video", {})
+                if not video:
+                    return list(roles_dict.values())  # VOD ended or not found
+
+                edges = (video.get("comments") or {}).get("edges") or []
+                if not edges:
+                    return list(roles_dict.values())  # Reached end
+
+                for edge in edges:
+                    if not edge:
+                        continue
+                    node = edge.get("node")
+                    if not node:
+                        continue
+                    commenter = node.get("commenter")
+                    if not commenter:
+                        continue
+                    user = commenter.get("login")
+                    if not user:
+                        continue
+
+                    display_name = commenter.get("displayName", user)
+                    message = node.get("message")
+                    if not message:
+                        continue
+
+                    timestamp = node.get("createdAt")
+                    user_badges = message.get("userBadges") or []
+                    badges = [b.get("setID") for b in user_badges if b]
+
+                    # Debug: print first badge encounter to verify GQL structure
+                    if total_messages < 3 and user_badges:
+                        print(f"\n   [DEBUG] userBadges raw: {user_badges}")
+                        print(f"   [DEBUG] parsed badges: {badges}")
+
+                    is_mod = "moderator" in badges or "broadcaster" in badges
+                    is_sub = "subscriber" in badges or "founder" in badges
+                    is_vip = "vip" in badges
+
+                    # Only store users who have at least one badge
+                    if is_mod or is_sub or is_vip:
+                        roles_dict[user.lower()] = {
+                            "username": user,
+                            "displayName": display_name,
+                            "isMod": is_mod,
+                            "isSub": is_sub,
+                            "isVip": is_vip,
+                            "timestamp": timestamp
+                        }
+                    total_messages += 1
+
+                print(f"   -> Обработано {total_messages} сообщений, найдено {len(roles_dict)} уникальных пользователей...", end="\r")
+
+                last_offset = edges[-1].get("node", {}).get("contentOffsetSeconds")
+                if last_offset is not None:
+                    next_offset = last_offset + 1
+                    if next_offset <= current_offset:
+                        next_offset = current_offset + 30
+                    current_offset = next_offset
+                else:
+                    return list(roles_dict.values())
+
+                success = True
+                break  # Success — go to next page
+
+            except Exception as e:
+                wait = 2 ** attempt  # 1, 2, 4, 8, 16 seconds
+                print(f"\n   [!] Ошибка (попытка {attempt+1}/{MAX_RETRIES}): {e}")
+                if attempt < MAX_RETRIES - 1:
+                    print(f"   [!] Повтор через {wait}с...")
+                    time.sleep(wait)
+
+        if not success:
+            print(f"\n   [!] Не удалось получить данные после {MAX_RETRIES} попыток. Останавливаемся на offset={current_offset}.")
             break
 
     return list(roles_dict.values())
 
+
 def upload_roles(roles, vod_id):
     """Send roles to backend in batches"""
     batch_size = 500
@@ -169,18 +241,13 @@ if __name__ == "__main__":
     vod_streams = [s for s in streams if str(s.get("twitch_stream_id", "")).startswith("vod-")]
 
     print(f"[+] Найдено стримов всего: {len(streams)}")
-    print(f"[+] Из них архивных VOD: {len(vod_streams)}")
     print()
 
-    if not vod_streams:
-        print("[!] Нет VOD-стримов для обработки. Выход.")
-        sys.exit(0)
-
-    # Process each VOD
-    for idx, stream in enumerate(vod_streams, 1):
+    # Process each stream
+    for idx, stream in enumerate(streams, 1):
         vod_id = extract_vod_id(stream)
         title = stream.get("title", "Без названия")
-        print(f"[{idx}/{len(vod_streams)}] VOD {vod_id} — «{title}»")
+        print(f"[{idx}/{len(streams)}] Stream — «{title[:50]}»")
 
         if not vod_id:
             print("   -> Пропуск: нет VOD ID")

local_worker/worker.py

@@ -21,6 +21,7 @@ if added_paths:
     else:
         os.environ["LD_LIBRARY_PATH"] = ":".join(added_paths)
 
+from datetime import datetime, timedelta
 import time
 import socket
 import select
@@ -332,34 +333,48 @@ def notify_stream_end():
 coverage_id = None
 coverage_stop_flag = threading.Event()
 
-def start_coverage(stream_id, source='live'):
+def start_coverage(stream_id, source='live', covered_from=None, covered_to=None):
     """Register a new coverage window with the backend"""
     global coverage_id
     headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
+    payload = {"streamId": stream_id, "source": source}
+    if covered_from:
+        payload["coveredFrom"] = covered_from
+    if covered_to:
+        payload["coveredTo"] = covered_to
     try:
         res = requests.post(f"{API_URL}/api/log/coverage/start",
-                            json={"streamId": stream_id, "source": source},
+                            json=payload,
                             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}")
+            cid = res.json().get("coverageId")
+            print(f"[Coverage] Started window #{cid} (source={source}) for stream {stream_id}")
+            if source == 'live':
+                coverage_id = cid
+            return cid
     except Exception as e:
         print(f"[Coverage] Error starting coverage: {e}")
+    return None
 
-def stop_coverage():
-    """Finalize the current coverage window"""
+def stop_coverage(cid=None, covered_to=None):
+    """Finalize a coverage window"""
     global coverage_id
-    if not coverage_id:
+    target_id = cid if cid is not None else coverage_id
+    if not target_id:
         return
     headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
+    payload = {"coverageId": target_id}
+    if covered_to:
+        payload["coveredTo"] = covered_to
     try:
         requests.post(f"{API_URL}/api/log/coverage/end",
-                      json={"coverageId": coverage_id},
+                      json=payload,
                       headers=headers, timeout=5)
-        print(f"[Coverage] Ended window #{coverage_id}")
+        print(f"[Coverage] Ended window #{target_id}")
     except Exception as e:
         print(f"[Coverage] Error ending coverage: {e}")
-    coverage_id = None
+    if cid is None or cid == coverage_id:
+        coverage_id = None
 
 def coverage_heartbeat_loop():
     """Background thread: ping coverage heartbeat every 2 minutes"""
@@ -433,9 +448,19 @@ def run_stream_capture(model):
                                 return None
 
                         start_epoch = parse_time(start_time)
-                        
-                        # Notify start of coverage for this backfill session
-                        start_coverage(stream_id, source='backfill')
+
+                        # Helper to compute ISO timestamp for given offset from start_time
+                        def get_offset_iso_time(start_time_str, offset_seconds):
+                            if not start_time_str or offset_seconds is None:
+                                return None
+                            try:
+                                clean_str = start_time_str.replace('Z', '+00:00')
+                                dt = datetime.fromisoformat(clean_str)
+                                dt_offset = dt + timedelta(seconds=int(offset_seconds))
+                                return dt_offset.strftime("%Y-%m-%dT%H:%M:%SZ")
+                            except Exception as e:
+                                print(f"[Coverage] Error calculating offset time: {e}")
+                                return None
 
                         gaps = target.get("gaps", [])
                         if not gaps:
@@ -445,7 +470,6 @@ def run_stream_capture(model):
                                               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:")
@@ -460,7 +484,13 @@ def run_stream_capture(model):
                             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}")
 
+                            # 1. Register a coverage window for this gap (start)
+                            covered_from_iso = get_offset_iso_time(start_time, from_off)
+                            gap_coverage_id = start_coverage(stream_id, source='backfill', covered_from=covered_from_iso)
+
+                            # 2. Download and upload chat comments
                             chat_comments = download_vod_chat(vod_id, start_offset=from_off, end_offset=to_off)
+                            actual_to_offset = to_off
                             if chat_comments:
                                 print(f"[Backfill] Uploading {len(chat_comments)} messages...")
                                 batch_size = 100
@@ -473,24 +503,59 @@ def run_stream_capture(model):
                                                       headers=headers_post, timeout=10)
                                     except Exception as e:
                                         print(f"[Backfill] Chat batch upload error: {e}")
+
+                                # 2.5 Extract and upload roles
+                                roles = [
+                                    {
+                                        "username": m["username"],
+                                        "displayName": m.get("displayName", m["username"]),
+                                        "isMod": m.get("isMod", False),
+                                        "isSub": m.get("isSub", False),
+                                        "isVip": m.get("isVip", False),
+                                        "timestamp": m.get("timestamp")
+                                    }
+                                    for m in chat_comments
+                                    if m.get("isMod") or m.get("isSub") or m.get("isVip")
+                                ]
+                                if roles:
+                                    print(f"[Backfill] Uploading {len(roles)} roles...")
+                                    for i in range(0, len(roles), 500):
+                                        batch = roles[i:i+500]
+                                        try:
+                                            requests.post(f"{API_URL}/api/log/roles",
+                                                          json={"roles": batch},
+                                                          headers=headers_post, timeout=10)
+                                        except Exception as e:
+                                            print(f"[Backfill] Role batch upload error: {e}")
+
+                                # If to_off was None, we can use the offset of the last downloaded message
+                                if to_off is None:
+                                    try:
+                                        last_msg_time_str = chat_comments[-1]["timestamp"]
+                                        last_msg_epoch = parse_time(last_msg_time_str)
+                                        if last_msg_epoch and start_epoch:
+                                            actual_to_offset = int(last_msg_epoch - start_epoch)
+                                    except Exception as e:
+                                        print(f"[Backfill] Error estimating end offset: {e}")
                             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}")
+                            # 3. Transcribe audio for this gap
+                            print(f"[Backfill] Starting Whisper audio transcription for this gap...")
                             try:
                                 start_time_iso = start_time.replace("+00:00", "").replace("Z", "") + "Z"
                                 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 and end coverage
+
+                            # 4. Register a coverage window for this gap (end)
+                            covered_to_iso = get_offset_iso_time(start_time, actual_to_offset)
+                            if not covered_to_iso:
+                                covered_to_iso = get_offset_iso_time(start_time, from_off + 60)
+                            stop_coverage(gap_coverage_id, covered_to=covered_to_iso)
+
+                        # 3. Mark completed
                         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)
@@ -500,8 +565,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
                     else:

server/db.js

@@ -1580,17 +1580,17 @@ export async function getModeratorProfilesData(streamId = null) {
 /**
  * Start a new coverage window (worker session started)
  */
-export async function startCoverageWindow(streamId, source = 'live') {
-  const now = new Date().toISOString();
+export async function startCoverageWindow(streamId, source = 'live', coveredFrom = null, coveredTo = null) {
+  const fromTime = coveredFrom || new Date().toISOString();
   if (dbMode === 'sqlite') {
     const result = sqliteDb.prepare(
-      'INSERT INTO stream_coverage (stream_id, covered_from, source) VALUES (?, ?, ?)'
-    ).run(streamId, now, source);
+      'INSERT INTO stream_coverage (stream_id, covered_from, covered_to, source) VALUES (?, ?, ?, ?)'
+    ).run(streamId, fromTime, coveredTo, source);
     return result.lastInsertRowid;
   } else {
     const { data, error } = await supabase
       .from('stream_coverage')
-      .insert({ stream_id: streamId, covered_from: now, source })
+      .insert({ stream_id: streamId, covered_from: fromTime, covered_to: coveredTo, source })
       .select('id')
       .single();
     if (error) console.error('[Supabase] startCoverageWindow error:', error.message);
@@ -1613,12 +1613,12 @@ export async function updateCoverageHeartbeat(coverageId) {
 /**
  * End a coverage window (worker session stopped)
  */
-export async function endCoverageWindow(coverageId) {
-  const now = new Date().toISOString();
+export async function endCoverageWindow(coverageId, coveredTo = null) {
+  const toTime = coveredTo || new Date().toISOString();
   if (dbMode === 'sqlite') {
-    sqliteDb.prepare('UPDATE stream_coverage SET covered_to = ? WHERE id = ?').run(now, coverageId);
+    sqliteDb.prepare('UPDATE stream_coverage SET covered_to = ? WHERE id = ?').run(toTime, coverageId);
   } else {
-    await supabase.from('stream_coverage').update({ covered_to: now }).eq('id', coverageId);
+    await supabase.from('stream_coverage').update({ covered_to: toTime }).eq('id', coverageId);
   }
 }
 

server/server.js

@@ -282,10 +282,10 @@ 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;
+  const { streamId, source, coveredFrom, coveredTo } = req.body;
   if (!streamId) return res.status(400).json({ error: 'Missing streamId' });
   try {
-    const coverageId = await startCoverageWindow(streamId, source || 'live');
+    const coverageId = await startCoverageWindow(streamId, source || 'live', coveredFrom, coveredTo);
     res.json({ success: true, coverageId });
   } catch (err) {
     console.error('[API Error] /api/log/coverage/start:', err);
@@ -307,10 +307,10 @@ app.patch('/api/log/coverage/:id/heartbeat', authenticateWorker, async (req, res
 
 // POST end coverage window
 app.post('/api/log/coverage/end', authenticateWorker, async (req, res) => {
-  const { coverageId } = req.body;
+  const { coverageId, coveredTo } = req.body;
   if (!coverageId) return res.status(400).json({ error: 'Missing coverageId' });
   try {
-    await endCoverageWindow(coverageId);
+    await endCoverageWindow(coverageId, coveredTo);
     res.json({ success: true });
   } catch (err) {
     console.error('[API Error] /api/log/coverage/end:', err);