Merge pull request #413 from Nezreka/fix/async-wishlist-cleanup

Fix/async wishlist cleanup

core/wishlist/processing.py

@@ -311,7 +311,66 @@ def start_manual_wishlist_download_batch(
     category: str | None = None,
     force_download_all: bool = False,
 ) -> tuple[Dict[str, Any], int]:
-    """Prepare and submit a manual wishlist batch."""
+    """Submit a manual wishlist batch.
+
+    The batch entry is created synchronously so the frontend can start polling
+    status immediately. The slow library-cleanup pass and master-worker hand-off
+    run in the background, freeing the request handler from a 30s+ block on
+    per-track DB checks for large wishlists.
+    """
+    logger = runtime.logger
+
+    try:
+        batch_id = str(uuid.uuid4())
+        playlist_id = "wishlist"
+        playlist_name = "Wishlist"
+
+        with runtime.tasks_lock:
+            runtime.download_batches[batch_id] = {
+                'phase': 'analysis',
+                'playlist_id': playlist_id,
+                'playlist_name': playlist_name,
+                'queue': [],
+                'active_count': 0,
+                'max_concurrent': runtime.get_batch_max_concurrent(),
+                'queue_index': 0,
+                # analysis_total starts at 0; the bg job updates it after cleanup
+                # finishes and the real track count is known.
+                'analysis_total': 0,
+                'analysis_processed': 0,
+                'analysis_results': [],
+                'permanently_failed_tracks': [],
+                'cancelled_tracks': set(),
+                'force_download_all': True,
+                'profile_id': runtime.profile_id,
+            }
+
+        runtime.missing_download_executor.submit(
+            _prepare_and_run_manual_wishlist_batch,
+            runtime,
+            batch_id,
+            track_ids,
+            category,
+        )
+
+        return {"success": True, "batch_id": batch_id}, 200
+
+    except Exception as e:
+        logger.error(f"Error starting wishlist download process: {e}")
+        import traceback
+
+        traceback.print_exc()
+        return {"success": False, "error": str(e)}, 500
+
+
+def _prepare_and_run_manual_wishlist_batch(
+    runtime: WishlistManualDownloadRuntime,
+    batch_id: str,
+    track_ids,
+    category: str | None,
+) -> None:
+    """Background worker for the manual wishlist batch — does the slow cleanup
+    + sanitize + filter + master-worker hand-off off the request thread."""
     logger = runtime.logger
 
     try:
