Harden all edge cases: auto-reconnect with health check, thread-safe connection pool, scoped artist CTEs (137K→30 row scan), ttl_cache on stats/filters, invalidate_cache on mutations, safe is_watched cast, remove test files

.gitignore

@@ -48,3 +48,5 @@ frontend/.swc/
 frontend/public/logo.png
 frontend/src/app/favicon.ico
 db/.fuse_hidden*
+*.pyc
+test_*.py

api/main.py

@@ -18,6 +18,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from db import get_connection, init_db
 from metrics import compute_artist_metrics, enrich_song_with_metrics, enrich_viral_alert, enrich_songs_bulk, enrich_viral_alerts_bulk
+from cache import ttl_cache, invalidate_cache
 
 app = FastAPI(title="Music Intelligence Dashboard API", version="1.0.0")
 
@@ -1029,6 +1030,10 @@ def get_spotify_artists(
 @app.get("/api/filters")
 def get_filter_options():
     """Get available genre and region options for filter dropdowns."""
+    return _get_filters_cached()
+
+@ttl_cache(ttl_seconds=300)
+def _get_filters_cached():
     conn = get_connection()
     genres = [r[0] for r in conn.execute(
         "SELECT DISTINCT genre FROM artists WHERE genre IS NOT NULL ORDER BY genre"
@@ -1052,17 +1057,22 @@ def get_artist_detail(artist_id: str, platform: str = Query("youtube")):
         conn.close()
         raise HTTPException(status_code=404, detail="Artist not found")
 
-    # Get songs for this platform
+    # Get songs for this platform — CTEs scoped to this artist's songs only
     songs = conn.execute("""
-        WITH latest_snaps AS (
-            SELECT song_id, MAX(id) as max_id
-            FROM play_snapshots
-            GROUP BY song_id
+        WITH artist_song_ids AS (
+            SELECT id FROM songs WHERE artist_id = ? AND platform = ?
+        ),
+        latest_snaps AS (
+            SELECT ps.song_id, MAX(ps.id) as max_id
+            FROM play_snapshots ps
+            WHERE ps.song_id IN (SELECT id FROM artist_song_ids)
+            GROUP BY ps.song_id
         ),
         first_snaps AS (
-            SELECT song_id, MIN(id) as min_id
-            FROM play_snapshots
-            GROUP BY song_id
+            SELECT ps.song_id, MIN(ps.id) as min_id
+            FROM play_snapshots ps
+            WHERE ps.song_id IN (SELECT id FROM artist_song_ids)
+            GROUP BY ps.song_id
         )
         SELECT
             s.*,
@@ -1081,7 +1091,7 @@ def get_artist_detail(artist_id: str, platform: str = Query("youtube")):
         LEFT JOIN play_snapshots fs_data ON fs.min_id = fs_data.id
         WHERE s.artist_id = ? AND s.platform = ?
         ORDER BY latest_play_count DESC NULLS LAST
-    """, (artist_id, platform)).fetchall()
+    """, (artist_id, platform, artist_id, platform)).fetchall()
 
     # Get viral alerts for this artist
     alerts = conn.execute("""
@@ -1123,17 +1133,20 @@ def get_spotify_artist_detail(artist_id: str):
         raise HTTPException(status_code=404, detail="Spotify artist not found")
 
     songs = conn.execute("""
-        WITH latest_snaps AS (
-            SELECT song_id, MAX(id) as max_id
-            FROM play_snapshots
-            WHERE platform = 'spotify'
-            GROUP BY song_id
+        WITH artist_song_ids AS (
+            SELECT id FROM songs WHERE artist_id = ? AND platform = 'spotify'
+        ),
+        latest_snaps AS (
+            SELECT ps.song_id, MAX(ps.id) as max_id
+            FROM play_snapshots ps
+            WHERE ps.song_id IN (SELECT id FROM artist_song_ids) AND ps.platform = 'spotify'
+            GROUP BY ps.song_id
         ),
         first_snaps AS (
-            SELECT song_id, MIN(id) as min_id
-            FROM play_snapshots
-            WHERE platform = 'spotify'
-            GROUP BY song_id
+            SELECT ps.song_id, MIN(ps.id) as min_id
+            FROM play_snapshots ps
+            WHERE ps.song_id IN (SELECT id FROM artist_song_ids) AND ps.platform = 'spotify'
+            GROUP BY ps.song_id
         )
         SELECT
             s.*,
@@ -1147,7 +1160,7 @@ def get_spotify_artist_detail(artist_id: str):
         LEFT JOIN play_snapshots fs_data ON fs.min_id = fs_data.id
         WHERE s.artist_id = ? AND s.platform = 'spotify'
         ORDER BY latest_play_count DESC NULLS LAST
-    """, (artist_id,)).fetchall()
+    """, (artist_id, artist_id)).fetchall()
 
     alerts = conn.execute("""
         SELECT
@@ -1197,10 +1210,11 @@ def toggle_watchlist(artist_id: str):
         conn.close()
         raise HTTPException(status_code=404, detail="Artist not found")
 
-    new_status = 0 if artist["is_watched"] else 1
+    new_status = 0 if int(artist["is_watched"] or 0) else 1
     conn.execute("UPDATE artists SET is_watched = ? WHERE id = ?", (new_status, artist_id))
     conn.commit()
     conn.close()
+    invalidate_cache()
     return {"artist_id": artist_id, "is_watched": bool(new_status)}
 
 
@@ -1236,6 +1250,7 @@ def delete_artist(artist_id: str):
     conn.execute("DELETE FROM artists WHERE id = ?", (artist_id,))
     conn.commit()
     conn.close()
+    invalidate_cache()
 
     return {
         "status": "deleted",
@@ -1249,6 +1264,10 @@ def delete_artist(artist_id: str):
 @app.get("/api/stats")
 def get_stats():
     """Get overall dashboard statistics."""
+    return _get_stats_cached()
+
+@ttl_cache(ttl_seconds=60)
+def _get_stats_cached():
     conn = get_connection()
     counts = conn.execute("""
         SELECT

collector/db.py

@@ -2,6 +2,8 @@
 
 import sqlite3
 import os
+import time
+import threading
 from datetime import datetime, timezone, timedelta
 from pathlib import Path
 from dotenv import load_dotenv
@@ -64,29 +66,50 @@ class LibsqlDictConnection:
     def executescript(self, sql):
         self.conn.executescript(sql)
 
+    def executemany(self, sql, params_list):
+        self.conn.executemany(sql, params_list)
+
     def commit(self):
         self.conn.commit()
 
     def close(self):
-        # Do not close the connection so it can be pooled globally across requests
+        # No-op for pooled Turso connection — connection is reused globally
         pass
 
 _TURSO_GLOBAL_CONN = None
+_TURSO_LAST_OK = 0  # timestamp of last successful use
+_TURSO_LOCK = threading.Lock()
 
 def get_connection():
-    """Get a SQLite connection with WAL mode for better concurrency."""
-    global _TURSO_GLOBAL_CONN
+    """Get a SQLite connection with WAL mode for better concurrency.
+    
+    For Turso: uses a global pooled connection with auto-reconnect.
+    Health-checks the connection if idle for >30 seconds.
+    """
+    global _TURSO_GLOBAL_CONN, _TURSO_LAST_OK
     if _USE_TURSO:
         if libsql is None:
             raise ImportError(
                 "libsql_experimental is required for Turso connections. "
                 "Install it with: pip install libsql-experimental"
             )
-        if _TURSO_GLOBAL_CONN is None:
-            safe_url = _TURSO_URL.replace("libsql://", "https://").replace("wss://", "https://")
-            conn = libsql.connect(safe_url, auth_token=_TURSO_TOKEN)
-            _TURSO_GLOBAL_CONN = LibsqlDictConnection(conn)
-        return _TURSO_GLOBAL_CONN
+        with _TURSO_LOCK:
+            now = time.time()
+            # Health check only if connection has been idle for >30 seconds
+            if _TURSO_GLOBAL_CONN is not None and (now - _TURSO_LAST_OK) > 30:
+                try:
+                    _TURSO_GLOBAL_CONN.execute("SELECT 1").fetchone()
+                except Exception:
+                    print("[db] Turso connection stale, reconnecting...")
+                    _TURSO_GLOBAL_CONN = None
+            # Create new connection if needed
+            if _TURSO_GLOBAL_CONN is None:
+                safe_url = _TURSO_URL.replace("libsql://", "https://").replace("wss://", "https://")
+                conn = libsql.connect(safe_url, auth_token=_TURSO_TOKEN)
+                _TURSO_GLOBAL_CONN = LibsqlDictConnection(conn)
+                print("[db] Turso connection established.")
+            _TURSO_LAST_OK = now
+            return _TURSO_GLOBAL_CONN
     else:
         # Local SQLite fallback for development
         conn = sqlite3.connect(str(DB_PATH))
@@ -124,9 +147,13 @@ def init_db():
     else:
         print(f"Connected to Turso at {_TURSO_URL}")
 
+_INDEXES_CREATED = False
 
 def ensure_indexes():
-    """Create performance indexes if they don't exist."""
+    """Create performance indexes if they don't exist. Runs once per process."""
+    global _INDEXES_CREATED
+    if _INDEXES_CREATED:
+        return
     conn = get_connection()
     indexes = [
         "CREATE INDEX IF NOT EXISTS idx_snapshots_song_platform_date ON play_snapshots(song_id, platform, collected_at DESC)",
@@ -142,9 +169,10 @@ def ensure_indexes():
             conn.commit()
         except Exception as e:
             print(f"[indexes] Skipping: {e}")
+    _INDEXES_CREATED = True
     print("[indexes] Performance indexes ensured.")
 
-# Auto-create indexes on import
+# Auto-create indexes on first import (runs once)
 try:
     ensure_indexes()
 except Exception:

test_db.py

@@ -1,10 +0,0 @@
-import os, sys
-sys.path.insert(0, 'collector')
-from dotenv import load_dotenv
-import db
-
-load_dotenv()
-conn = db.get_connection()
-res = conn.execute("SELECT release_date, title, platform FROM songs ORDER BY release_date DESC LIMIT 5").fetchall()
-for r in res:
-    print(r)

test_perf.py

@@ -1,53 +0,0 @@
-import os, sys, time
-sys.path.insert(0, 'collector')
-from dotenv import load_dotenv
-import db
-
-load_dotenv()
-conn = db.get_connection()
-
-q1 = """
-SELECT
-    a.id,
-    (SELECT COUNT(*) FROM songs WHERE artist_id = a.id AND platform = 'youtube') as yt_song_count,
-    (SELECT SUM(ps.play_count) FROM play_snapshots ps
-     JOIN songs s ON ps.song_id = s.id
-     WHERE s.artist_id = a.id AND ps.platform = 'youtube'
-     AND ps.id IN (SELECT MAX(id) FROM play_snapshots GROUP BY song_id)
-    ) as total_yt_views
-FROM artists a
-LIMIT 5
-"""
-
-q2 = """
-WITH latest_snapshots AS (
-    SELECT song_id, MAX(id) as max_id
-    FROM play_snapshots
-    GROUP BY song_id
-),
-song_stats AS (
-    SELECT 
-        s.artist_id,
-        COUNT(CASE WHEN s.platform = 'youtube' THEN 1 END) as yt_song_count,
-        SUM(CASE WHEN s.platform = 'youtube' THEN ps.play_count ELSE 0 END) as total_yt_views
-    FROM songs s
-    LEFT JOIN latest_snapshots ls ON s.id = ls.song_id
-    LEFT JOIN play_snapshots ps ON ls.max_id = ps.id
-    GROUP BY s.artist_id
-)
-SELECT 
-    a.id,
-    COALESCE(ss.yt_song_count, 0) as yt_song_count,
-    COALESCE(ss.total_yt_views, 0) as total_yt_views
-FROM artists a
-LEFT JOIN song_stats ss ON a.id = ss.artist_id
-LIMIT 5
-"""
-
-start = time.time()
-conn.execute(q1).fetchall()
-print("Q1 (Old):", time.time() - start)
-
-start = time.time()
-conn.execute(q2).fetchall()
-print("Q2 (New):", time.time() - start)

test_perf_artist.py

@@ -1,65 +0,0 @@
-import os, sys, time
-sys.path.insert(0, 'collector')
-from dotenv import load_dotenv
-import db
-
-load_dotenv()
-conn = db.get_connection()
-
-artist_id = "arooh"
-platform = "youtube"
-
-q1 = """
-        SELECT
-            s.*,
-            (SELECT play_count FROM play_snapshots
-             WHERE song_id = s.id ORDER BY collected_at DESC LIMIT 1) as latest_play_count,
-            (SELECT play_count FROM play_snapshots
-             WHERE song_id = s.id ORDER BY collected_at ASC LIMIT 1) as first_play_count,
-            (SELECT like_count FROM play_snapshots
-             WHERE song_id = s.id ORDER BY collected_at DESC LIMIT 1) as latest_like_count,
-            (SELECT comment_count FROM play_snapshots
-             WHERE song_id = s.id ORDER BY collected_at DESC LIMIT 1) as latest_comment_count,
-            (SELECT ytmusic_play_count FROM play_snapshots
-             WHERE song_id = s.id AND ytmusic_play_count IS NOT NULL ORDER BY collected_at DESC LIMIT 1) as ytmusic_play_count
-        FROM songs s
-        WHERE s.artist_id = ? AND s.platform = ?
-        ORDER BY latest_play_count DESC NULLS LAST
-"""
-
-q2 = """
-        WITH latest_snaps AS (
-            SELECT song_id, MAX(id) as max_id
-            FROM play_snapshots
-            WHERE platform = ?
-            GROUP BY song_id
-        ),
-        first_snaps AS (
-            SELECT song_id, MIN(id) as min_id
-            FROM play_snapshots
-            WHERE platform = ?
-            GROUP BY song_id
-        )
-        SELECT
-            s.*,
-            ls_data.play_count as latest_play_count,
-            fs_data.play_count as first_play_count,
-            ls_data.like_count as latest_like_count,
-            ls_data.comment_count as latest_comment_count,
-            ls_data.ytmusic_play_count as ytmusic_play_count
-        FROM songs s
-        LEFT JOIN latest_snaps ls ON s.id = ls.song_id
-        LEFT JOIN play_snapshots ls_data ON ls.max_id = ls_data.id
-        LEFT JOIN first_snaps fs ON s.id = fs.song_id
-        LEFT JOIN play_snapshots fs_data ON fs.min_id = fs_data.id
-        WHERE s.artist_id = ? AND s.platform = ?
-        ORDER BY latest_play_count DESC NULLS LAST
-"""
-
-start = time.time()
-conn.execute(q1, (artist_id, platform)).fetchall()
-print("Q1 (Old):", time.time() - start)
-
-start = time.time()
-conn.execute(q2, (platform, platform, artist_id, platform)).fetchall()
-print("Q2 (New):", time.time() - start)