diff --git a/api/__init__.py b/api/__init__.py index 35faf2be..cf6b8bd1 100644 --- a/api/__init__.py +++ b/api/__init__.py @@ -43,6 +43,7 @@ def create_api_blueprint(): from .listenbrainz import register_routes as reg_listenbrainz from .cache import register_routes as reg_cache from .request import register_routes as reg_request + from .request import start_cleanup_thread as _start_request_cleanup # ---- rate-limit only /api/v1 routes (not the whole app) ---- limiter.limit("60 per minute")(bp) @@ -62,6 +63,11 @@ def create_api_blueprint(): reg_cache(bp) reg_request(bp) + # Start the periodic cleanup timer for in-memory request tracking so + # idle periods don't leave stale entries in memory. Idempotent across + # calls; safe with multi-blueprint registration. + _start_request_cleanup() + # ---- error handlers (scoped to this Blueprint) ---- @bp.errorhandler(400) def _bad_request(e): diff --git a/api/auth.py b/api/auth.py index 3ddecc42..c6f0fb33 100644 --- a/api/auth.py +++ b/api/auth.py @@ -4,8 +4,9 @@ API key authentication for the SoulSync public API. import hashlib import secrets +import threading import uuid -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from functools import wraps from flask import request, current_app @@ -13,6 +14,27 @@ from flask import request, current_app from .helpers import api_error +# Throttle persistence of `last_used_at` so every authenticated request +# does not rewrite the full app config. Maps key_hash -> last-persisted datetime. +_USAGE_WRITE_INTERVAL = timedelta(minutes=15) +_last_persisted_usage: dict[str, datetime] = {} +_usage_lock = threading.Lock() + + +def _should_persist_usage(key_hash: str, now: datetime) -> bool: + """Return True if `last_used_at` for the given key should be written to disk. + + Thread-safe: tracks the last write per key hash in memory and only returns + True once per `_USAGE_WRITE_INTERVAL`. + """ + with _usage_lock: + previous = _last_persisted_usage.get(key_hash) + if previous is None or (now - previous) >= _USAGE_WRITE_INTERVAL: + _last_persisted_usage[key_hash] = now + return True + return False + + def generate_api_key(label=""): """Generate a new API key. @@ -67,9 +89,12 @@ def require_api_key(f): if not matched: return api_error("INVALID_KEY", "Invalid API key.", 403) - # Update last-used timestamp (best-effort) - matched["last_used_at"] = datetime.now(timezone.utc).isoformat() - config_mgr.set("api_keys", stored_keys) + # Update last-used timestamp (best-effort, throttled to avoid rewriting + # the full app config on every authenticated request). + now = datetime.now(timezone.utc) + matched["last_used_at"] = now.isoformat() + if _should_persist_usage(key_hash, now): + config_mgr.set("api_keys", stored_keys) return f(*args, **kwargs) diff --git a/api/downloads.py b/api/downloads.py index fd64ed7e..448b80cd 100644 --- a/api/downloads.py +++ b/api/downloads.py @@ -40,16 +40,68 @@ def register_routes(bp): @bp.route("/downloads", methods=["GET"]) @require_api_key def list_downloads(): - """List active and recent download tasks.""" + """List download tasks with optional filtering and pagination. + + Query params: + status: comma-separated statuses to include (e.g. "downloading,queued"). + Default includes all. + limit: max tasks to return (default 100, max 500). + offset: skip the first N tasks (default 0). + + Response includes `total` (post-filter count) so clients can paginate + without fetching everything. Tasks are sorted by `status_change_time` + descending so newest/in-flight tasks appear first. + """ try: from web_server import download_tasks, tasks_lock - tasks = [] - with tasks_lock: - for task_id, task in download_tasks.items(): - tasks.append(_serialize_download(task_id, task)) + # Parse pagination params + try: + limit = int(request.args.get("limit", 100)) + except (TypeError, ValueError): + limit = 100 + try: + offset = int(request.args.get("offset", 0)) + except (TypeError, ValueError): + offset = 0 + # Clamp to sensible bounds + limit = max(1, min(limit, 500)) + offset = max(0, offset) - return api_success({"downloads": tasks}) + status_param = request.args.get("status", "").strip() + status_filter = ( + {s.strip() for s in status_param.split(",") if s.strip()} + if status_param + else None + ) + + # Snapshot under the lock, then sort/slice outside. + with tasks_lock: + snapshot = list(download_tasks.items()) + + if status_filter: + snapshot = [ + (tid, t) for tid, t in snapshot + if (t.get("status") or "") in status_filter + ] + + # Sort newest-first by status_change_time; fall back to string id + # so ordering is stable when timestamps are missing or tied. + snapshot.sort( + key=lambda item: (item[1].get("status_change_time") or "", item[0]), + reverse=True, + ) + + total = len(snapshot) + page = snapshot[offset:offset + limit] + tasks = [_serialize_download(tid, t) for tid, t in page] + + return api_success({ + "downloads": tasks, + "total": total, + "limit": limit, + "offset": offset, + }) except ImportError: return api_error("NOT_AVAILABLE", "Download tracking not available.", 501) except Exception as e: diff --git a/api/library.py b/api/library.py index b5f363c5..5352f971 100644 --- a/api/library.py +++ b/api/library.py @@ -186,15 +186,8 @@ def register_routes(bp): try: db = get_database() - tracks = db.search_tracks(title=title, artist=artist, limit=limit) - if not tracks: - return api_success({"tracks": []}) - - # Re-query by IDs to get full row data - track_ids = [t.id for t in tracks] - full_tracks = db.api_get_tracks_by_ids(track_ids) - - return api_success({"tracks": [serialize_track(t, fields) for t in full_tracks]}) + tracks = db.api_search_tracks(title=title, artist=artist, limit=limit) + return api_success({"tracks": [serialize_track(t, fields) for t in tracks]}) except Exception as e: return api_error("LIBRARY_ERROR", str(e), 500) diff --git a/api/request.py b/api/request.py index b782cf98..820c6fc3 100644 --- a/api/request.py +++ b/api/request.py @@ -23,6 +23,16 @@ _requests_lock = threading.Lock() # Max age before auto-cleanup _MAX_REQUEST_AGE = timedelta(hours=1) +# How often the background cleanup timer runs. Short enough to keep memory +# bounded during idle periods, long enough that slow-polling external clients +# still see their request for close to the TTL. +_CLEANUP_INTERVAL_SECONDS = 300 # 5 minutes + +# Guards for the singleton background cleanup thread. +_cleanup_thread: "threading.Thread | None" = None +_cleanup_stop_event = threading.Event() +_cleanup_thread_lock = threading.Lock() + def _cleanup_old_requests(): """Remove requests older than 1 hour to prevent unbounded growth.""" @@ -32,6 +42,57 @@ def _cleanup_old_requests(): if r.get('created_at', datetime.now()) < cutoff] for rid in expired: del _pending_requests[rid] + return len(expired) if expired else 0 + + +def _cleanup_loop(): + """Background thread: periodically evict expired requests.""" + while not _cleanup_stop_event.is_set(): + # wait() returns True if the event was set (shutdown), False on timeout + if _cleanup_stop_event.wait(timeout=_CLEANUP_INTERVAL_SECONDS): + return + try: + removed = _cleanup_old_requests() + if removed: + logger.debug(f"Request cleanup: evicted {removed} stale entries") + except Exception as e: + logger.warning(f"Request cleanup loop error: {e}") + + +def start_cleanup_thread() -> bool: + """Start the background cleanup timer once per process. + + Returns True if a new thread was started, False if one was already + running. Safe to call multiple times; callers in multi-worker setups + should still gate on worker identity if they want exactly one thread + across the entire deployment. + """ + global _cleanup_thread + with _cleanup_thread_lock: + if _cleanup_thread is not None and _cleanup_thread.is_alive(): + return False + _cleanup_stop_event.clear() + _cleanup_thread = threading.Thread( + target=_cleanup_loop, + name="api-request-cleanup", + daemon=True, + ) + _cleanup_thread.start() + logger.info("Started api/request cleanup timer (interval=%ss)" % _CLEANUP_INTERVAL_SECONDS) + return True + + +def stop_cleanup_thread(timeout: float = 2.0) -> None: + """Signal the cleanup thread to exit. Used in tests and shutdown paths.""" + global _cleanup_thread + with _cleanup_thread_lock: + thread = _cleanup_thread + _cleanup_stop_event.set() + if thread is not None and thread.is_alive(): + thread.join(timeout=timeout) + with _cleanup_thread_lock: + _cleanup_thread = None + _cleanup_stop_event.clear() def _run_search_and_download(request_id, query, notify_url): diff --git a/api/wishlist.py b/api/wishlist.py index bff40e63..7482ed47 100644 --- a/api/wishlist.py +++ b/api/wishlist.py @@ -19,21 +19,19 @@ def register_routes(bp): fields = parse_fields(request) profile_id = parse_profile_id(request) + category_filter = category if category in ("singles", "albums") else None + try: from database.music_database import get_database db = get_database() - raw_tracks = db.get_wishlist_tracks(profile_id=profile_id) - - # Category filter - if category in ("singles", "albums"): - raw_tracks = [ - t for t in raw_tracks - if _track_category(t) == category - ] - - total = len(raw_tracks) - start = (page - 1) * limit - tracks = raw_tracks[start:start + limit] + offset = (page - 1) * limit + tracks = db.get_wishlist_tracks( + profile_id=profile_id, + category=category_filter, + limit=limit, + offset=offset, + ) + total = db.get_wishlist_count(profile_id=profile_id, category=category_filter) return api_success( {"tracks": [serialize_wishlist_track(t, fields) for t in tracks]}, @@ -104,15 +102,3 @@ def register_routes(bp): return api_error("NOT_AVAILABLE", "Wishlist processing function not available.", 501) except Exception as e: return api_error("WISHLIST_ERROR", str(e), 500) - - -def _track_category(track): - """Determine if a wishlist track is a single or album track.""" - album_type = "" - if isinstance(track, dict): - sd = track.get("spotify_data", {}) - if isinstance(sd, dict): - album = sd.get("album", {}) - if isinstance(album, dict): - album_type = album.get("album_type", "") - return "albums" if album_type == "album" else "singles" diff --git a/core/lastfm_worker.py b/core/lastfm_worker.py index f876c380..2509f4aa 100644 --- a/core/lastfm_worker.py +++ b/core/lastfm_worker.py @@ -307,7 +307,12 @@ class LastFMWorker: logger.error(f"Error updating item status: {e2}") def _get_existing_id(self, entity_type: str, entity_id: int) -> Optional[str]: - """Check if an entity already has a lastfm_id (e.g. from manual match).""" + """Check if an entity already has a lastfm_url (e.g. from manual match). + + The Last.fm schema uses `lastfm_url` (not `lastfm_id`) on artists, albums, + and tracks. This helper always returned None before because it queried a + non-existent column and silently caught the exception. + """ table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'} table = table_map.get(entity_type) if not table: @@ -316,7 +321,7 @@ class LastFMWorker: try: conn = self.db._get_connection() cursor = conn.cursor() - cursor.execute(f"SELECT lastfm_id FROM {table} WHERE id = ?", (entity_id,)) + cursor.execute(f"SELECT lastfm_url FROM {table} WHERE id = ?", (entity_id,)) row = cursor.fetchone() return row[0] if row and row[0] else None except Exception: @@ -329,7 +334,10 @@ class LastFMWorker: """Process an artist: get full info from Last.fm""" existing_id = self._get_existing_id('artist', artist_id) if existing_id: - logger.debug(f"Preserving existing Last.fm ID for artist '{artist_name}': {existing_id}") + # Row already has a Last.fm URL but status is NULL (legacy/manual match). + # Mark as matched so the worker does not re-select it forever. + logger.debug(f"Preserving existing Last.fm URL for artist '{artist_name}': {existing_id}") + self._mark_status('artist', artist_id, 'matched') return # Use get_artist_info for detailed data (includes stats, bio, tags, similar) @@ -353,7 +361,8 @@ class LastFMWorker: """Process an album: get full info from Last.fm""" existing_id = self._get_existing_id('album', album_id) if existing_id: - logger.debug(f"Preserving existing Last.fm ID for album '{album_name}': {existing_id}") + logger.debug(f"Preserving existing Last.fm URL for album '{album_name}': {existing_id}") + self._mark_status('album', album_id, 'matched') return result = self.client.get_album_info(artist_name, album_name) @@ -376,7 +385,8 @@ class LastFMWorker: """Process a track: get full info from Last.fm""" existing_id = self._get_existing_id('track', track_id) if existing_id: - logger.debug(f"Preserving existing Last.fm ID for track '{track_name}': {existing_id}") + logger.debug(f"Preserving existing Last.fm URL for track '{track_name}': {existing_id}") + self._mark_status('track', track_id, 'matched') return result = self.client.get_track_info(artist_name, track_name) diff --git a/core/listening_stats_worker.py b/core/listening_stats_worker.py index 414f857a..72bdaf16 100644 --- a/core/listening_stats_worker.py +++ b/core/listening_stats_worker.py @@ -180,12 +180,18 @@ class ListeningStatsWorker: 'played_at': entry.get('played_at'), 'duration_ms': entry.get('duration_ms', 0), 'server_source': active_server, - 'db_track_id': self._resolve_db_track_id( - entry.get('track_title', ''), - entry.get('artist', '') - ), + # db_track_id filled in below by a single batched lookup + 'db_track_id': None, }) + # Batch-resolve track IDs for all events at once (was N+1 before). + id_map = self._resolve_db_track_ids_batch(events) + for ev in events: + title_l = (ev.get('title') or '').strip().lower() + artist_l = (ev.get('artist') or '').strip().lower() + if title_l: + ev['db_track_id'] = id_map.get((title_l, artist_l)) + inserted = self.db.insert_listening_events(events) self.stats['events_added'] += inserted logger.info(f"Inserted {inserted} new listening events (of {len(events)} total)") @@ -269,59 +275,129 @@ class ListeningStatsWorker: logger.error(f"Failed to build stats cache: {e}") def _enrich_stats_items(self, cache): - """Add image URLs, IDs, and Last.fm data to cached stats items.""" + """Add image URLs, IDs, and Last.fm data to cached stats items. + + Previously ran one SELECT per artist / album / track entry. Now each + of the three lists is resolved with a single batched IN query so + cache rebuilds scale with the number of result sets, not with the + number of items in them. + """ + top_artists = cache.get('top_artists') or [] + top_albums = cache.get('top_albums') or [] + top_tracks = cache.get('top_tracks') or [] + + if not (top_artists or top_albums or top_tracks): + return + conn = None try: conn = self.db._get_connection() cursor = conn.cursor() - for artist in (cache.get('top_artists') or []): - try: - cursor.execute(""" - SELECT thumb_url, id, lastfm_listeners, lastfm_playcount, soul_id - FROM artists WHERE LOWER(name) = LOWER(?) LIMIT 1 - """, (artist['name'],)) - r = cursor.fetchone() - if r: - artist['image_url'] = r[0] or None - artist['id'] = r[1] - artist['global_listeners'] = r[2] - artist['global_playcount'] = r[3] - artist['soul_id'] = r[4] - except Exception: - pass + # ---- top_artists: match by LOWER(name) ---- + if top_artists: + names = [a.get('name') or '' for a in top_artists] + unique_names = {n.lower() for n in names if n} + artist_rows = {} + if unique_names: + name_list = list(unique_names) + chunk = 500 + for i in range(0, len(name_list), chunk): + sub = name_list[i:i + chunk] + placeholders = ','.join(['?'] * len(sub)) + cursor.execute( + f""" + SELECT LOWER(name), thumb_url, id, lastfm_listeners, + lastfm_playcount, soul_id + FROM artists + WHERE LOWER(name) IN ({placeholders}) + """, + sub, + ) + for row in cursor.fetchall(): + # Keep first match per lowered name (LIMIT 1 equiv). + artist_rows.setdefault(row[0], row) - for album in (cache.get('top_albums') or []): - try: - cursor.execute(""" - SELECT al.thumb_url, al.id, al.artist_id FROM albums al - WHERE LOWER(al.title) = LOWER(?) LIMIT 1 - """, (album['name'],)) - r = cursor.fetchone() + for artist in top_artists: + key = (artist.get('name') or '').lower() + r = artist_rows.get(key) if r: - album['image_url'] = r[0] or None - album['id'] = r[1] - album['artist_id'] = r[2] - except Exception: - pass + artist['image_url'] = r[1] or None + artist['id'] = r[2] + artist['global_listeners'] = r[3] + artist['global_playcount'] = r[4] + artist['soul_id'] = r[5] - for track in (cache.get('top_tracks') or []): - try: - cursor.execute(""" - SELECT al.thumb_url, t.id, t.artist_id FROM tracks t - JOIN albums al ON al.id = t.album_id - JOIN artists ar ON ar.id = t.artist_id - WHERE LOWER(t.title) = LOWER(?) AND LOWER(ar.name) = LOWER(?) LIMIT 1 - """, (track['name'], track.get('artist', ''))) - r = cursor.fetchone() + # ---- top_albums: match by LOWER(title) ---- + if top_albums: + titles = [a.get('name') or '' for a in top_albums] + unique_titles = {t.lower() for t in titles if t} + album_rows = {} + if unique_titles: + title_list = list(unique_titles) + chunk = 500 + for i in range(0, len(title_list), chunk): + sub = title_list[i:i + chunk] + placeholders = ','.join(['?'] * len(sub)) + cursor.execute( + f""" + SELECT LOWER(title), thumb_url, id, artist_id + FROM albums + WHERE LOWER(title) IN ({placeholders}) + """, + sub, + ) + for row in cursor.fetchall(): + album_rows.setdefault(row[0], row) + + for album in top_albums: + key = (album.get('name') or '').lower() + r = album_rows.get(key) if r: - track['image_url'] = r[0] or None - track['id'] = r[1] - track['artist_id'] = r[2] - except Exception: - pass - except Exception: - pass + album['image_url'] = r[1] or None + album['id'] = r[2] + album['artist_id'] = r[3] + + # ---- top_tracks: match by (LOWER(title), LOWER(artist name)) ---- + if top_tracks: + pairs = set() + for t in top_tracks: + name = (t.get('name') or '').lower() + artist = (t.get('artist') or '').lower() + if name: + pairs.add((name, artist)) + track_rows = {} + if pairs: + pair_list = list(pairs) + chunk = 500 + for i in range(0, len(pair_list), chunk): + sub = pair_list[i:i + chunk] + placeholders = ','.join(['(?,?)'] * len(sub)) + flat = [v for pair in sub for v in pair] + cursor.execute( + f""" + SELECT LOWER(t.title), LOWER(ar.name), + al.thumb_url, t.id, t.artist_id + FROM tracks t + JOIN albums al ON al.id = t.album_id + JOIN artists ar ON ar.id = t.artist_id + WHERE (LOWER(t.title), LOWER(ar.name)) IN ({placeholders}) + """, + flat, + ) + for row in cursor.fetchall(): + track_rows.setdefault((row[0], row[1]), row) + + for track in top_tracks: + key = ((track.get('name') or '').lower(), + (track.get('artist') or '').lower()) + r = track_rows.get(key) + if r: + track['image_url'] = r[2] or None + track['id'] = r[3] + track['artist_id'] = r[4] + except Exception as e: + logger.error(f"Error enriching stats items: {e}") finally: if conn: conn.close() @@ -454,30 +530,95 @@ class ListeningStatsWorker: if conn: conn.close() + def _resolve_db_track_ids_batch(self, events): + """Batch-resolve DB track IDs for a list of history events. + + Returns a dict ``{(title_lower, artist_lower): track_id}`` so callers + can look up without another DB round-trip. Replaces the former N+1 + pattern of one SELECT per event (500 events = 500 queries). + + Uses row-value IN with chunking (500 pairs = 1000 variables, well + under SQLite's default limit). Case-insensitive matching is preserved. + """ + pairs = set() + for ev in events: + title = (ev.get('title') or '').strip() + artist = (ev.get('artist') or '').strip() + if title: + pairs.add((title.lower(), artist.lower())) + + result = {} + if not pairs: + return result + + pair_list = list(pairs) + chunk_size = 500 + + conn = None + try: + conn = self.db._get_connection() + cursor = conn.cursor() + for i in range(0, len(pair_list), chunk_size): + chunk = pair_list[i:i + chunk_size] + placeholders = ','.join(['(?,?)'] * len(chunk)) + flat_args = [v for pair in chunk for v in pair] + cursor.execute( + f""" + SELECT LOWER(t.title), LOWER(ar.name), t.id + FROM tracks t + JOIN artists ar ON ar.id = t.artist_id + WHERE (LOWER(t.title), LOWER(ar.name)) IN ({placeholders}) + """, + flat_args, + ) + for title_l, artist_l, tid in cursor.fetchall(): + # Keep first match per pair to match the LIMIT 1 semantics + # of the original per-event query. + result.setdefault((title_l, artist_l), tid) + except Exception as e: + logger.error(f"Error batch-resolving track IDs: {e}") + finally: + if conn: + conn.close() + + return result + def _map_play_counts_to_db(self, server_counts, server_source): """Map server track IDs to DB track IDs for play count updates. - Looks up tracks by matching the server's track ID stored in - the tracks table (from library sync). + Looks up which server IDs exist in the tracks table. Replaces a + previous N+1 pattern of one SELECT per server ID with a single + batched IN query (chunked for safety). """ + if not server_counts: + return [] + conn = None try: conn = self.db._get_connection() cursor = conn.cursor() - # Build a lookup of server_id → db_track_id - # The tracks table stores server IDs as the primary 'id' column - updates = [] - for server_id, play_count in server_counts.items(): - cursor.execute("SELECT id FROM tracks WHERE id = ?", (server_id,)) - row = cursor.fetchone() - if row: - updates.append({ - 'db_track_id': row[0], - 'play_count': play_count, - 'last_played': None, # Could be fetched separately - }) - return updates + ids = list(server_counts.keys()) + existing = set() + chunk_size = 500 + for i in range(0, len(ids), chunk_size): + chunk = ids[i:i + chunk_size] + placeholders = ','.join(['?'] * len(chunk)) + cursor.execute( + f"SELECT id FROM tracks WHERE id IN ({placeholders})", + chunk, + ) + existing.update(r[0] for r in cursor.fetchall()) + + return [ + { + 'db_track_id': server_id, + 'play_count': play_count, + 'last_played': None, # Could be fetched separately + } + for server_id, play_count in server_counts.items() + if server_id in existing + ] except Exception as e: logger.error(f"Error mapping play counts: {e}") return [] diff --git a/core/metadata_cache.py b/core/metadata_cache.py index 50d547de..c52c91ac 100644 --- a/core/metadata_cache.py +++ b/core/metadata_cache.py @@ -326,20 +326,28 @@ class MetadataCache: """, (row['id'],)) conn.commit() - # Resolve entity IDs to full data + # Resolve entity IDs to full data via a single batched query + # (chunked to stay below SQLite's default variable limit). result_ids = json.loads(row['result_ids']) if not result_ids: return [] - results = [] - for eid in result_ids: - cursor.execute(""" - SELECT raw_json FROM metadata_cache_entities - WHERE source = ? AND entity_type = ? AND entity_id = ? - """, (source, search_type, eid)) - erow = cursor.fetchone() - if erow: - results.append(json.loads(erow['raw_json'])) + raw_by_id: Dict[str, dict] = {} + for i in range(0, len(result_ids), 500): + chunk = result_ids[i:i + 500] + placeholders = ','.join('?' * len(chunk)) + cursor.execute(f""" + SELECT entity_id, raw_json FROM metadata_cache_entities + WHERE source = ? AND entity_type = ? AND entity_id IN ({placeholders}) + """, [source, search_type, *chunk]) + for erow in cursor.fetchall(): + try: + raw_by_id[erow['entity_id']] = json.loads(erow['raw_json']) + except (ValueError, TypeError): + continue + + # Preserve the original result_ids ordering. + results = [raw_by_id[eid] for eid in result_ids if eid in raw_by_id] # Only return if we found all (or most) entries — partial results are unreliable if len(results) >= len(result_ids) * 0.8: diff --git a/core/musicbrainz_worker.py b/core/musicbrainz_worker.py index 79496d2b..08e02248 100644 --- a/core/musicbrainz_worker.py +++ b/core/musicbrainz_worker.py @@ -254,16 +254,27 @@ class MusicBrainzWorker: conn.close() def _get_existing_id(self, entity_type: str, entity_id: int) -> Optional[str]: - """Check if an entity already has a musicbrainz_id (e.g. from manual match).""" - table_map = {'artist': 'artists', 'album': 'albums', 'track': 'tracks'} - table = table_map.get(entity_type) - if not table: + """Check if an entity already has a MusicBrainz ID (e.g. from manual match). + + MusicBrainz ID columns differ per entity type: artists use `musicbrainz_id`, + albums use `musicbrainz_release_id`, and tracks use `musicbrainz_recording_id`. + Before this fix, all three were queried as `musicbrainz_id`, so the + existing-ID check silently failed for albums and tracks. + """ + table_config = { + 'artist': ('artists', 'musicbrainz_id'), + 'album': ('albums', 'musicbrainz_release_id'), + 'track': ('tracks', 'musicbrainz_recording_id'), + } + cfg = table_config.get(entity_type) + if not cfg: return None + table, column = cfg conn = None try: conn = self.db._get_connection() cursor = conn.cursor() - cursor.execute(f"SELECT musicbrainz_id FROM {table} WHERE id = ?", (entity_id,)) + cursor.execute(f"SELECT {column} FROM {table} WHERE id = ?", (entity_id,)) row = cursor.fetchone() return row[0] if row and row[0] else None except Exception: @@ -285,6 +296,17 @@ class MusicBrainzWorker: existing_id = self._get_existing_id(item_type, item_id) if existing_id: logger.debug(f"Preserving existing MusicBrainz ID for {item_type} '{item_name}': {existing_id}") + # Mark as matched so this row is not re-selected forever when + # match_status is NULL but the MBID is already populated. + try: + if item_type == 'artist': + self.mb_service.update_artist_mbid(item_id, existing_id, 'matched') + elif item_type == 'album': + self.mb_service.update_album_mbid(item_id, existing_id, 'matched') + elif item_type == 'track': + self.mb_service.update_track_mbid(item_id, existing_id, 'matched') + except Exception as mark_err: + logger.error(f"Error marking {item_type} #{item_id} matched: {mark_err}") return if item_type == 'artist': diff --git a/core/qobuz_worker.py b/core/qobuz_worker.py index 15f7da98..5e85c508 100644 --- a/core/qobuz_worker.py +++ b/core/qobuz_worker.py @@ -388,6 +388,8 @@ class QobuzWorker: existing_id = self._get_existing_id('artist', artist_id) if existing_id: logger.debug(f"Preserving existing Qobuz ID for artist '{artist_name}': {existing_id}") + # Mark as matched so this row is not re-selected on every loop. + self._mark_status('artist', artist_id, 'matched') return result = self.client.search_artist(artist_name) @@ -429,6 +431,7 @@ class QobuzWorker: existing_id = self._get_existing_id('album', album_id) if existing_id: logger.debug(f"Preserving existing Qobuz ID for album '{album_name}': {existing_id}") + self._mark_status('album', album_id, 'matched') return result = self.client.search_album(artist_name, album_name) @@ -484,6 +487,7 @@ class QobuzWorker: existing_id = self._get_existing_id('track', track_id) if existing_id: logger.debug(f"Preserving existing Qobuz ID for track '{track_name}': {existing_id}") + self._mark_status('track', track_id, 'matched') return result = self.client.search_track(artist_name, track_name) diff --git a/core/tidal_worker.py b/core/tidal_worker.py index 097227fc..fa578323 100644 --- a/core/tidal_worker.py +++ b/core/tidal_worker.py @@ -401,6 +401,8 @@ class TidalWorker: existing_id = self._get_existing_id('artist', artist_id) if existing_id: logger.debug(f"Preserving existing Tidal ID for artist '{artist_name}': {existing_id}") + # Mark as matched so this row is not re-selected on every loop. + self._mark_status('artist', artist_id, 'matched') return result = self.client.search_artist(artist_name) @@ -438,6 +440,7 @@ class TidalWorker: existing_id = self._get_existing_id('album', album_id) if existing_id: logger.debug(f"Preserving existing Tidal ID for album '{album_name}': {existing_id}") + self._mark_status('album', album_id, 'matched') return result = self.client.search_album(artist_name, album_name) @@ -486,6 +489,7 @@ class TidalWorker: existing_id = self._get_existing_id('track', track_id) if existing_id: logger.debug(f"Preserving existing Tidal ID for track '{track_name}': {existing_id}") + self._mark_status('track', track_id, 'matched') return result = self.client.search_track(artist_name, track_name) diff --git a/database/music_database.py b/database/music_database.py index 0690d1ab..8bc5684e 100644 --- a/database/music_database.py +++ b/database/music_database.py @@ -409,6 +409,12 @@ class MusicDatabase: # Add Discogs enrichment columns (migration) self._add_discogs_columns(cursor) + # Backfill match_status for rows that already have an external ID but + # NULL status. Prevents enrichment workers from re-processing these + # rows forever. Must run AFTER all *_match_status columns have been + # created by the migrations above. + self._backfill_match_status_for_existing_ids(cursor) + # Bubble snapshots table for persisting UI state across page refreshes cursor.execute(""" CREATE TABLE IF NOT EXISTS bubble_snapshots ( @@ -1922,6 +1928,56 @@ class MusicDatabase: except Exception as e: logger.error(f"Error adding Discogs columns: {e}") + def _backfill_match_status_for_existing_ids(self, cursor): + """Set `_match_status = 'matched'` for rows that already have a + populated external ID but NULL match_status. + + Prevents enrichment workers from re-selecting the same rows forever when + the ID was populated outside the worker (file tags, manual match, + pre-migration legacy data) without a corresponding status update. + + Only runs columns that actually exist, so pre-migration databases are + handled safely. UPDATE statements are cheap no-ops when nothing matches. + """ + # (table, id_column, status_column) + targets = [ + ('artists', 'lastfm_url', 'lastfm_match_status'), + ('albums', 'lastfm_url', 'lastfm_match_status'), + ('tracks', 'lastfm_url', 'lastfm_match_status'), + ('artists', 'musicbrainz_id', 'musicbrainz_match_status'), + ('albums', 'musicbrainz_release_id', 'musicbrainz_match_status'), + ('tracks', 'musicbrainz_recording_id', 'musicbrainz_match_status'), + ('artists', 'tidal_id', 'tidal_match_status'), + ('albums', 'tidal_id', 'tidal_match_status'), + ('tracks', 'tidal_id', 'tidal_match_status'), + ('artists', 'qobuz_id', 'qobuz_match_status'), + ('albums', 'qobuz_id', 'qobuz_match_status'), + ('tracks', 'qobuz_id', 'qobuz_match_status'), + ] + + total_backfilled = 0 + for table, id_col, status_col in targets: + try: + cursor.execute(f"PRAGMA table_info({table})") + cols = {row[1] for row in cursor.fetchall()} + if id_col not in cols or status_col not in cols: + continue + cursor.execute( + f"UPDATE {table} SET {status_col} = 'matched' " + f"WHERE {status_col} IS NULL AND {id_col} IS NOT NULL AND {id_col} != ''" + ) + if cursor.rowcount and cursor.rowcount > 0: + total_backfilled += cursor.rowcount + logger.info( + f"Backfilled {cursor.rowcount} rows in {table}.{status_col} " + f"where {id_col} was already set." + ) + except Exception as e: + logger.error(f"Error backfilling {table}.{status_col}: {e}") + + if total_backfilled == 0: + logger.debug("Match-status backfill: no rows needed updating.") + def _add_deezer_columns(self, cursor): """Add Deezer tracking + generic metadata columns for enrichment (artists, albums, tracks)""" try: @@ -5063,12 +5119,41 @@ class MusicDatabase: except Exception as e: logger.error(f"Error searching tracks with title='{title}', artist='{artist}': {e}") return [] + + def api_search_tracks(self, title: str = "", artist: str = "", limit: int = 50, + server_source: Optional[str] = None) -> List[Dict[str, Any]]: + """Search tracks and return full dict rows (all track columns plus artist_name, + album_title, album_thumb_url). Avoids the double-query pattern of calling + search_tracks() followed by api_get_tracks_by_ids(). + """ + try: + if not title and not artist: + return [] + + conn = self._get_connection() + cursor = conn.cursor() + + basic_rows = self._search_tracks_basic_rows(cursor, title, artist, limit, server_source) + if basic_rows: + return [dict(r) for r in basic_rows] + + fuzzy_rows = self._search_tracks_fuzzy_rows(cursor, title, artist, limit, server_source) + return [dict(r) for r in fuzzy_rows] + except Exception as e: + logger.error(f"API: Error searching tracks with title='{title}', artist='{artist}': {e}") + return [] def _search_tracks_basic(self, cursor, title: str, artist: str, limit: int, server_source: str = None) -> List[DatabaseTrack]: """Basic SQL LIKE search - fastest method""" + rows = self._search_tracks_basic_rows(cursor, title, artist, limit, server_source) + return self._rows_to_tracks(rows) + + def _search_tracks_basic_rows(self, cursor, title: str, artist: str, limit: int, + server_source: Optional[str] = None): + """Basic SQL LIKE search returning raw rows (shared by DatabaseTrack and dict-returning callers).""" where_conditions = [] params = [] - + if title: where_conditions.append("unidecode_lower(tracks.title) LIKE ?") params.append(f"%{self._normalize_for_comparison(title)}%") @@ -5083,13 +5168,13 @@ class MusicDatabase: if server_source: where_conditions.append("tracks.server_source = ?") params.append(server_source) - + if not where_conditions: return [] - + where_clause = " AND ".join(where_conditions) params.append(limit) - + cursor.execute(f""" SELECT tracks.*, artists.name as artist_name, albums.title as album_title, albums.thumb_url as album_thumb_url FROM tracks @@ -5100,45 +5185,47 @@ class MusicDatabase: LIMIT ? """, params) - return self._rows_to_tracks(cursor.fetchall()) + return cursor.fetchall() def _search_tracks_fuzzy_fallback(self, cursor, title: str, artist: str, limit: int, server_source: str = None) -> List[DatabaseTrack]: """Broadest fuzzy search - partial word matching""" + rows = self._search_tracks_fuzzy_rows(cursor, title, artist, limit, server_source) + return self._rows_to_tracks(rows) + + def _search_tracks_fuzzy_rows(self, cursor, title: str, artist: str, limit: int, + server_source: Optional[str] = None): + """Broadest fuzzy search returning raw rows (shared by DatabaseTrack and dict-returning callers).""" # Get broader results by searching for individual words search_terms = [] if title: - # Split title into words and search for each (normalized for diacritics) title_words = [w.strip() for w in self._normalize_for_comparison(title).split() if len(w.strip()) >= 3] search_terms.extend(title_words) if artist: - # Split artist into words and search for each (normalized for diacritics) artist_words = [w.strip() for w in self._normalize_for_comparison(artist).split() if len(w.strip()) >= 3] search_terms.extend(artist_words) - + if not search_terms: return [] - - # Build a query that searches for any of the words + like_conditions = [] params = [] - - for term in search_terms[:5]: # Limit to 5 terms to avoid too broad search + + for term in search_terms[:5]: like_conditions.append("(unidecode_lower(tracks.title) LIKE ? OR unidecode_lower(artists.name) LIKE ? OR unidecode_lower(COALESCE(tracks.track_artist, '')) LIKE ?)") params.extend([f"%{term}%", f"%{term}%", f"%{term}%"]) - + if not like_conditions: return [] - - # Build WHERE clause with optional server filter + where_parts = [f"({' OR '.join(like_conditions)})"] if server_source: where_parts.append("tracks.server_source = ?") - params.append(server_source) # Append after LIKE params, before LIMIT - + params.append(server_source) + where_clause = " AND ".join(where_parts) - params.append(limit * 3) # Get more results for scoring - + params.append(limit * 3) + cursor.execute(f""" SELECT tracks.*, artists.name as artist_name, albums.title as album_title, albums.thumb_url as album_thumb_url FROM tracks @@ -5154,23 +5241,19 @@ class MusicDatabase: # Score and filter results scored_results = [] for row in rows: - # Simple scoring based on how many search terms match score = 0 db_title_lower = self._normalize_for_comparison(row['title']) db_artist_lower = self._normalize_for_comparison(row['artist_name']) - + for term in search_terms: if term in db_title_lower or term in db_artist_lower: score += 1 - + if score > 0: scored_results.append((score, row)) - - # Sort by score and take top results + scored_results.sort(key=lambda x: x[0], reverse=True) - top_rows = [row for score, row in scored_results[:limit]] - - return self._rows_to_tracks(top_rows) + return [row for score, row in scored_results[:limit]] def _rows_to_tracks(self, rows) -> List[DatabaseTrack]: """Convert database rows to DatabaseTrack objects""" @@ -6632,8 +6715,14 @@ class MusicDatabase: logger.error(f"Error removing track from wishlist: {e}") return False - def get_wishlist_tracks(self, limit: Optional[int] = None, profile_id: int = 1) -> List[Dict[str, Any]]: - """Get all tracks in the wishlist for the given profile, ordered by date added (oldest first for retry priority)""" + def get_wishlist_tracks(self, limit: Optional[int] = None, profile_id: int = 1, + offset: int = 0, category: Optional[str] = None) -> List[Dict[str, Any]]: + """Get tracks in the wishlist for the given profile, ordered by date added + (oldest first for retry priority). + + Supports SQL-level pagination via limit/offset and optional category + filtering (singles vs albums) pushed down to SQL using json_extract. + """ try: with self._get_connection() as conn: cursor = conn.cursor() @@ -6643,12 +6732,26 @@ class MusicDatabase: last_attempted, date_added, source_type, source_info FROM wishlist_tracks WHERE profile_id = ? - ORDER BY date_added """ - params = [profile_id] + params: List[Any] = [profile_id] + + if category == "albums": + query += " AND json_extract(spotify_data, '$.album.album_type') = 'album'" + elif category == "singles": + query += ( + " AND (json_extract(spotify_data, '$.album.album_type') IS NULL" + " OR json_extract(spotify_data, '$.album.album_type') != 'album')" + ) + + query += " ORDER BY date_added" + if limit: - query += f" LIMIT {limit}" + query += " LIMIT ?" + params.append(int(limit)) + if offset: + query += " OFFSET ?" + params.append(int(offset)) cursor.execute(query, params) rows = cursor.fetchall() @@ -6679,7 +6782,7 @@ class MusicDatabase: except Exception as e: logger.error(f"Error getting wishlist tracks: {e}") return [] - + def update_wishlist_retry(self, spotify_track_id: str, success: bool, error_message: str = None, profile_id: int = 1) -> bool: """Update retry count and status for a wishlist track""" try: @@ -6706,12 +6809,22 @@ class MusicDatabase: logger.error(f"Error updating wishlist retry status: {e}") return False - def get_wishlist_count(self, profile_id: int = 1) -> int: - """Get the total number of tracks in the wishlist for the given profile""" + def get_wishlist_count(self, profile_id: int = 1, category: Optional[str] = None) -> int: + """Get the total number of tracks in the wishlist for the given profile, + optionally filtered by category ('singles' or 'albums').""" try: with self._get_connection() as conn: cursor = conn.cursor() - cursor.execute("SELECT COUNT(*) FROM wishlist_tracks WHERE profile_id = ?", (profile_id,)) + query = "SELECT COUNT(*) FROM wishlist_tracks WHERE profile_id = ?" + params: List[Any] = [profile_id] + if category == "albums": + query += " AND json_extract(spotify_data, '$.album.album_type') = 'album'" + elif category == "singles": + query += ( + " AND (json_extract(spotify_data, '$.album.album_type') IS NULL" + " OR json_extract(spotify_data, '$.album.album_type') != 'album')" + ) + cursor.execute(query, params) result = cursor.fetchone() return result[0] if result else 0 except Exception as e: diff --git a/tests/test_auth_usage_throttle.py b/tests/test_auth_usage_throttle.py new file mode 100644 index 00000000..690b0d45 --- /dev/null +++ b/tests/test_auth_usage_throttle.py @@ -0,0 +1,110 @@ +"""Unit tests for the auth last_used_at write throttle. + +Fix 1.2: every authenticated API request previously called +`config_mgr.set("api_keys", ...)`, which rewrites the entire app config +blob to SQLite. Writes are now throttled per key hash. +""" + +import sys +import threading +import types +from datetime import datetime, timedelta, timezone + +import pytest + + +# api/__init__.py eagerly imports flask_limiter. Tests only need the auth +# module, so stub flask_limiter before importing the api package. +def _install_flask_limiter_stub(): + if "flask_limiter" in sys.modules: + return + stub = types.ModuleType("flask_limiter") + + class _Limiter: + def __init__(self, *args, **kwargs): + pass + + def limit(self, *args, **kwargs): + def decorator(target): + return target + return decorator + + def init_app(self, app): + pass + + stub.Limiter = _Limiter + sys.modules["flask_limiter"] = stub + + util_stub = types.ModuleType("flask_limiter.util") + util_stub.get_remote_address = lambda: "127.0.0.1" + sys.modules["flask_limiter.util"] = util_stub + + +_install_flask_limiter_stub() + +from api import auth # noqa: E402 + + +@pytest.fixture(autouse=True) +def _reset_usage_cache(): + """Ensure a clean throttle cache for each test.""" + with auth._usage_lock: + auth._last_persisted_usage.clear() + yield + with auth._usage_lock: + auth._last_persisted_usage.clear() + + +def test_first_call_persists(): + now = datetime.now(timezone.utc) + assert auth._should_persist_usage("hash-a", now) is True + + +def test_second_call_within_interval_does_not_persist(): + start = datetime.now(timezone.utc) + assert auth._should_persist_usage("hash-a", start) is True + # 5 minutes later, still inside the 15-minute window + assert auth._should_persist_usage("hash-a", start + timedelta(minutes=5)) is False + + +def test_call_after_interval_persists_again(): + start = datetime.now(timezone.utc) + assert auth._should_persist_usage("hash-a", start) is True + later = start + auth._USAGE_WRITE_INTERVAL + assert auth._should_persist_usage("hash-a", later) is True + + +def test_different_keys_have_independent_throttles(): + now = datetime.now(timezone.utc) + assert auth._should_persist_usage("hash-a", now) is True + assert auth._should_persist_usage("hash-b", now) is True + # Both keys should now be throttled for the next 15 minutes + assert auth._should_persist_usage("hash-a", now + timedelta(minutes=1)) is False + assert auth._should_persist_usage("hash-b", now + timedelta(minutes=1)) is False + + +def test_concurrent_access_is_thread_safe(): + """Many threads racing on the same key should only produce one persist per window.""" + now = datetime.now(timezone.utc) + results: list[bool] = [] + results_lock = threading.Lock() + + def worker(): + decision = auth._should_persist_usage("hash-shared", now) + with results_lock: + results.append(decision) + + threads = [threading.Thread(target=worker) for _ in range(20)] + for t in threads: + t.start() + for t in threads: + t.join() + + # Exactly one thread should have won the race and persisted. + assert results.count(True) == 1 + assert results.count(False) == 19 + + +def test_usage_interval_matches_spec(): + """The throttle window should be 15 minutes (documented contract).""" + assert auth._USAGE_WRITE_INTERVAL == timedelta(minutes=15) diff --git a/tests/test_downloads_pagination.py b/tests/test_downloads_pagination.py new file mode 100644 index 00000000..af2c7b9a --- /dev/null +++ b/tests/test_downloads_pagination.py @@ -0,0 +1,174 @@ +"""Unit tests for downloads endpoint pagination. + +Fix 4.1: `GET /api/v1/downloads` previously returned every task in the +in-memory `download_tasks` dict on every call. With many downloads this +produces a huge payload. The endpoint now supports `limit`, `offset`, +and `status` query params and includes a `total` count. +""" + +import sys +import threading +import types +from unittest.mock import patch + +import pytest + + +# api/__init__.py eagerly imports flask_limiter. Stub before import. +def _install_flask_limiter_stub(): + if "flask_limiter" in sys.modules: + return + stub = types.ModuleType("flask_limiter") + + class _Limiter: + def __init__(self, *args, **kwargs): + pass + + def limit(self, *args, **kwargs): + def decorator(target): + return target + return decorator + + def init_app(self, app): + pass + + stub.Limiter = _Limiter + sys.modules["flask_limiter"] = stub + + util_stub = types.ModuleType("flask_limiter.util") + util_stub.get_remote_address = lambda: "127.0.0.1" + sys.modules["flask_limiter.util"] = util_stub + + +_install_flask_limiter_stub() + +from flask import Flask, Blueprint # noqa: E402 + +from api import downloads as downloads_mod # noqa: E402 + + +def _make_task(status="downloading", when=None): + return { + "status": status, + "track_name": f"Track {when}", + "artist_name": "Artist", + "album_name": "Album", + "username": "user", + "filename": "file.mp3", + "progress": 0, + "size": 1000, + "status_change_time": when or "2026-01-01T00:00:00", + } + + +def _make_app_with_tasks(tasks_dict): + """Create a minimal Flask app with the downloads blueprint mounted and + a fake `web_server` module exposing the given download_tasks dict.""" + fake_ws = types.ModuleType("web_server") + fake_ws.download_tasks = tasks_dict + fake_ws.tasks_lock = threading.RLock() + sys.modules["web_server"] = fake_ws + + # Bypass API key auth for tests. + def _passthrough(f): + return f + + app = Flask(__name__) + bp = Blueprint("v1", __name__, url_prefix="/api/v1") + + with patch.object(downloads_mod, "require_api_key", _passthrough): + # downloads.register_routes was already imported with the real + # decorator bound, but register_routes runs fresh decorators at + # call time against the passed blueprint. + downloads_mod.register_routes(bp) + + app.register_blueprint(bp) + return app + + +@pytest.fixture +def client(): + tasks = { + f"task-{i:03d}": _make_task( + status="downloading" if i % 2 == 0 else "queued", + when=f"2026-01-{i+1:02d}T00:00:00", + ) + for i in range(25) + } + app = _make_app_with_tasks(tasks) + with app.test_client() as c: + yield c + + +def test_default_limit_applied(client): + # 25 tasks, default limit is 100 -> all fit in one page. + resp = client.get("/api/v1/downloads") + assert resp.status_code == 200 + data = resp.get_json()["data"] + assert data["total"] == 25 + assert data["limit"] == 100 + assert data["offset"] == 0 + assert len(data["downloads"]) == 25 + + +def test_limit_and_offset_slice_correctly(client): + resp = client.get("/api/v1/downloads?limit=5&offset=0") + data = resp.get_json()["data"] + assert len(data["downloads"]) == 5 + assert data["total"] == 25 + + resp2 = client.get("/api/v1/downloads?limit=5&offset=20") + data2 = resp2.get_json()["data"] + assert len(data2["downloads"]) == 5 + # Pages should not overlap. + page1_ids = {t["id"] for t in data["downloads"]} + page5_ids = {t["id"] for t in data2["downloads"]} + assert page1_ids.isdisjoint(page5_ids) + + +def test_status_filter_single(client): + resp = client.get("/api/v1/downloads?status=downloading&limit=100") + data = resp.get_json()["data"] + # 13 even-indexed tasks (0,2,...,24) + assert data["total"] == 13 + for t in data["downloads"]: + assert t["status"] == "downloading" + + +def test_status_filter_multiple_comma_separated(client): + resp = client.get("/api/v1/downloads?status=downloading,queued&limit=100") + data = resp.get_json()["data"] + assert data["total"] == 25 + + +def test_status_filter_no_match_returns_empty(client): + resp = client.get("/api/v1/downloads?status=nonexistent_status") + data = resp.get_json()["data"] + assert data["total"] == 0 + assert data["downloads"] == [] + + +def test_limit_is_clamped_to_max(client): + resp = client.get("/api/v1/downloads?limit=99999") + data = resp.get_json()["data"] + assert data["limit"] == 500 + + +def test_negative_offset_is_normalized(client): + resp = client.get("/api/v1/downloads?offset=-5") + data = resp.get_json()["data"] + assert data["offset"] == 0 + + +def test_invalid_limit_falls_back_to_default(client): + resp = client.get("/api/v1/downloads?limit=not_a_number") + data = resp.get_json()["data"] + assert data["limit"] == 100 + + +def test_tasks_sorted_newest_first(client): + resp = client.get("/api/v1/downloads?limit=3&offset=0") + data = resp.get_json()["data"] + times = [t["status_change_time"] for t in data["downloads"]] + # Most recent (2026-01-25) should come first. + assert times == sorted(times, reverse=True) diff --git a/tests/test_listening_stats_batch_queries.py b/tests/test_listening_stats_batch_queries.py new file mode 100644 index 00000000..ea79b4be --- /dev/null +++ b/tests/test_listening_stats_batch_queries.py @@ -0,0 +1,266 @@ +"""Tests for batched queries in listening stats worker (fix 2.1). + +Before this fix the worker ran one SELECT per item for: + - resolving db_track_id when inserting history events + - mapping server play-count IDs to existing DB track IDs + - enriching top_artists / top_albums / top_tracks in the stats cache + +Each pattern was N+1 on the DB. The fix replaces them with single +batched IN queries (chunked for safety). +""" + +from __future__ import annotations + +import pytest + +from database.music_database import MusicDatabase +from core.listening_stats_worker import ListeningStatsWorker + + +class _FakeConfigManager: + def get(self, key, default=None): + return default + + +@pytest.fixture +def db(tmp_path): + return MusicDatabase(str(tmp_path / "music.db")) + + +@pytest.fixture +def worker(db): + return ListeningStatsWorker(db, _FakeConfigManager()) + + +def _install_query_counter(db): + """Replace db._get_connection with a wrapper that counts execute() calls. + + Returns the counter dict (has key 'n') and a restore callback. + """ + original_get_connection = db._get_connection + counter = {"n": 0} + + class _CursorProxy: + def __init__(self, real_cursor): + self._real = real_cursor + + def execute(self, sql, params=()): + counter["n"] += 1 + return self._real.execute(sql, params) + + def __getattr__(self, name): + return getattr(self._real, name) + + def __iter__(self): + return iter(self._real) + + class _ConnProxy: + def __init__(self, real_conn): + self._real = real_conn + + def cursor(self, *a, **k): + return _CursorProxy(self._real.cursor(*a, **k)) + + def __enter__(self): + self._real.__enter__() + return self + + def __exit__(self, *exc): + return self._real.__exit__(*exc) + + def __getattr__(self, name): + return getattr(self._real, name) + + def wrapped(): + return _ConnProxy(original_get_connection()) + + db._get_connection = wrapped + + def restore(): + db._get_connection = original_get_connection + + return counter, restore + + +def _insert_track(db, track_id, title, artist_id, artist_name, album_id, album_title): + with db._get_connection() as conn: + conn.execute( + "INSERT OR IGNORE INTO artists (id, name) VALUES (?, ?)", + (artist_id, artist_name), + ) + conn.execute( + "INSERT OR IGNORE INTO albums (id, title, artist_id, thumb_url) VALUES (?, ?, ?, ?)", + (album_id, album_title, artist_id, f"http://img/{album_id}.jpg"), + ) + conn.execute( + """INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path) + VALUES (?, ?, ?, ?, 1, 180, ?)""", + (track_id, album_id, artist_id, title, f"/music/{title}.mp3"), + ) + conn.commit() + + +# --------------------------------------------------------------------------- +# _resolve_db_track_ids_batch +# --------------------------------------------------------------------------- + +class TestResolveDbTrackIdsBatch: + def test_batch_returns_same_ids_as_per_event_lookup(self, db, worker): + _insert_track(db, "t1", "Alpha", "a1", "Band One", "al1", "X") + _insert_track(db, "t2", "Bravo", "a1", "Band One", "al1", "X") + _insert_track(db, "t3", "Alpha", "a2", "Band Two", "al2", "Y") + + events = [ + {"title": "Alpha", "artist": "Band One"}, + {"title": "Bravo", "artist": "Band One"}, + {"title": "Alpha", "artist": "Band Two"}, + {"title": "Nonexistent", "artist": "Nobody"}, + ] + + id_map = worker._resolve_db_track_ids_batch(events) + + assert id_map[("alpha", "band one")] == "t1" + assert id_map[("bravo", "band one")] == "t2" + assert id_map[("alpha", "band two")] == "t3" + assert ("nonexistent", "nobody") not in id_map + + def test_is_case_insensitive(self, db, worker): + _insert_track(db, "t1", "Great Song", "a1", "Some Band", "al1", "X") + + id_map = worker._resolve_db_track_ids_batch( + [{"title": "GREAT SONG", "artist": "SOME BAND"}] + ) + assert id_map[("great song", "some band")] == "t1" + + def test_empty_list_returns_empty_dict(self, worker): + assert worker._resolve_db_track_ids_batch([]) == {} + + def test_events_without_title_are_skipped(self, db, worker): + _insert_track(db, "t1", "Song", "a1", "Band", "al1", "X") + id_map = worker._resolve_db_track_ids_batch( + [{"title": "", "artist": "Band"}, {"title": "Song", "artist": "Band"}] + ) + assert id_map == {("song", "band"): "t1"} + + def test_runs_single_query_regardless_of_event_count(self, db, worker): + """The whole point: 50 events must not trigger 50 queries.""" + for i in range(50): + _insert_track(db, f"t{i}", f"Song {i}", "a1", "Band", "al1", "Album") + + counter, restore = _install_query_counter(db) + try: + events = [{"title": f"Song {i}", "artist": "Band"} for i in range(50)] + id_map = worker._resolve_db_track_ids_batch(events) + finally: + restore() + + assert len(id_map) == 50 + # One batched query (everything fits in one chunk). + assert counter["n"] == 1 + + +# --------------------------------------------------------------------------- +# _map_play_counts_to_db +# --------------------------------------------------------------------------- + +class TestMapPlayCountsToDb: + def test_returns_updates_only_for_existing_ids(self, db, worker): + _insert_track(db, "t1", "A", "a1", "Band", "al1", "Album") + _insert_track(db, "t2", "B", "a1", "Band", "al1", "Album") + + server_counts = {"t1": 5, "t2": 3, "ghost": 99} + updates = worker._map_play_counts_to_db(server_counts, "plex") + + ids = {u["db_track_id"]: u["play_count"] for u in updates} + assert ids == {"t1": 5, "t2": 3} + + def test_empty_input_returns_empty_list(self, worker): + assert worker._map_play_counts_to_db({}, "plex") == [] + + def test_runs_single_query_regardless_of_count_size(self, db, worker): + for i in range(30): + _insert_track(db, f"t{i}", f"S{i}", "a1", "Band", "al1", "Album") + + counter, restore = _install_query_counter(db) + try: + server_counts = {f"t{i}": i for i in range(30)} + updates = worker._map_play_counts_to_db(server_counts, "plex") + finally: + restore() + + assert len(updates) == 30 + assert counter["n"] == 1 + + +# --------------------------------------------------------------------------- +# _enrich_stats_items +# --------------------------------------------------------------------------- + +class TestEnrichStatsItems: + def test_enriches_artists_albums_and_tracks(self, db, worker): + _insert_track(db, "t1", "Alpha", "a1", "Band One", "al1", "First Album") + _insert_track(db, "t2", "Bravo", "a2", "Band Two", "al2", "Second Album") + + cache = { + "top_artists": [{"name": "Band One"}, {"name": "Band Two"}], + "top_albums": [{"name": "First Album"}, {"name": "Second Album"}], + "top_tracks": [ + {"name": "Alpha", "artist": "Band One"}, + {"name": "Bravo", "artist": "Band Two"}, + ], + } + + worker._enrich_stats_items(cache) + + by_name = {a["name"]: a for a in cache["top_artists"]} + assert by_name["Band One"]["id"] == "a1" + assert by_name["Band Two"]["id"] == "a2" + + album_by_name = {a["name"]: a for a in cache["top_albums"]} + assert album_by_name["First Album"]["id"] == "al1" + assert album_by_name["First Album"]["image_url"] == "http://img/al1.jpg" + + track_by_name = {t["name"]: t for t in cache["top_tracks"]} + assert track_by_name["Alpha"]["id"] == "t1" + assert track_by_name["Bravo"]["id"] == "t2" + assert track_by_name["Alpha"]["artist_id"] == "a1" + + def test_unknown_entries_left_untouched(self, db, worker): + _insert_track(db, "t1", "Real", "a1", "Real Band", "al1", "Real Album") + + cache = { + "top_artists": [{"name": "Unknown Band"}], + "top_albums": [{"name": "Unknown Album"}], + "top_tracks": [{"name": "Unknown", "artist": "Nobody"}], + } + worker._enrich_stats_items(cache) + + assert "id" not in cache["top_artists"][0] + assert "id" not in cache["top_albums"][0] + assert "id" not in cache["top_tracks"][0] + + def test_empty_cache_is_safe(self, worker): + worker._enrich_stats_items({}) # must not raise + worker._enrich_stats_items({"top_artists": [], "top_albums": [], "top_tracks": []}) + + def test_runs_one_query_per_section(self, db, worker): + for i in range(20): + _insert_track(db, f"t{i}", f"Song {i}", f"a{i}", f"Band {i}", + f"al{i}", f"Album {i}") + + cache = { + "top_artists": [{"name": f"Band {i}"} for i in range(20)], + "top_albums": [{"name": f"Album {i}"} for i in range(20)], + "top_tracks": [ + {"name": f"Song {i}", "artist": f"Band {i}"} for i in range(20) + ], + } + + counter, restore = _install_query_counter(db) + try: + worker._enrich_stats_items(cache) + finally: + restore() + + # 3 batched queries total (artists + albums + tracks), not 60. + assert counter["n"] == 3 diff --git a/tests/test_metadata_cache_batch_lookup.py b/tests/test_metadata_cache_batch_lookup.py new file mode 100644 index 00000000..947edb83 --- /dev/null +++ b/tests/test_metadata_cache_batch_lookup.py @@ -0,0 +1,217 @@ +"""Unit tests for batched metadata cache entity lookups. + +Fix 1.3: `MetadataCache.get_search_results` previously resolved cached +entity IDs one-by-one, producing N extra SELECT queries per cached +search. The resolution now runs as a single batched `IN` query (chunked +to stay below SQLite's variable limit) and preserves the original +`result_ids` ordering. +""" + +import json +import sqlite3 +from datetime import datetime, timedelta +from unittest.mock import MagicMock + +import pytest + +from core.metadata_cache import MetadataCache + + +@pytest.fixture +def cache_with_db(tmp_path): + """MetadataCache wired to a temporary SQLite DB with the required tables.""" + db_path = tmp_path / "cache.db" + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.executescript( + """ + CREATE TABLE metadata_cache_searches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + search_type TEXT NOT NULL, + query_normalized TEXT NOT NULL, + search_limit INTEGER NOT NULL, + result_ids TEXT NOT NULL, + created_at TEXT NOT NULL, + last_accessed_at TEXT, + access_count INTEGER DEFAULT 0 + ); + CREATE TABLE metadata_cache_entities ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + entity_type TEXT NOT NULL, + entity_id TEXT NOT NULL, + raw_json TEXT NOT NULL + ); + """ + ) + + # Fake MusicDatabase that yields this sqlite connection. + fake_db = MagicMock() + + def _get_connection(): + c = sqlite3.connect(str(db_path)) + c.row_factory = sqlite3.Row + return c + + fake_db._get_connection.side_effect = _get_connection + + cache = MetadataCache() + cache._get_db = lambda: fake_db # type: ignore[method-assign] + + return cache, conn + + +def _insert_search(conn, source, search_type, query, limit, result_ids, created_at=None): + created_at = created_at or datetime.now().isoformat() + conn.execute( + """INSERT INTO metadata_cache_searches + (source, search_type, query_normalized, search_limit, result_ids, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (source, search_type, query.lower(), limit, json.dumps(result_ids), created_at), + ) + conn.commit() + + +def _insert_entity(conn, source, entity_type, entity_id, payload): + conn.execute( + """INSERT INTO metadata_cache_entities + (source, entity_type, entity_id, raw_json) VALUES (?, ?, ?, ?)""", + (source, entity_type, entity_id, json.dumps(payload)), + ) + conn.commit() + + +def test_returns_results_in_original_order(cache_with_db): + cache, conn = cache_with_db + ids = ["z", "a", "m", "b"] + _insert_search(conn, "spotify", "track", "hello", 50, ids) + for eid in ids: + _insert_entity(conn, "spotify", "track", eid, {"id": eid, "name": f"track {eid}"}) + + results = cache.get_search_results("spotify", "track", "hello", 50) + + assert results is not None + assert [r["id"] for r in results] == ids + + +def test_missing_entities_below_threshold_returns_none(cache_with_db): + cache, conn = cache_with_db + ids = [f"id{i}" for i in range(10)] + _insert_search(conn, "spotify", "track", "partial", 50, ids) + # Only insert 5/10 entities — below the 80 percent threshold. + for eid in ids[:5]: + _insert_entity(conn, "spotify", "track", eid, {"id": eid}) + + assert cache.get_search_results("spotify", "track", "partial", 50) is None + + +def test_missing_entities_at_threshold_returns_partial(cache_with_db): + cache, conn = cache_with_db + ids = [f"id{i}" for i in range(10)] + _insert_search(conn, "spotify", "track", "threshold", 50, ids) + # Insert 8/10 = exactly 80 percent — should return the 8 found. + for eid in ids[:8]: + _insert_entity(conn, "spotify", "track", eid, {"id": eid}) + + results = cache.get_search_results("spotify", "track", "threshold", 50) + assert results is not None + assert len(results) == 8 + assert [r["id"] for r in results] == ids[:8] + + +def test_empty_result_ids_returns_empty_list(cache_with_db): + cache, conn = cache_with_db + _insert_search(conn, "spotify", "track", "empty", 50, []) + assert cache.get_search_results("spotify", "track", "empty", 50) == [] + + +def test_expired_search_returns_none(cache_with_db): + cache, conn = cache_with_db + old = (datetime.now() - timedelta(days=10)).isoformat() + _insert_search(conn, "spotify", "track", "stale", 50, ["a"], created_at=old) + _insert_entity(conn, "spotify", "track", "a", {"id": "a"}) + + assert cache.get_search_results("spotify", "track", "stale", 50) is None + + +def test_cache_miss_on_unknown_query(cache_with_db): + cache, _ = cache_with_db + assert cache.get_search_results("spotify", "track", "nothing cached", 50) is None + + +def test_batch_lookup_uses_single_round_trip(tmp_path): + """Sanity check that resolution does not issue one SELECT per entity_id.""" + db_path = tmp_path / "cache2.db" + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.executescript( + """ + CREATE TABLE metadata_cache_searches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT, search_type TEXT, query_normalized TEXT, + search_limit INTEGER, result_ids TEXT, created_at TEXT, + last_accessed_at TEXT, access_count INTEGER DEFAULT 0 + ); + CREATE TABLE metadata_cache_entities ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT, entity_type TEXT, entity_id TEXT, raw_json TEXT + ); + """ + ) + ids = [f"e{i}" for i in range(50)] + _insert_search(conn, "spotify", "track", "bulk", 50, ids) + for eid in ids: + _insert_entity(conn, "spotify", "track", eid, {"id": eid}) + + raw_selects = {"n": 0} + + class CountingConnection: + def __init__(self, inner): + self._inner = inner + + def cursor(self): + inner_cursor = self._inner.cursor() + + class CountingCursor: + def __init__(self, c): + self._c = c + + def execute(self, sql, params=()): + if "raw_json FROM metadata_cache_entities" in sql: + raw_selects["n"] += 1 + return self._c.execute(sql, params) + + def fetchone(self): + return self._c.fetchone() + + def fetchall(self): + return self._c.fetchall() + + return CountingCursor(inner_cursor) + + def commit(self): + return self._inner.commit() + + def close(self): + return self._inner.close() + + def __getattr__(self, name): + return getattr(self._inner, name) + + def _connect(): + c = sqlite3.connect(str(db_path)) + c.row_factory = sqlite3.Row + return CountingConnection(c) + + fake_db = MagicMock() + fake_db._get_connection.side_effect = _connect + cache = MetadataCache() + cache._get_db = lambda: fake_db # type: ignore[method-assign] + + results = cache.get_search_results("spotify", "track", "bulk", 50) + + assert results is not None and len(results) == 50 + # With batching, 50 entities should resolve in a single SELECT, not 50. + assert raw_selects["n"] == 1 + diff --git a/tests/test_request_cleanup_timer.py b/tests/test_request_cleanup_timer.py new file mode 100644 index 00000000..d54ae267 --- /dev/null +++ b/tests/test_request_cleanup_timer.py @@ -0,0 +1,137 @@ +"""Tests for api/request.py periodic cleanup timer (fix 4.2). + +Before this fix, `_cleanup_old_requests()` was only invoked on +`create_request`. During idle periods stale entries lingered for the +full uptime of the server. A background timer now runs every +_CLEANUP_INTERVAL_SECONDS and evicts anything older than _MAX_REQUEST_AGE. +""" + +import sys +import threading +import time +import types +from datetime import datetime, timedelta + +import pytest + + +# api/__init__.py imports flask_limiter at module load. Stub it. +def _install_flask_limiter_stub(): + if "flask_limiter" in sys.modules: + return + stub = types.ModuleType("flask_limiter") + + class _Limiter: + def __init__(self, *args, **kwargs): + pass + + def limit(self, *args, **kwargs): + def decorator(target): + return target + return decorator + + def init_app(self, app): + pass + + stub.Limiter = _Limiter + sys.modules["flask_limiter"] = stub + + util_stub = types.ModuleType("flask_limiter.util") + util_stub.get_remote_address = lambda: "127.0.0.1" + sys.modules["flask_limiter.util"] = util_stub + + +_install_flask_limiter_stub() + +from api import request as request_mod # noqa: E402 + + +@pytest.fixture(autouse=True) +def _clean_state(): + # Ensure no leftover thread from a previous test. + request_mod.stop_cleanup_thread(timeout=1.0) + with request_mod._requests_lock: + request_mod._pending_requests.clear() + yield + request_mod.stop_cleanup_thread(timeout=1.0) + with request_mod._requests_lock: + request_mod._pending_requests.clear() + + +def _add_request(request_id, age_minutes): + with request_mod._requests_lock: + request_mod._pending_requests[request_id] = { + "request_id": request_id, + "query": "q", + "status": "queued", + "created_at": datetime.now() - timedelta(minutes=age_minutes), + "completed_at": None, + "download_id": None, + "error": None, + } + + +class TestCleanupOldRequests: + def test_evicts_entries_older_than_ttl(self): + _add_request("old-1", age_minutes=120) # > 60 min TTL + _add_request("old-2", age_minutes=61) + _add_request("fresh", age_minutes=5) + + removed = request_mod._cleanup_old_requests() + + assert removed == 2 + with request_mod._requests_lock: + ids = set(request_mod._pending_requests.keys()) + assert ids == {"fresh"} + + def test_returns_zero_when_nothing_to_evict(self): + _add_request("fresh", age_minutes=5) + assert request_mod._cleanup_old_requests() == 0 + + def test_empty_map_is_safe(self): + assert request_mod._cleanup_old_requests() == 0 + + +class TestCleanupThreadLifecycle: + def test_start_returns_true_first_time_false_after(self): + assert request_mod.start_cleanup_thread() is True + # Second call in the same process should not start a new thread. + assert request_mod.start_cleanup_thread() is False + + def test_stop_joins_thread(self): + request_mod.start_cleanup_thread() + assert request_mod._cleanup_thread is not None + assert request_mod._cleanup_thread.is_alive() + + request_mod.stop_cleanup_thread(timeout=2.0) + assert request_mod._cleanup_thread is None + + def test_thread_evicts_on_wakeup(self, monkeypatch): + # Force a tiny interval so the test doesn't wait 5 minutes. + monkeypatch.setattr(request_mod, "_CLEANUP_INTERVAL_SECONDS", 0.05) + + _add_request("old", age_minutes=120) + request_mod.start_cleanup_thread() + + # Give the loop time to wake up and run at least once. + deadline = time.time() + 2.0 + while time.time() < deadline: + with request_mod._requests_lock: + remaining = set(request_mod._pending_requests.keys()) + if "old" not in remaining: + break + time.sleep(0.05) + + with request_mod._requests_lock: + assert "old" not in request_mod._pending_requests + + def test_stop_signals_thread_to_exit_promptly(self, monkeypatch): + # With a huge interval, stop must still make the thread exit via the + # stop event, not wait for the next timeout. + monkeypatch.setattr(request_mod, "_CLEANUP_INTERVAL_SECONDS", 30.0) + request_mod.start_cleanup_thread() + thread = request_mod._cleanup_thread + assert thread is not None + + request_mod.stop_cleanup_thread(timeout=2.0) + assert not thread.is_alive() diff --git a/tests/test_track_search_single_query.py b/tests/test_track_search_single_query.py new file mode 100644 index 00000000..061e3f3e --- /dev/null +++ b/tests/test_track_search_single_query.py @@ -0,0 +1,99 @@ +"""Unit tests for api_search_tracks (fix 2.2). + +The track search endpoint previously called search_tracks() then +api_get_tracks_by_ids() to re-hydrate the same rows. api_search_tracks +now returns the full dict rows in a single query. +""" + +import pytest + +from database.music_database import MusicDatabase + + +@pytest.fixture +def db(tmp_path): + return MusicDatabase(str(tmp_path / "music.db")) + + +def _insert_track(db, track_id, title, artist_id, artist_name, album_id, album_title): + with db._get_connection() as conn: + conn.execute( + "INSERT OR IGNORE INTO artists (id, name) VALUES (?, ?)", + (artist_id, artist_name), + ) + conn.execute( + "INSERT OR IGNORE INTO albums (id, title, artist_id) VALUES (?, ?, ?)", + (album_id, album_title, artist_id), + ) + conn.execute( + """INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path) + VALUES (?, ?, ?, ?, 1, 180, ?)""", + (track_id, album_id, artist_id, title, f"/music/{title}.mp3"), + ) + conn.commit() + + +def test_api_search_tracks_returns_dict_rows_with_full_columns(db): + _insert_track(db, 100, "Great Song", 1, "Band One", 10, "Album X") + + results = db.api_search_tracks(title="great") + + assert len(results) == 1 + row = results[0] + assert isinstance(row, dict) + assert str(row["id"]) == "100" + assert row["title"] == "Great Song" + assert row["artist_name"] == "Band One" + assert row["album_title"] == "Album X" + # file_path is a tracks.* column that must be present + assert row["file_path"] == "/music/Great Song.mp3" + + +def test_api_search_tracks_empty_query_returns_empty(db): + _insert_track(db, 1, "Track", 1, "Artist", 1, "Album") + assert db.api_search_tracks() == [] + + +def test_api_search_tracks_no_matches_returns_empty(db): + _insert_track(db, 1, "Alpha", 1, "Artist A", 1, "Album A") + assert db.api_search_tracks(title="nonexistent") == [] + + +def test_api_search_tracks_matches_by_artist(db): + _insert_track(db, 1, "Song A", 1, "Mystery Band", 1, "Album A") + _insert_track(db, 2, "Song B", 2, "Other Artist", 2, "Album B") + + results = db.api_search_tracks(artist="mystery") + assert len(results) == 1 + assert str(results[0]["id"]) == "1" + + +def test_api_search_tracks_respects_limit(db): + for i in range(10): + _insert_track(db, i + 1, f"Common Title {i}", 1, "Artist", 1, "Album") + + results = db.api_search_tracks(title="common", limit=3) + assert len(results) == 3 + + +def test_api_search_tracks_falls_back_to_fuzzy(db): + """When basic LIKE returns nothing, fuzzy fallback should still find matches.""" + _insert_track(db, 1, "Bohemian Rhapsody", 1, "Queen", 1, "A Night at the Opera") + + # Basic search "bohemian" matches directly; the fuzzy path exists for + # cases where basic fails. Verify at least that the method surfaces a + # result for a direct query. + results = db.api_search_tracks(title="bohemian rhapsody") + assert any(str(r["id"]) == "1" for r in results) + + +def test_search_tracks_still_returns_database_track_objects(db): + """Existing callers rely on the DatabaseTrack return shape of search_tracks().""" + _insert_track(db, 1, "Legacy Track", 1, "Legacy Artist", 1, "Legacy Album") + + results = db.search_tracks(title="legacy") + assert len(results) == 1 + # DatabaseTrack exposes attributes, not dict keys. + assert str(results[0].id) == "1" + assert results[0].title == "Legacy Track" + assert results[0].artist_name == "Legacy Artist" diff --git a/tests/test_wishlist_pagination.py b/tests/test_wishlist_pagination.py new file mode 100644 index 00000000..7543db4b --- /dev/null +++ b/tests/test_wishlist_pagination.py @@ -0,0 +1,131 @@ +"""Unit tests for SQL-level wishlist pagination and category filtering. + +Fix 1.4: the API endpoint previously loaded the entire wishlist, filtered +by category in Python, then sliced for the requested page. Pagination and +category filtering are now pushed to SQL via LIMIT/OFFSET and json_extract. +""" + +import json + +import pytest + +from database.music_database import MusicDatabase + + +@pytest.fixture +def db(tmp_path): + return MusicDatabase(str(tmp_path / "music.db")) + + +def _insert(db, spotify_track_id, name, album_type, profile_id=1, date_added=None): + spotify_data = { + "id": spotify_track_id, + "name": name, + "artists": [{"name": "Test Artist"}], + "album": {"name": f"Album for {name}", "album_type": album_type, "images": []}, + } + date_added = date_added or f"2026-01-{len(spotify_track_id):02d}T00:00:00" + with db._get_connection() as conn: + conn.execute( + """INSERT INTO wishlist_tracks + (spotify_track_id, spotify_data, failure_reason, retry_count, + date_added, source_type, source_info, profile_id) + VALUES (?, ?, '', 0, ?, 'test', '{}', ?)""", + (spotify_track_id, json.dumps(spotify_data), date_added, profile_id), + ) + conn.commit() + + +def test_pagination_limit_offset(db): + for i in range(5): + _insert(db, f"t{i}", f"Track {i}", "album", date_added=f"2026-01-0{i+1}T00:00:00") + + page1 = db.get_wishlist_tracks(limit=2, offset=0) + page2 = db.get_wishlist_tracks(limit=2, offset=2) + page3 = db.get_wishlist_tracks(limit=2, offset=4) + + assert [t["spotify_track_id"] for t in page1] == ["t0", "t1"] + assert [t["spotify_track_id"] for t in page2] == ["t2", "t3"] + assert [t["spotify_track_id"] for t in page3] == ["t4"] + + +def test_category_filter_albums(db): + _insert(db, "a1", "Album Track 1", "album", date_added="2026-01-01T00:00:00") + _insert(db, "s1", "Single 1", "single", date_added="2026-01-02T00:00:00") + _insert(db, "a2", "Album Track 2", "album", date_added="2026-01-03T00:00:00") + _insert(db, "e1", "EP Track", "ep", date_added="2026-01-04T00:00:00") + + albums = db.get_wishlist_tracks(category="albums") + assert sorted(t["spotify_track_id"] for t in albums) == ["a1", "a2"] + + +def test_category_filter_singles_includes_non_album_types(db): + _insert(db, "a1", "Album", "album", date_added="2026-01-01T00:00:00") + _insert(db, "s1", "Single", "single", date_added="2026-01-02T00:00:00") + _insert(db, "e1", "EP", "ep", date_added="2026-01-03T00:00:00") + _insert(db, "c1", "Compilation", "compilation", date_added="2026-01-04T00:00:00") + + singles = db.get_wishlist_tracks(category="singles") + assert sorted(t["spotify_track_id"] for t in singles) == ["c1", "e1", "s1"] + + +def test_category_filter_singles_includes_missing_album_type(db): + # Manually insert a row whose album has no album_type (malformed) + with db._get_connection() as conn: + conn.execute( + """INSERT INTO wishlist_tracks + (spotify_track_id, spotify_data, failure_reason, retry_count, + date_added, source_type, source_info, profile_id) + VALUES ('x1', ?, '', 0, '2026-01-01', 'test', '{}', 1)""", + (json.dumps({"id": "x1", "name": "X", "album": {"name": "A"}}),), + ) + conn.commit() + + singles = db.get_wishlist_tracks(category="singles") + assert [t["spotify_track_id"] for t in singles] == ["x1"] + + +def test_get_wishlist_count_no_filter(db): + for i in range(7): + _insert(db, f"t{i}", f"Track {i}", "album", date_added=f"2026-01-0{i+1}T00:00:00") + + assert db.get_wishlist_count() == 7 + + +def test_get_wishlist_count_with_category_filter(db): + _insert(db, "a1", "A1", "album", date_added="2026-01-01T00:00:00") + _insert(db, "a2", "A2", "album", date_added="2026-01-02T00:00:00") + _insert(db, "s1", "S1", "single", date_added="2026-01-03T00:00:00") + + assert db.get_wishlist_count(category="albums") == 2 + assert db.get_wishlist_count(category="singles") == 1 + assert db.get_wishlist_count() == 3 + + +def test_profile_isolation(db): + _insert(db, "p1-a", "A", "album", profile_id=1, date_added="2026-01-01T00:00:00") + _insert(db, "p1-b", "B", "album", profile_id=1, date_added="2026-01-02T00:00:00") + _insert(db, "p2-a", "C", "album", profile_id=2, date_added="2026-01-03T00:00:00") + + assert db.get_wishlist_count(profile_id=1) == 2 + assert db.get_wishlist_count(profile_id=2) == 1 + + p1 = db.get_wishlist_tracks(profile_id=1) + assert sorted(t["spotify_track_id"] for t in p1) == ["p1-a", "p1-b"] + + +def test_backward_compat_no_args_returns_all(db): + """Existing callers (wishlist_service) pass no limit/offset — must still work.""" + for i in range(3): + _insert(db, f"t{i}", f"Track {i}", "album", date_added=f"2026-01-0{i+1}T00:00:00") + + rows = db.get_wishlist_tracks() + assert len(rows) == 3 + + +def test_ordering_by_date_added(db): + _insert(db, "newer", "Newer", "album", date_added="2026-02-01T00:00:00") + _insert(db, "older", "Older", "album", date_added="2026-01-01T00:00:00") + + rows = db.get_wishlist_tracks() + assert [r["spotify_track_id"] for r in rows] == ["older", "newer"] diff --git a/tests/test_worker_existing_id_marks_matched.py b/tests/test_worker_existing_id_marks_matched.py new file mode 100644 index 00000000..2853e1fc --- /dev/null +++ b/tests/test_worker_existing_id_marks_matched.py @@ -0,0 +1,333 @@ +"""Tests for Fix 1.1: worker re-processing loop. + +Before this fix: + * `musicbrainz_worker._get_existing_id` always queried `musicbrainz_id` even + for `albums`/`tracks` (which use `musicbrainz_release_id` / + `musicbrainz_recording_id`), so the existence check silently failed and + every row was re-processed on every loop. + * `lastfm_worker._get_existing_id` queried a non-existent `lastfm_id` + column (the real column is `lastfm_url`), with the same effect. + * Even when workers did find an existing external ID, they returned + without setting `_match_status`, so the row stayed NULL and + the next worker loop re-selected it forever. + +This test module covers: + 1. The backfill migration that retroactively sets match_status='matched' + for rows that already have an external ID populated. + 2. `_get_existing_id` returns the correct column per entity type for + MusicBrainz and Last.fm. + 3. Each worker's `_process_*` short-circuit path sets match_status to + 'matched' when an existing external ID is found (lastfm, tidal, + qobuz, musicbrainz). +""" + +from __future__ import annotations + +import sys +import types +from unittest.mock import MagicMock, patch + +import pytest + +from database.music_database import MusicDatabase + + +# --------------------------------------------------------------------------- +# Minimal stubs for optional deps some workers import at module load. +# --------------------------------------------------------------------------- + +def _ensure_stub_module(name: str, attrs: dict | None = None) -> None: + if name in sys.modules: + return + mod = types.ModuleType(name) + for k, v in (attrs or {}).items(): + setattr(mod, k, v) + sys.modules[name] = mod + + +# TidalClient / QobuzClient live in core.* and are safe to import but require +# config_manager. We patch the classes at instantiation time instead. + + +@pytest.fixture +def db(tmp_path): + return MusicDatabase(str(tmp_path / "music.db")) + + +# --------------------------------------------------------------------------- +# Backfill migration +# --------------------------------------------------------------------------- + +class TestBackfillMigration: + def test_lastfm_url_set_but_status_null_gets_matched(self, db): + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, lastfm_url) VALUES (?, ?, ?)", + ("a1", "Artist A", "https://last.fm/music/Artist%20A"), + ) + cur.execute( + "INSERT INTO artists (id, name, lastfm_url, lastfm_match_status) VALUES (?, ?, ?, ?)", + ("a2", "Artist B", "https://last.fm/music/B", "matched"), + ) + cur.execute( + "INSERT INTO artists (id, name) VALUES (?, ?)", + ("a3", "Artist C"), # no url, status stays NULL + ) + conn.commit() + + # Run backfill a second time (first already ran during __init__) + db._backfill_match_status_for_existing_ids(cur) + conn.commit() + + rows = cur.execute( + "SELECT name, lastfm_match_status FROM artists ORDER BY name" + ).fetchall() + + by_name = {r[0]: r[1] for r in rows} + assert by_name["Artist A"] == "matched" + assert by_name["Artist B"] == "matched" # untouched + assert by_name["Artist C"] is None # no id => no backfill + + def test_musicbrainz_release_id_on_albums(self, db): + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name) VALUES (?, ?)", + ("art1", "A"), + ) + cur.execute( + "INSERT INTO albums (id, artist_id, title, musicbrainz_release_id) " + "VALUES (?, ?, ?, ?)", + ("alb1", "art1", "Album X", "mb-release-uuid"), + ) + conn.commit() + + db._backfill_match_status_for_existing_ids(cur) + conn.commit() + + status = cur.execute( + "SELECT musicbrainz_match_status FROM albums WHERE title = 'Album X'" + ).fetchone()[0] + assert status == "matched" + + def test_musicbrainz_recording_id_on_tracks(self, db): + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute("INSERT INTO artists (id, name) VALUES ('art2', 'A')") + cur.execute( + "INSERT INTO albums (id, artist_id, title) VALUES (?, ?, 'Alb')", + ("alb2", "art2"), + ) + cur.execute( + "INSERT INTO tracks (id, artist_id, album_id, title, musicbrainz_recording_id) " + "VALUES (?, ?, ?, ?, ?)", + ("trk1", "art2", "alb2", "T1", "mb-rec-uuid"), + ) + conn.commit() + + db._backfill_match_status_for_existing_ids(cur) + conn.commit() + + status = cur.execute( + "SELECT musicbrainz_match_status FROM tracks WHERE title = 'T1'" + ).fetchone()[0] + assert status == "matched" + + def test_tidal_and_qobuz_ids_backfilled(self, db): + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, tidal_id, qobuz_id) VALUES (?, ?, ?, ?)", + ("art3", "A", "tidal123", "qobuz456"), + ) + conn.commit() + + db._backfill_match_status_for_existing_ids(cur) + conn.commit() + + row = cur.execute( + "SELECT tidal_match_status, qobuz_match_status FROM artists WHERE id = 'art3'" + ).fetchone() + assert tuple(row) == ("matched", "matched") + + def test_empty_string_id_is_not_backfilled(self, db): + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, tidal_id) VALUES (?, ?, ?)", + ("art4", "Empty", ""), + ) + conn.commit() + + db._backfill_match_status_for_existing_ids(cur) + conn.commit() + + status = cur.execute( + "SELECT tidal_match_status FROM artists WHERE id = 'art4'" + ).fetchone()[0] + assert status is None + + +# --------------------------------------------------------------------------- +# _get_existing_id column-mapping correctness +# --------------------------------------------------------------------------- + +class TestGetExistingIdColumnMapping: + def _insert_tree(self, db): + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, lastfm_url, musicbrainz_id) " + "VALUES (?, ?, ?, ?)", + ("art_x", "A", "https://last.fm/a", "mb-artist"), + ) + cur.execute( + "INSERT INTO albums (id, artist_id, title, lastfm_url, musicbrainz_release_id) " + "VALUES (?, ?, ?, ?, ?)", + ("alb_x", "art_x", "Album", "https://last.fm/album", "mb-release"), + ) + cur.execute( + "INSERT INTO tracks (id, artist_id, album_id, title, lastfm_url, musicbrainz_recording_id) " + "VALUES (?, ?, ?, ?, ?, ?)", + ("trk_x", "art_x", "alb_x", "Track", "https://last.fm/track", "mb-rec"), + ) + conn.commit() + return "art_x", "alb_x", "trk_x" + + def test_lastfm_worker_reads_lastfm_url_for_all_entity_types(self, db): + # Import lazily so test collection doesn't fail if config_manager is unavailable. + from core import lastfm_worker as lw + + artist_id, album_id, track_id = self._insert_tree(db) + + with patch.object(lw.LastFMWorker, "_init_client", return_value=None): + worker = lw.LastFMWorker(db) + assert worker._get_existing_id("artist", artist_id) == "https://last.fm/a" + assert worker._get_existing_id("album", album_id) == "https://last.fm/album" + assert worker._get_existing_id("track", track_id) == "https://last.fm/track" + + def test_musicbrainz_worker_reads_correct_column_per_entity(self, db): + from core import musicbrainz_worker as mbw + + artist_id, album_id, track_id = self._insert_tree(db) + + with patch.object(mbw, "MusicBrainzService", return_value=MagicMock()): + worker = mbw.MusicBrainzWorker(db) + assert worker._get_existing_id("artist", artist_id) == "mb-artist" + assert worker._get_existing_id("album", album_id) == "mb-release" + assert worker._get_existing_id("track", track_id) == "mb-rec" + + +# --------------------------------------------------------------------------- +# Worker _process_* short-circuit marks status='matched' +# --------------------------------------------------------------------------- + +def _read_status(db, table: str, column: str, row_id: int): + with db._get_connection() as conn: + row = conn.execute( + f"SELECT {column} FROM {table} WHERE id = ?", (row_id,) + ).fetchone() + return row[0] if row else None + + +class TestLastFMWorkerMarksMatched: + def test_existing_url_triggers_matched_status(self, db): + from core import lastfm_worker as lw + + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, lastfm_url) VALUES (?, ?, ?)", + ("art_lf", "A", "https://last.fm/a"), + ) + artist_id = "art_lf" + # Explicitly null out status to simulate legacy row + cur.execute( + "UPDATE artists SET lastfm_match_status = NULL WHERE id = ?", + (artist_id,), + ) + conn.commit() + + with patch.object(lw.LastFMWorker, "_init_client", return_value=None): + worker = lw.LastFMWorker(db) + worker.client = MagicMock() + worker._process_artist(artist_id, "A") + # Client must NOT be called because we short-circuited. + worker.client.get_artist_info.assert_not_called() + + assert _read_status(db, "artists", "lastfm_match_status", artist_id) == "matched" + + +class TestTidalWorkerMarksMatched: + def test_existing_tidal_id_triggers_matched_status(self, db): + from core import tidal_worker as tw + + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, tidal_id) VALUES (?, ?, ?)", + ("art_td", "A", "tidal-123"), + ) + artist_id = "art_td" + cur.execute( + "UPDATE artists SET tidal_match_status = NULL WHERE id = ?", + (artist_id,), + ) + conn.commit() + + fake_client = MagicMock() + worker = tw.TidalWorker(db, client=fake_client) + worker._process_artist(artist_id, "A") + + fake_client.search_artist.assert_not_called() + assert _read_status(db, "artists", "tidal_match_status", artist_id) == "matched" + + +class TestQobuzWorkerMarksMatched: + def test_existing_qobuz_id_triggers_matched_status(self, db): + from core import qobuz_worker as qw + + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, qobuz_id) VALUES (?, ?, ?)", + ("art_qz", "A", "qobuz-xyz"), + ) + artist_id = "art_qz" + cur.execute( + "UPDATE artists SET qobuz_match_status = NULL WHERE id = ?", + (artist_id,), + ) + conn.commit() + + fake_client = MagicMock() + worker = qw.QobuzWorker(db, client=fake_client) + worker._process_artist(artist_id, "A") + + fake_client.search_artist.assert_not_called() + assert _read_status(db, "artists", "qobuz_match_status", artist_id) == "matched" + + +class TestMusicBrainzWorkerMarksMatched: + def test_existing_mbid_triggers_matched_status_via_service(self, db): + from core import musicbrainz_worker as mbw + + with db._get_connection() as conn: + cur = conn.cursor() + cur.execute( + "INSERT INTO artists (id, name, musicbrainz_id) VALUES (?, ?, ?)", + ("art_mb", "A", "mb-uuid"), + ) + artist_id = "art_mb" + conn.commit() + + fake_service = MagicMock() + with patch.object(mbw, "MusicBrainzService", return_value=fake_service): + worker = mbw.MusicBrainzWorker(db) + # mb_service on the instance is the MagicMock + worker._process_item({"type": "artist", "id": artist_id, "name": "A"}) + + fake_service.update_artist_mbid.assert_called_once_with( + artist_id, "mb-uuid", "matched" + )