@@ -324,23 +383,21 @@ def start_manual_wishlist_download_batch(
         if duplicates_removed > 0:
             logger.warning(f"[Manual-Wishlist] Removed {duplicates_removed} duplicate tracks")
 
-        logger.info("[Manual-Wishlist] Checking wishlist against library for already-owned tracks...")
+        # NOTE: We deliberately do NOT call remove_tracks_already_in_library here.
-        cleanup_removed = remove_tracks_already_in_library(
+        # Wishlist tracks are already known-missing (force_download_all=True is set on
-            wishlist_service,
+        # the batch). The library check duplicates the work the master worker would
-            SimpleNamespace(get_all_profiles=lambda: [{"id": manual_profile_id}]),
+        # skip, and on large wishlists costs ~1s per track in serial DB lookups.
-            db,
+        # The standalone /api/wishlist/cleanup endpoint still runs that pass when
-            runtime.active_server,
+        # users explicitly ask for maintenance.
-            logger=logger,
-            skip_track_fn=lambda track: track.get('source_type') == 'enhance',
-            log_prefix="[Manual-Wishlist]",
-        )
-
-        if cleanup_removed > 0:
-            logger.info(f"[Manual-Wishlist] Cleaned up {cleanup_removed} already-owned tracks from wishlist")
 
         raw_wishlist_tracks = wishlist_service.get_wishlist_tracks_for_download(profile_id=manual_profile_id)
         if not raw_wishlist_tracks:
-            return {"success": False, "error": "No tracks in wishlist"}, 400
+            logger.warning("[Manual-Wishlist] No tracks in wishlist after cleanup — marking batch complete")
+            with runtime.tasks_lock:
+                if batch_id in runtime.download_batches:
+                    runtime.download_batches[batch_id]['phase'] = 'complete'
+                    runtime.download_batches[batch_id]['error'] = 'No tracks in wishlist'
+            return
 
         wishlist_tracks, duplicates_found = sanitize_and_dedupe_wishlist_tracks(raw_wishlist_tracks)
         if duplicates_found > 0:
@@ -372,41 +429,25 @@ def start_manual_wishlist_download_batch(
         for i, track in enumerate(wishlist_tracks):
             track['_original_index'] = i
 
+        # Update batch with the real track count now that filtering is done
+        with runtime.tasks_lock:
+            if batch_id in runtime.download_batches:
+                runtime.download_batches[batch_id]['analysis_total'] = len(wishlist_tracks)
+
         runtime.add_activity_item("", "Wishlist Download Started", f"{len(wishlist_tracks)} tracks", "Now")
 
-        batch_id = str(uuid.uuid4())
-        playlist_id = "wishlist"
-        playlist_name = "Wishlist"
-        task_queue = []
-        with runtime.tasks_lock:
-            runtime.download_batches[batch_id] = {
-                'phase': 'analysis',
-                'playlist_id': playlist_id,
-                'playlist_name': playlist_name,
-                'queue': task_queue,
-                'active_count': 0,
-                'max_concurrent': runtime.get_batch_max_concurrent(),
-                'queue_index': 0,
-                'analysis_total': len(wishlist_tracks),
-                'analysis_processed': 0,
-                'analysis_results': [],
-                'permanently_failed_tracks': [],
-                'cancelled_tracks': set(),
-                'force_download_all': True,
-                'profile_id': manual_profile_id,
-            }
-
         logger.info(f"Starting wishlist batch {batch_id} with {len(wishlist_tracks)} tracks")
-        runtime.missing_download_executor.submit(runtime.run_full_missing_tracks_process, batch_id, playlist_id, wishlist_tracks)
+        runtime.run_full_missing_tracks_process(batch_id, "wishlist", wishlist_tracks)
 
-        return {"success": True, "batch_id": batch_id}, 200
+    except Exception as exc:
-
+        logger.error(f"Error preparing manual wishlist batch {batch_id}: {exc}")
-    except Exception as e:
-        logger.error(f"Error starting wishlist download process: {e}")
         import traceback
 
         traceback.print_exc()
-        return {"success": False, "error": str(e)}, 500
+        with runtime.tasks_lock:
+            if batch_id in runtime.download_batches:
+                runtime.download_batches[batch_id]['phase'] = 'error'
+                runtime.download_batches[batch_id]['error'] = str(exc)
 
 
 def cleanup_wishlist_against_library(
@@ -506,27 +547,19 @@ def process_wishlist_automatically(runtime: WishlistAutoProcessingRuntime, autom
                     if duplicates_removed > 0:
                         logger.warning(f"[Auto-Wishlist] Removed {duplicates_removed} duplicate tracks from profile {profile['id']}")
 
-                # CLEANUP: Remove tracks from wishlist that already exist in library
+                # NOTE: We deliberately do NOT call remove_tracks_already_in_library here.
-                # This prevents wasting bandwidth on tracks we already have
+                # The batch sets force_download_all=True (see comment a few lines below),
-                logger.debug("[Auto-Wishlist] Checking wishlist against library for already-owned tracks...")
+                # so wishlist tracks are treated as known-missing and the master worker
-                active_server = runtime.get_active_server()
+                # skips per-track library lookups. Doing the same expensive scan here
-                cleanup_removed = remove_tracks_already_in_library(
+                # before submitting the batch defeats that optimization and adds
-                    wishlist_service,
+                # ~1s per track in serial DB queries. The standalone
-                    database,
+                # /api/wishlist/cleanup endpoint still exposes that pass for users
-                    music_database,
+                # who want explicit maintenance.
-                    active_server,
+                runtime.update_automation_progress(automation_id, progress=25, phase='Preparing wishlist',
-                    logger=logger,
+                                                   log_line='Skipped library scan — wishlist tracks treated as known-missing',
-                )
+                                                   log_type='info')
 
-                if cleanup_removed > 0:
+                # Get wishlist tracks for processing - combine all profiles
-                    logger.info(f"[Auto-Wishlist] Cleaned up {cleanup_removed} already-owned tracks from wishlist")
-                    runtime.update_automation_progress(automation_id, progress=25, phase='Cleaned up duplicates',
-                                                       log_line=f'Removed {cleanup_removed} already-owned tracks', log_type='success')
-                else:
-                    runtime.update_automation_progress(automation_id, progress=25, phase='Cleanup done',
-                                                       log_line='No duplicates or already-owned tracks found', log_type='skip')
-
-                # Get wishlist tracks for processing (after cleanup) - combine all profiles
                 raw_wishlist_tracks = []
                 for profile in all_profiles:
                     raw_wishlist_tracks.extend(wishlist_service.get_wishlist_tracks_for_download(profile_id=profile['id']))

tests/wishlist/test_manual_download.py

@@ -85,6 +85,7 @@ def _build_runtime(tracks, owned_matches=None, batch_map=None):
     executor = _FakeExecutor()
     logger = _FakeLogger()
     activity_calls = []
+    master_calls = []
     batch_map = batch_map or {}
 
     runtime = WishlistManualDownloadRuntime(
@@ -92,18 +93,57 @@ def _build_runtime(tracks, owned_matches=None, batch_map=None):
         download_batches=batch_map,
         tasks_lock=_FakeLock(),
         missing_download_executor=executor,
-        run_full_missing_tracks_process=lambda *args, **kwargs: None,
+        run_full_missing_tracks_process=lambda *args, **kwargs: master_calls.append((args, kwargs)),
         get_batch_max_concurrent=lambda: 4,
         add_activity_item=lambda *args: activity_calls.append(args),
         active_server="navidrome",
         logger=logger,
         profile_id=1,
     )
-    return runtime, wishlist_service, music_db, executor, logger, activity_calls, batch_map
+    return runtime, wishlist_service, music_db, executor, logger, activity_calls, batch_map, master_calls
+
+
+def _run_submitted_bg_job(executor):
+    """Execute the bg job the executor received — simulates ThreadPoolExecutor."""
+    assert len(executor.submissions) == 1, "expected exactly one bg submission"
+    fn, args, kwargs = executor.submissions[0]
+    fn(*args, **kwargs)
+
+
+def test_start_manual_wishlist_download_batch_returns_immediately_with_placeholder():
+    """Endpoint returns 200 immediately; cleanup runs in the bg job."""
+    runtime, service, _db, executor, _logger, activity_calls, batch_map, master_calls = _build_runtime(
+        tracks=[
+            {
+                "id": "track-1",
+                "name": "Song 1",
+                "artists": [{"name": "Artist 1"}],
+                "album": {"name": "Album 1", "album_type": "album"},
+            },
+        ]
+    )
+
+    payload, status = processing.start_manual_wishlist_download_batch(runtime)
+
+    # Synchronous response: 200 with batch_id, batch entry created with placeholder count.
+    assert status == 200
+    assert payload["success"] is True
+    assert "batch_id" in payload
+    assert batch_map[payload["batch_id"]]["analysis_total"] == 0  # placeholder
+    assert batch_map[payload["batch_id"]]["phase"] == "analysis"
+    assert batch_map[payload["batch_id"]]["force_download_all"] is True
+
+    # Cleanup has NOT yet run (no DB calls, no master worker invocation).
+    assert service.removed_ids == set()
+    assert master_calls == []
+    assert activity_calls == []
+
+    # The bg job is queued.
+    assert len(executor.submissions) == 1
 
 
 def test_start_manual_wishlist_download_batch_filters_track_ids_and_starts_batch():
-    runtime, _service, _db, executor, logger, activity_calls, batch_map = _build_runtime(
+    runtime, _service, _db, executor, logger, activity_calls, batch_map, master_calls = _build_runtime(
         tracks=[
             {
                 "id": "track-1",
@@ -129,19 +169,31 @@ def test_start_manual_wishlist_download_batch_filters_track_ids_and_starts_batch
     assert status == 200
     assert payload["success"] is True
     assert "batch_id" in payload
+
+    # Run the bg job that the executor would have run on a real thread.
+    _run_submitted_bg_job(executor)
+
     assert activity_calls == [("", "Wishlist Download Started", "1 tracks", "Now")]
-    assert len(executor.submissions) == 1
+    assert len(master_calls) == 1
-    _submitted_fn, submitted_args, _submitted_kwargs = executor.submissions[0]
+    master_args, _ = master_calls[0]
-    assert submitted_args[1] == "wishlist"
+    assert master_args[1] == "wishlist"
-    assert submitted_args[2][0]["id"] == "track-2"
+    assert master_args[2][0]["id"] == "track-2"
-    assert submitted_args[2][0]["_original_index"] == 0
+    assert master_args[2][0]["_original_index"] == 0
     assert batch_map[payload["batch_id"]]["analysis_total"] == 1
     assert batch_map[payload["batch_id"]]["force_download_all"] is True
     assert any("Filtered to 1 specific tracks by ID" in msg for msg in logger.info_messages)
 
 
-def test_start_manual_wishlist_download_batch_skips_enhance_tracks_during_cleanup():
+def test_start_manual_wishlist_download_batch_does_not_run_library_cleanup():
-    runtime, service, _db, executor, logger, activity_calls, batch_map = _build_runtime(
+    """Manual flow does NOT scan the library for already-owned tracks.
+
+    The batch sets force_download_all=True so owned tracks get downloaded
+    anyway. Running remove_tracks_already_in_library here would just add a
+    serial DB query per track (~30s on a 24-track wishlist) and contradict
+    force_download_all. The standalone /api/wishlist/cleanup endpoint
+    still exposes that pass for users who want explicit maintenance.
+    """
+    runtime, service, db, executor, _logger, activity_calls, batch_map, master_calls = _build_runtime(
         tracks=[
             {
                 "id": "enhance-1",
@@ -164,10 +216,34 @@ def test_start_manual_wishlist_download_batch_skips_enhance_tracks_during_cleanu
 
     assert status == 200
     assert payload["success"] is True
-    assert service.removed_ids == {"owned-1"}
+
-    assert len(executor.submissions) == 1
+    _run_submitted_bg_job(executor)
-    _submitted_fn, submitted_args, _submitted_kwargs = executor.submissions[0]
+
-    assert [track["id"] for track in submitted_args[2]] == ["enhance-1"]
+    # Owned-track removal pass does NOT run — wishlist still has the owned track.
-    assert batch_map[payload["batch_id"]]["analysis_total"] == 1
+    assert service.removed_ids == set()
-    assert activity_calls == [("", "Wishlist Download Started", "1 tracks", "Now")]
+    # The library check is skipped entirely — no per-track DB lookups.
-    assert any("Cleaned up 1 already-owned tracks" in msg for msg in logger.info_messages)
+    assert db.track_checks == []
+
+    # All tracks are submitted to the master worker — including the "owned" one.
+    assert len(master_calls) == 1
+    master_args, _ = master_calls[0]
+    assert [track["id"] for track in master_args[2]] == ["enhance-1", "owned-1"]
+    assert batch_map[payload["batch_id"]]["analysis_total"] == 2
+    assert activity_calls == [("", "Wishlist Download Started", "2 tracks", "Now")]
+
+
+def test_bg_job_marks_batch_complete_when_wishlist_genuinely_empty():
+    """If the wishlist is empty before the manual click, the bg job marks the batch complete."""
+    runtime, _service, _db, executor, _logger, _activity, batch_map, master_calls = _build_runtime(
+        tracks=[],
+    )
+
+    payload, status = processing.start_manual_wishlist_download_batch(runtime)
+    assert status == 200
+
+    _run_submitted_bg_job(executor)
+
+    # No tracks → master worker never called, batch marked complete with explanatory error.
+    assert master_calls == []
+    assert batch_map[payload["batch_id"]]["phase"] == "complete"
+    assert batch_map[payload["batch_id"]]["error"] == "No tracks in wishlist"

web_server.py

@@ -17136,182 +17136,6 @@ def _execute_retag(group_id, album_id):
                 "phase": "Error",
                 "error_message": str(e)
             })
-def _check_and_remove_from_wishlist(context):
-    """
-    Check if a successfully downloaded track should be removed from wishlist.
-    Extracts Spotify track data from download context and removes from wishlist if found.
-    """
-    try:
-        from core.wishlist_service import get_wishlist_service
-        from core.imports.context import get_import_source, get_import_source_ids
-        wishlist_service = get_wishlist_service()
-        
-        # Try to extract a source-aware track ID from the context
-        spotify_track_id = None
-        # Populated lazily by Method 3 or Method 4. Initialized here so Method 4's
-        # `if not wishlist_tracks` guard doesn't UnboundLocalError when Methods 1/2
-        # found nothing and Method 3 never ran (no wishlist_id in track_info).
-        wishlist_tracks = []
-
-        # Method 1: Source-specific track lookup from track_info / source_ids
-        track_info = context.get('track_info', {})
-        source = get_import_source(context)
-        source_ids = get_import_source_ids(context)
-        source_label = {
-            'spotify': 'Spotify',
-            'itunes': 'iTunes',
-            'deezer': 'Deezer',
-            'discogs': 'Discogs',
-            'hydrabase': 'Hydrabase',
-        }.get(source, 'Source')
-
-        if source == 'spotify' and source_ids.get('track_id'):
-            spotify_track_id = source_ids['track_id']
-            logger.info(f"[Wishlist] Found {source_label} track ID from source_ids: {spotify_track_id}")
-        
-        # Method 2: Fallback to the original search result for source-specific IDs
-        elif source == 'spotify' and context.get('original_search_result', {}).get('id'):
-            spotify_track_id = context['original_search_result']['id']
-            logger.info(f"[Wishlist] Found {source_label} track ID from original_search_result: {spotify_track_id}")
-        
-        # Method 3: Check if this is a wishlist download (context has wishlist_id)
-        elif 'wishlist_id' in track_info:
-            wishlist_id = track_info['wishlist_id']
-            logger.info(f"[Wishlist] Found wishlist_id in context: {wishlist_id}")
-            
-            # Get the track ID from the wishlist entry (search all profiles)
-            database = get_database()
-            all_profiles = database.get_all_profiles()
-            wishlist_tracks = []
-            for p in all_profiles:
-                wishlist_tracks.extend(wishlist_service.get_wishlist_tracks_for_download(profile_id=p['id']))
-            for wl_track in wishlist_tracks:
-                if wl_track.get('wishlist_id') == wishlist_id:
-                    spotify_track_id = wl_track.get('spotify_track_id') or wl_track.get('id')
-                    logger.info(f"[Wishlist] Found track ID from wishlist entry: {spotify_track_id}")
-                    break
-
-        # Method 4: Try to construct a track ID from metadata for fuzzy matching
-        if not spotify_track_id:
-            track_name = track_info.get('name') or context.get('original_search_result', {}).get('title', '')
-            artist_name = _get_track_artist_name(track_info) or _get_track_artist_name(context.get('original_search_result', {}))
-
-            if track_name and artist_name:
-                logger.warning(f"[Wishlist] No track ID found, checking for fuzzy match: '{track_name}' by '{artist_name}'")
-
-                # Get all wishlist tracks and find potential matches (search all profiles)
-                if not wishlist_tracks:
-                    database = get_database()
-                    all_profiles = database.get_all_profiles()
-                    wishlist_tracks = []
-                    for p in all_profiles:
-                        wishlist_tracks.extend(wishlist_service.get_wishlist_tracks_for_download(profile_id=p['id']))
-                for wl_track in wishlist_tracks:
-                    wl_name = wl_track.get('name', '').lower()
-                    wl_artists = wl_track.get('artists', [])
-                    wl_artist_name = ''
-                    
-                    # Extract artist name from wishlist track
-                    if wl_artists:
-                        if isinstance(wl_artists[0], dict):
-                            wl_artist_name = wl_artists[0].get('name', '').lower()
-                        else:
-                            wl_artist_name = str(wl_artists[0]).lower()
-                    
-                    # Simple fuzzy matching
-                    if (wl_name == track_name.lower() and wl_artist_name == artist_name.lower()):
-                        spotify_track_id = wl_track.get('spotify_track_id') or wl_track.get('id')
-                        logger.info(f"[Wishlist] Found fuzzy match - track ID: {spotify_track_id}")
-                        break
-        
-        # If we found a track ID, remove it from wishlist
-        if spotify_track_id:
-            logger.info(f"[Wishlist] Attempting to remove track from wishlist: {spotify_track_id}")
-            removed = wishlist_service.mark_track_download_result(spotify_track_id, success=True)
-            if removed:
-                logger.info(f"[Wishlist] Successfully removed track from wishlist: {spotify_track_id}")
-            else:
-                logger.warning(f"ℹ️ [Wishlist] Track not found in wishlist or already removed: {spotify_track_id}")
-        else:
-            logger.warning("ℹ️ [Wishlist] No track ID found for wishlist removal check")
-            
-    except Exception as e:
-        logger.error(f"[Wishlist] Error in wishlist removal check: {e}")
-        import traceback
-        traceback.print_exc()
-
-def _check_and_remove_track_from_wishlist_by_metadata(track_data):
-    """
-    Check if a track found during database analysis should be removed from wishlist.
-    Uses track metadata (name, artists, id) to find and remove from wishlist.
-    """
-    try:
-        from core.wishlist_service import get_wishlist_service
-        wishlist_service = get_wishlist_service()
-        
-        # Extract track info
-        track_name = track_data.get('name', '')
-        track_id = track_data.get('id', '')
-        artists = track_data.get('artists', [])
-        
-        logger.info(f"[Analysis] Checking if track should be removed from wishlist: '{track_name}' (ID: {track_id})")
-        
-        # Method 1: Direct Spotify ID match
-        if track_id:
-            removed = wishlist_service.mark_track_download_result(track_id, success=True)
-            if removed:
-                logger.info(f"[Analysis] Removed track from wishlist via direct ID match: {track_id}")
-                return True
-        
-        # Method 2: Fuzzy matching by name and artist if no direct ID match
-        if track_name and artists:
-            # Extract primary artist name
-            primary_artist = ''
-            if isinstance(artists[0], dict) and 'name' in artists[0]:
-                primary_artist = artists[0]['name']
-            elif isinstance(artists[0], str):
-                primary_artist = artists[0]
-            else:
-                primary_artist = str(artists[0])
-            
-            logger.warning(f"[Analysis] No direct ID match, trying fuzzy match: '{track_name}' by '{primary_artist}'")
-
-            # Get all wishlist tracks and find matches (search all profiles)
-            database = get_database()
-            all_profiles = database.get_all_profiles()
-            wishlist_tracks = []
-            for p in all_profiles:
-                wishlist_tracks.extend(wishlist_service.get_wishlist_tracks_for_download(profile_id=p['id']))
-            for wl_track in wishlist_tracks:
-                wl_name = wl_track.get('name', '').lower()
-                wl_artists = wl_track.get('artists', [])
-                wl_artist_name = ''
-                
-                # Extract artist name from wishlist track
-                if wl_artists:
-                    if isinstance(wl_artists[0], dict):
-                        wl_artist_name = wl_artists[0].get('name', '').lower()
-                    else:
-                        wl_artist_name = str(wl_artists[0]).lower()
-                
-                # Fuzzy matching - normalize strings for comparison
-                if (wl_name == track_name.lower() and wl_artist_name == primary_artist.lower()):
-                    spotify_track_id = wl_track.get('spotify_track_id') or wl_track.get('id')
-                    if spotify_track_id:
-                        removed = wishlist_service.mark_track_download_result(spotify_track_id, success=True)
-                        if removed:
-                            logger.info(f"[Analysis] Removed track from wishlist via fuzzy match: {spotify_track_id}")
-                            return True
-        
-        logger.warning(f"ℹ️ [Analysis] Track not found in wishlist or already removed: '{track_name}'")
-        return False
-        
-    except Exception as e:
-        logger.error(f"[Analysis] Error checking wishlist removal by metadata: {e}")
-        import traceback
-        traceback.print_exc()
-        return False
-
 def _automatic_wishlist_cleanup_after_db_update():
     """Automatic wishlist cleanup that runs after database updates."""
     return _cleanup_wishlist_after_db_update(logger=logger)