Merge pull request #727 from Nezreka/refactor/discovery-endpoints-lift

Refactor/discovery endpoints lift

core/discovery/endpoints.py

@@ -0,0 +1,738 @@
+"""Generic, source-agnostic helpers for the playlist-discovery route layer.
+
+The discovery/sync endpoints in ``web_server.py`` were copy-pasted once per
+source (Tidal, Deezer, Qobuz, Spotify-public, iTunes-link, YouTube,
+ListenBrainz, Beatport). The per-source copies differ only by a source label
+string and which ``<source>_discovery_states`` global they read. This module
+lifts the source-agnostic pieces into importable, unit-testable helpers so the
+route functions become thin wrappers — exactly preserving behavior (1:1).
+
+Each helper is lifted verbatim from its web_server.py counterpart; any
+per-source quirk that genuinely differs (e.g. Beatport's distinct result
+shape) is intentionally NOT routed through here and stays in its own function.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, List, Tuple
+
+from utils.logging_config import get_logger
+
+logger = get_logger("discovery.endpoints")
+
+
+def convert_results_to_spotify_tracks(
+    discovery_results: List[Dict[str, Any]],
+    source_label: str,
+) -> List[Dict[str, Any]]:
+    """Convert a source's discovery results into the Spotify-track dicts the
+    sync pipeline expects.
+
+    Lifted verbatim from the per-source ``convert_<source>_results_to_spotify_tracks``
+    functions (and the already-generic ``_convert_link_results_to_spotify_tracks``),
+    which were byte-identical apart from the ``source_label`` used in the log
+    line. Two input shapes are supported, matching the originals exactly:
+
+    - ``spotify_data`` (manual-fix shape): copied through, preserving optional
+      ``track_number`` / ``disc_number``.
+    - ``spotify_track`` + ``status_class == 'found'`` (auto-discovery shape):
+      rebuilt from the flat ``spotify_*`` fields.
+
+    Any result matching neither shape is skipped, identical to the originals.
+
+    NOTE: Beatport deliberately does NOT use this — its converter coerces
+    artist objects to strings and emits a different track shape (``source``
+    field, album dict), so it keeps its own implementation.
+    """
+    spotify_tracks: List[Dict[str, Any]] = []
+
+    for result in discovery_results:
+        # Support both data formats: spotify_data (manual fixes) and individual
+        # fields (automatic discovery).
+        if result.get('spotify_data'):
+            spotify_data = result['spotify_data']
+            track = {
+                'id': spotify_data['id'],
+                'name': spotify_data['name'],
+                'artists': spotify_data['artists'],
+                'album': spotify_data['album'],
+                'duration_ms': spotify_data.get('duration_ms', 0),
+            }
+            if spotify_data.get('track_number'):
+                track['track_number'] = spotify_data['track_number']
+            if spotify_data.get('disc_number'):
+                track['disc_number'] = spotify_data['disc_number']
+            spotify_tracks.append(track)
+        elif result.get('spotify_track') and result.get('status_class') == 'found':
+            spotify_tracks.append({
+                'id': result.get('spotify_id', 'unknown'),
+                'name': result.get('spotify_track', 'Unknown Track'),
+                'artists': [result.get('spotify_artist', 'Unknown Artist')] if result.get('spotify_artist') else ['Unknown Artist'],
+                'album': result.get('spotify_album', 'Unknown Album'),
+                'duration_ms': 0,
+            })
+
+    logger.info(f"Converted {len(spotify_tracks)} {source_label} matches to Spotify tracks for sync")
+    return spotify_tracks
+
+
+def cancel_sync(
+    states: Dict[str, Any],
+    key: str,
+    *,
+    label: str,
+    not_found_message: str,
+    sync_lock: Any,
+    sync_states: Dict[str, Any],
+    active_sync_workers: Dict[str, Any],
+) -> Tuple[Dict[str, Any], int]:
+    """Cancel an in-progress sync for one discovery playlist.
+
+    1:1 lift of the byte-identical ``cancel_<source>_sync`` bodies (Tidal,
+    Deezer, Qobuz, Spotify-Public, iTunes-Link, YouTube, ListenBrainz). The
+    caller passes the already-resolved state key (ListenBrainz transforms it
+    via ``_lb_state_key`` first), the source ``label``, the exact 404 message
+    (iTunes-Link uses "iTunes Link not found", not "... playlist not found"),
+    and the shared sync infrastructure (so this stays free of web_server
+    globals / Flask).
+
+    Returns ``(payload_dict, status_code)``; the caller wraps in ``jsonify``.
+
+    Beatport is NOT routed here — it cancels a stored ``sync_future`` and
+    returns a different payload.
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        state = states[key]
+        state['last_accessed'] = time.time()
+        sync_playlist_id = state.get('sync_playlist_id')
+
+        if sync_playlist_id:
+            with sync_lock:
+                sync_states[sync_playlist_id] = {"status": "cancelled"}
+            if sync_playlist_id in active_sync_workers:
+                del active_sync_workers[sync_playlist_id]
+
+        state['phase'] = 'discovered'
+        state['sync_playlist_id'] = None
+        state['sync_progress'] = {}
+
+        return {"success": True, "message": f"{label} sync cancelled"}, 200
+    except Exception as e:
+        logger.error(f"Error cancelling {label} sync: {e}")
+        return {"error": str(e)}, 500
+
+
+def delete_playlist_state(
+    states: Dict[str, Any],
+    key: str,
+    *,
+    label: str,
+    not_found_message: str,
+) -> Tuple[Dict[str, Any], int]:
+    """Delete a discovery playlist's state entry, cancelling any active
+    discovery first.
+
+    1:1 lift of the byte-identical ``delete_<source>_playlist`` bodies
+    (Tidal, Deezer, Qobuz, Spotify-Public). Returns ``(payload, status_code)``.
+
+    The iTunes-Link / YouTube / ListenBrainz / Beatport deletes intentionally
+    keep their own bodies — they differ in success message, info-log wording,
+    name extraction, and/or key transform.
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        state = states[key]
+        if 'discovery_future' in state and state['discovery_future']:
+            state['discovery_future'].cancel()
+        del states[key]
+
+        logger.info(f"Deleted {label} playlist state: {key}")
+        return {"success": True, "message": "Playlist deleted"}, 200
+    except Exception as e:
+        logger.error(f"Error deleting {label} playlist: {e}")
+        return {"error": str(e)}, 500
+
+
+# --- playlist-name accessors -------------------------------------------------
+# The per-source sync-status handlers read the display name three different
+# ways. Each is reproduced verbatim so the 1:1 behavior (including which ones
+# raise vs. fall back to 'Unknown Playlist') is preserved.
+
+def playlist_name_attr_or_unknown(state: Dict[str, Any]) -> str:
+    """Tidal: playlist is an object — use ``.name`` or 'Unknown Playlist'."""
+    pl = state.get('playlist')
+    return pl.name if pl and hasattr(pl, 'name') else 'Unknown Playlist'
+
+
+def playlist_name_strict(state: Dict[str, Any]) -> str:
+    """Deezer / Qobuz / Spotify-Public / iTunes-Link: strict dict access —
+    raises (→ 500) if 'playlist' is missing, exactly like the originals."""
+    return state['playlist']['name']
+
+
+def playlist_name_safe(state: Dict[str, Any]) -> str:
+    """YouTube / ListenBrainz: safe dict access, defaulting to 'Unknown
+    Playlist'."""
+    return state.get('playlist', {}).get('name', 'Unknown Playlist')
+
+
+def playlist_name_obj(state: Dict[str, Any]) -> str:
+    """Tidal start-sync: playlist is an object — strict ``.name`` (raises if
+    absent, exactly like the original)."""
+    return state['playlist'].name
+
+
+def playlist_image_obj(state: Dict[str, Any]) -> str:
+    """Tidal: ``getattr(playlist, 'image_url', '')`` (object attribute)."""
+    return getattr(state['playlist'], 'image_url', '')
+
+
+def playlist_image_dict(state: Dict[str, Any]) -> str:
+    """Deezer/Qobuz/Spotify-Public/YouTube: ``playlist.get('image_url', '')``
+    (dict access)."""
+    return state['playlist'].get('image_url', '')
+
+
+def get_sync_status(
+    states: Dict[str, Any],
+    key: str,
+    *,
+    not_found_message: str,
+    error_label: str,
+    activity_subject: str,
+    playlist_name_getter,
+    sync_lock: Any,
+    sync_states: Dict[str, Any],
+    add_activity_item,
+) -> Tuple[Dict[str, Any], int]:
+    """Report sync status for one discovery playlist, posting an activity-feed
+    item when the sync finishes or errors.
+
+    1:1 lift of the ``get_<source>_sync_status`` bodies (Tidal, Deezer, Qobuz,
+    Spotify-Public, iTunes-Link, YouTube, ListenBrainz). Per-source variation
+    is captured by the parameters:
+
+    - ``not_found_message`` — the 404 string (iTunes-Link drops "playlist").
+    - ``error_label`` — used in the except log ("Error getting <X> sync status").
+    - ``activity_subject`` — the activity-feed prefix; note Spotify-Public uses
+      "Spotify Link playlist" while its error_label is "Spotify Public".
+    - ``playlist_name_getter`` — one of the accessors above (attr/strict/safe);
+      the strict one can raise, matching the originals (→ 500). The state's
+      phase/sync_progress are mutated BEFORE the name is read, so a raising
+      getter leaves the same partial mutation the original did.
+
+    Beatport is NOT routed here — it returns a different payload (``status``
+    not ``sync_status``, includes ``sync_id``, no lock, ``chart`` key).
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        state = states[key]
+        state['last_accessed'] = time.time()
+        sync_playlist_id = state.get('sync_playlist_id')
+
+        if not sync_playlist_id:
+            return {"error": "No sync in progress"}, 404
+
+        with sync_lock:
+            sync_state = sync_states.get(sync_playlist_id, {})
+
+        response = {
+            'phase': state['phase'],
+            'sync_status': sync_state.get('status', 'unknown'),
+            'progress': sync_state.get('progress', {}),
+            'complete': sync_state.get('status') == 'finished',
+            'error': sync_state.get('error'),
+        }
+
+        if sync_state.get('status') == 'finished':
+            state['phase'] = 'sync_complete'
+            state['sync_progress'] = sync_state.get('progress', {})
+            playlist_name = playlist_name_getter(state)
+            add_activity_item("", "Sync Complete", f"{activity_subject} '{playlist_name}' synced successfully", "Now")
+        elif sync_state.get('status') == 'error':
+            state['phase'] = 'discovered'
+            playlist_name = playlist_name_getter(state)
+            add_activity_item("", "Sync Failed", f"{activity_subject} '{playlist_name}' sync failed", "Now")
+
+        return response, 200
+    except Exception as e:
+        logger.error(f"Error getting {error_label} sync status: {e}")
+        return {"error": str(e)}, 500
+
+
+def get_discovery_status(
+    states: Dict[str, Any],
+    key: str,
+    *,
+    not_found_message: str,
+    error_label: str,
+) -> Tuple[Dict[str, Any], int]:
+    """Report real-time discovery progress/results for one playlist.
+
+    1:1 lift of the byte-identical ``get_<source>_discovery_status`` bodies.
+    Unlike sync-status, this shape is identical for ALL eight sources —
+    Beatport included — so it folds in too. Only the 404 message
+    (".../discovery not found" vs ".../playlist not found" vs "Beatport chart
+    not found") and the except-log label vary, both passed in. The caller
+    resolves the key (ListenBrainz via ``_lb_state_key``).
+
+    Returns ``(payload, status_code)``.
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        state = states[key]
+        state['last_accessed'] = time.time()
+
+        return {
+            'phase': state['phase'],
+            'status': state['status'],
+            'progress': state['discovery_progress'],
+            'spotify_matches': state['spotify_matches'],
+            'spotify_total': state['spotify_total'],
+            'results': state['discovery_results'],
+            'complete': state['phase'] == 'discovered',
+        }, 200
+    except Exception as e:
+        logger.error(f"Error getting {error_label} discovery status: {e}")
+        return {"error": str(e)}, 500
+
+
+def reset_playlist(
+    states: Dict[str, Any],
+    key: str,
+    *,
+    label: str,
+    not_found_message: str,
+) -> Tuple[Dict[str, Any], int]:
+    """Reset a discovery playlist back to the 'fresh' phase, clearing all
+    discovery/sync data while preserving the original playlist payload.
+
+    1:1 lift of the byte-identical ``reset_<source>_playlist`` bodies
+    (Tidal, Deezer, Qobuz, Spotify-Public). Returns ``(payload, status_code)``.
+
+    NOT folded in (genuinely divergent): YouTube (status -> 'parsed', no
+    download_process_id, logs the playlist name, "reset to fresh state"),
+    ListenBrainz (status -> 'cached', logs playlist title, returns
+    {"phase": "fresh"}), iTunes-Link (uses state.update, no info log, distinct
+    message). Those keep their own bodies.
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        state = states[key]
+        if 'discovery_future' in state and state['discovery_future']:
+            state['discovery_future'].cancel()
+
+        state['phase'] = 'fresh'
+        state['status'] = 'fresh'
+        state['discovery_results'] = []
+        state['discovery_progress'] = 0
+        state['spotify_matches'] = 0
+        state['sync_playlist_id'] = None
+        state['converted_spotify_playlist_id'] = None
+        state['download_process_id'] = None
+        state['sync_progress'] = {}
+        state['discovery_future'] = None
+        state['last_accessed'] = time.time()
+
+        logger.info(f"Reset {label} playlist to fresh: {key}")
+        return {"success": True, "message": "Playlist reset to fresh phase"}, 200
+    except Exception as e:
+        logger.error(f"Error resetting {label} playlist: {e}")
+        return {"error": str(e)}, 500
+
+
+def get_playlist_states(
+    states: Dict[str, Any],
+    *,
+    error_label: str,
+    info_log_label: str = None,
+) -> Tuple[Dict[str, Any], int]:
+    """Return all stored discovery states for a source as a list for frontend
+    card hydration (``{"states": [...]}``).
+
+    1:1 lift of the ``get_<source>_playlist_states`` bodies (Tidal, Deezer,
+    Qobuz, Spotify-Public, iTunes-Link), which build the same per-entry dict.
+    iTunes-Link is the only one without the "Returning N ..." info log, so
+    ``info_log_label`` is optional (pass None to suppress it, as iTunes did).
+
+    NOT folded in: the YouTube/ListenBrainz ``get_all_*_playlists`` endpoints —
+    they return ``{"playlists": [...]}`` (different key + fields: url/created_at,
+    no discovery_results) and filter mirrored/profile-scoped entries.
+    """
+    try:
+        result = []
+        current_time = time.time()
+
+        for key, state in states.items():
+            state['last_accessed'] = current_time
+            result.append({
+                'playlist_id': key,
+                'phase': state['phase'],
+                'status': state['status'],
+                'discovery_progress': state['discovery_progress'],
+                'spotify_matches': state['spotify_matches'],
+                'spotify_total': state['spotify_total'],
+                'discovery_results': state['discovery_results'],
+                'converted_spotify_playlist_id': state.get('converted_spotify_playlist_id'),
+                'download_process_id': state.get('download_process_id'),
+                'last_accessed': state['last_accessed'],
+            })
+
+        if info_log_label:
+            logger.info(f"Returning {len(result)} stored {info_log_label} playlist states for hydration")
+        return {"states": result}, 200
+    except Exception as e:
+        logger.error(f"Error getting {error_label} playlist states: {e}")
+        return {"error": str(e)}, 500
+
+
+def save_bubble_snapshot(
+    get_json,
+    *,
+    payload_key: str,
+    no_data_error: str,
+    snapshot_kind: str,
+    success_noun: str,
+    log_subject: str,
+    log_noun: str,
+    get_database,
+    get_current_profile_id,
+) -> Tuple[Dict[str, Any], int]:
+    """Persist a bubble/download snapshot for cross-refresh hydration.
+
+    1:1 lift of the four structurally-identical snapshot endpoints
+    (discover_downloads, artist_bubbles, search_bubbles, beatport_bubbles),
+    which differ only by:
+
+    - ``payload_key`` ('downloads' for discover, 'bubbles' for the rest) and
+      its ``no_data_error`` message.
+    - ``snapshot_kind`` — the db.save_bubble_snapshot category.
+    - ``success_noun`` — fills "Snapshot saved with N <noun>".
+    - ``log_subject`` / ``log_noun`` — the info ("Saved <subject>: N <noun>")
+      and except ("Error saving <subject>") log lines.
+
+    Returns ``(payload, status_code)``. ``get_json`` is invoked inside the try
+    like the original ``request.json``.
+    """
+    try:
+        from datetime import datetime
+
+        data = get_json()
+        if not data or payload_key not in data:
+            return {'success': False, 'error': no_data_error}, 400
+
+        items = data[payload_key]
+
+        db = get_database()
+        db.save_bubble_snapshot(snapshot_kind, items, profile_id=get_current_profile_id())
+
+        count = len(items)
+        logger.info(f"Saved {log_subject}: {count} {log_noun}")
+
+        return {
+            'success': True,
+            'message': f'Snapshot saved with {count} {success_noun}',
+            'timestamp': datetime.now().isoformat(),
+        }, 200
+    except Exception as e:
+        logger.error(f"Error saving {log_subject}: {e}")
+        import traceback
+        traceback.print_exc()
+        return {'success': False, 'error': str(e)}, 500
+
+
+def update_playlist_phase(
+    states: Dict[str, Any],
+    key: str,
+    get_json,
+    *,
+    not_found_message: str,
+    error_label: str,
+    valid_phases: List[str],
+    apply_extra_fields: bool,
+) -> Tuple[Dict[str, Any], int]:
+    """Update a discovery playlist's phase (used when the modal closes, e.g. to
+    reset download_complete -> discovered).
+
+    1:1 lift of the ``update_<source>_playlist_phase`` bodies for the five
+    sources with the identical validation + full-message response (Tidal,
+    Deezer, Qobuz, Spotify-Public, YouTube). Per-source params:
+
+    - ``valid_phases`` — YouTube's list additionally includes 'parsed'.
+    - ``apply_extra_fields`` — Deezer/Qobuz/Spotify-Public also persist
+      download_process_id / converted_spotify_playlist_id from the body;
+      Tidal/YouTube do NOT (so pass False to keep them 1:1).
+    - ``not_found_message`` / ``error_label``; ``get_json`` invoked inside the
+      try like the original ``request.get_json()``.
+
+    Returns ``(payload, status_code)``.
+
+    NOT folded in: iTunes-Link — it uses ``data.get('phase')`` (no separate
+    "Phase not provided" 400) and returns a no-message payload.
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        data = get_json()
+        if not data or 'phase' not in data:
+            return {"error": "Phase not provided"}, 400
+
+        new_phase = data['phase']
+        if new_phase not in valid_phases:
+            return {"error": f"Invalid phase. Must be one of: {', '.join(valid_phases)}"}, 400
+
+        state = states[key]
+        old_phase = state.get('phase', 'unknown')
+        state['phase'] = new_phase
+        state['last_accessed'] = time.time()
+
+        if apply_extra_fields:
+            if 'download_process_id' in data:
+                state['download_process_id'] = data['download_process_id']
+            if 'converted_spotify_playlist_id' in data:
+                state['converted_spotify_playlist_id'] = data['converted_spotify_playlist_id']
+
+        logger.info(f"Updated {error_label} playlist {key} phase: {old_phase} → {new_phase}")
+        return {"success": True, "message": f"Phase updated to {new_phase}", "old_phase": old_phase, "new_phase": new_phase}, 200
+    except Exception as e:
+        logger.error(f"Error updating {error_label} playlist phase: {e}")
+        return {"error": str(e)}, 500
+
+
+def first_artist_str_or_obj(original_track: Dict[str, Any]) -> str:
+    """Tidal: first artist from an artists list that may hold strings OR
+    objects ({'name': ...}); '' when empty."""
+    artists = original_track.get('artists', [])
+    if artists:
+        return artists[0] if isinstance(artists[0], str) else artists[0].get('name', '')
+    return ''
+
+
+def first_artist_plain(original_track: Dict[str, Any]) -> str:
+    """Deezer/Qobuz/Spotify-Public: first artist assuming a list of strings;
+    '' when empty."""
+    artists = original_track.get('artists', [])
+    return artists[0] if artists else ''
+
+
+def update_discovery_match(
+    states: Dict[str, Any],
+    get_json,
+    *,
+    source_log_label: str,
+    error_label: str,
+    original_track_key: str,
+    original_artist_getter,
+    join_artist_names,
+    extract_artist_name,
+    build_fix_modal_spotify_data,
+    get_discovery_cache_key,
+    get_database,
+    get_active_discovery_source,
+) -> Tuple[Dict[str, Any], int]:
+    """Apply a manually-selected Spotify track to a discovery result (the
+    fix-modal flow) and persist it to the discovery cache.
+
+    1:1 lift of the ``update_<source>_discovery_match`` bodies for the four
+    sources with the identical structure (Tidal, Deezer, Qobuz, Spotify-Public).
+    Per-source pieces are params:
+
+    - ``source_log_label`` (lowercase, e.g. "tidal") for the "Manual match
+      updated: ..." line; ``error_label`` for the except log.
+    - ``original_track_key`` — the raw-source track key on the result
+      ('tidal_track', 'deezer_track', ...).
+    - ``original_artist_getter`` — Tidal handles string-or-object artists
+      (``first_artist_str_or_obj``); the rest assume strings
+      (``first_artist_plain``).
+    - the web_server helpers (join/extract artist, build_fix_modal_spotify_data,
+      cache-key, get_database, active-discovery-source) are injected so this
+      stays free of those globals.
+    - ``get_json`` is called INSIDE the try (like the original's
+      ``request.get_json()``) so a malformed body yields the same 500.
+
+    Returns ``(payload, status_code)``.
+
+    NOT folded in: iTunes-Link (saves spotify_data directly via a different
+    cache signature), YouTube (multi-key original_track fallback), ListenBrainz
+    (entirely different unmatch-capable structure, no cache write), Beatport.
+    """
+    try:
+        data = get_json()
+        identifier = data.get('identifier')
+        track_index = data.get('track_index')
+        spotify_track = data.get('spotify_track')
+
+        if not identifier or track_index is None or not spotify_track:
+            return {'error': 'Missing required fields'}, 400
+
+        state = states.get(identifier)
+        if not state:
+            return {'error': 'Discovery state not found'}, 404
+
+        if track_index >= len(state['discovery_results']):
+            return {'error': 'Invalid track index'}, 400
+
+        result = state['discovery_results'][track_index]
+        old_status = result.get('status')
+
+        result['status'] = 'Found'
+        result['status_class'] = 'found'
+        result['spotify_track'] = spotify_track['name']
+        result['spotify_artist'] = join_artist_names(spotify_track['artists']) if isinstance(spotify_track['artists'], list) else extract_artist_name(spotify_track['artists'])
+        result['spotify_album'] = spotify_track['album']
+        result['spotify_id'] = spotify_track['id']
+
+        duration_ms = spotify_track.get('duration_ms', 0)
+        if duration_ms:
+            minutes = duration_ms // 60000
+            seconds = (duration_ms % 60000) // 1000
+            result['duration'] = f"{minutes}:{seconds:02d}"
+        else:
+            result['duration'] = '0:00'
+
+        result['spotify_data'] = build_fix_modal_spotify_data(spotify_track)
+        result['wing_it_fallback'] = False
+        result['manual_match'] = True
+
+        if old_status != 'found' and old_status != 'Found':
+            state['spotify_matches'] = state.get('spotify_matches', 0) + 1
+
+        logger.info(f"Manual match updated: {source_log_label} - {identifier} - track {track_index}")
+        logger.info(f"   → {result['spotify_artist']} - {result['spotify_track']}")
+
+        try:
+            original_track = result.get(original_track_key, {})
+            original_name = original_track.get('name', spotify_track['name'])
+            original_artist = original_artist_getter(original_track)
+
+            cache_key = get_discovery_cache_key(original_name, original_artist)
+            artists_list = spotify_track['artists']
+            if isinstance(artists_list, list):
+                artists_list = [a if isinstance(a, str) else a.get('name', '') for a in artists_list]
+            image_url = spotify_track.get('image_url') or ''
+            album_raw = spotify_track.get('album', '')
+            if isinstance(album_raw, dict):
+                album_obj = dict(album_raw)
+                if image_url and not album_obj.get('image_url'):
+                    album_obj['image_url'] = image_url
+                if image_url and not album_obj.get('images'):
+                    album_obj['images'] = [{'url': image_url}]
+            else:
+                album_obj = {'name': album_raw or ''}
+                if image_url:
+                    album_obj['image_url'] = image_url
+                    album_obj['images'] = [{'url': image_url}]
+
+            matched_data = {
+                'id': spotify_track['id'],
+                'name': spotify_track['name'],
+                'artists': artists_list,
+                'album': album_obj,
+                'duration_ms': spotify_track.get('duration_ms', 0),
+                'image_url': image_url,
+                'source': 'spotify',
+            }
+            cache_db = get_database()
+            cache_db.save_discovery_cache_match(
+                cache_key[0], cache_key[1], get_active_discovery_source(), 1.0, matched_data,
+                original_name, original_artist
+            )
+            logger.info(f"Manual fix saved to discovery cache: {original_name} by {original_artist}")
+        except Exception as cache_err:
+            logger.error(f"Error saving manual fix to discovery cache: {cache_err}")
+
+        return {'success': True, 'result': result}, 200
+    except Exception as e:
+        logger.error(f"Error updating {error_label} discovery match: {e}")
+        return {'error': str(e)}, 500
+
+
+def start_sync(
+    states: Dict[str, Any],
+    key: str,
+    *,
+    sync_id_prefix: str,
+    not_found_message: str,
+    not_ready_message: str,
+    convert_fn,
+    playlist_name_getter,
+    playlist_image_getter,
+    activity_label: str,
+    error_label: str,
+    sync_lock: Any,
+    sync_states: Dict[str, Any],
+    active_sync_workers: Dict[str, Any],
+    submit_sync_task,
+    add_activity_item,
+) -> Tuple[Dict[str, Any], int]:
+    """Kick off a playlist sync from a source's discovered Spotify matches.
+
+    1:1 lift of the ``start_<source>_sync`` bodies for the five sources with
+    the identical flow (Tidal, Deezer, Qobuz, Spotify-Public, YouTube). The
+    per-source pieces are parameters:
+
+    - ``sync_id_prefix`` — the ``f"{prefix}_{key}"`` sync id.
+    - ``convert_fn`` — the source's discovery->spotify-tracks converter.
+    - ``playlist_name_getter`` / ``playlist_image_getter`` — Tidal reads an
+      object (``.name`` / ``getattr``), the rest read a dict; lifted as the
+      ``playlist_name_obj``/``playlist_image_obj`` vs ``playlist_name_strict``/
+      ``playlist_image_dict`` accessors.
+    - ``activity_label`` vs ``error_label`` — these DIFFER for Spotify-Public:
+      activity says "Spotify Link Sync Started" while logs say "Spotify Public".
+    - ``submit_sync_task(sync_playlist_id, playlist_name, spotify_tracks,
+      playlist_image_url) -> Future`` — wraps sync_executor/_run_sync_task/
+      get_current_profile_id so this stays free of those globals.
+
+    Returns ``(payload, status_code)``.
+
+    NOT folded in: iTunes-Link (no final info log), ListenBrainz (submits the
+    task without an image arg), Beatport (extra debug logging, 'chart' key).
+    """
+    try:
+        if key not in states:
+            return {"error": not_found_message}, 404
+
+        state = states[key]
+        state['last_accessed'] = time.time()
+
+        if state['phase'] not in ['discovered', 'sync_complete', 'download_complete']:
+            return {"error": not_ready_message}, 400
+
+        spotify_tracks = convert_fn(state['discovery_results'])
+        if not spotify_tracks:
+            return {"error": "No Spotify matches found for sync"}, 400
+
+        sync_playlist_id = f"{sync_id_prefix}_{key}"
+        playlist_name = playlist_name_getter(state)
+
+        add_activity_item("", f"{activity_label} Sync Started", f"'{playlist_name}' - {len(spotify_tracks)} tracks", "Now")
+
+        state['phase'] = 'syncing'
+        state['sync_playlist_id'] = sync_playlist_id
+        state['sync_progress'] = {}
+
+        with sync_lock:
+            sync_states[sync_playlist_id] = {"status": "starting", "progress": {}}
+
+        playlist_image_url = playlist_image_getter(state)
+        future = submit_sync_task(sync_playlist_id, playlist_name, spotify_tracks, playlist_image_url)
+        active_sync_workers[sync_playlist_id] = future
+
+        logger.info(f"Started {error_label} sync for: {playlist_name} ({len(spotify_tracks)} tracks)")
+        return {"success": True, "sync_playlist_id": sync_playlist_id}, 200
+    except Exception as e:
+        logger.error(f"Error starting {error_label} sync: {e}")
+        return {"error": str(e)}, 500

tests/discovery/test_discovery_endpoints.py

@@ -0,0 +1,957 @@
+"""Tests for the lifted, source-agnostic discovery route helpers in
+``core.discovery.endpoints``.
+
+These pin the exact behavior the per-source ``convert_<source>_results_to_spotify_tracks``
+functions had in web_server.py, so the lift is provably 1:1. Each input shape
+the originals handled is exercised here.
+"""
+
+from __future__ import annotations
+
+import threading
+
+from core.discovery.endpoints import (
+    convert_results_to_spotify_tracks,
+    cancel_sync,
+    delete_playlist_state,
+    get_sync_status,
+    playlist_name_strict as _pl_name_strict,
+    playlist_name_safe as _pl_name_safe,
+)
+
+
+class _FakeFuture:
+    def __init__(self):
+        self.cancelled = False
+
+    def cancel(self):
+        self.cancelled = True
+
+
+def _cancel_infra():
+    """Fresh sync infra (lock + the two shared dicts) for cancel_sync tests."""
+    return {
+        'sync_lock': threading.Lock(),
+        'sync_states': {},
+        'active_sync_workers': {},
+    }
+
+
+# ---------------------------------------------------------------------------
+# spotify_data (manual-fix) shape
+# ---------------------------------------------------------------------------
+
+def test_spotify_data_shape_basic():
+    results = [{
+        'spotify_data': {
+            'id': 'sp1', 'name': 'Song', 'artists': ['A'], 'album': 'Alb',
+            'duration_ms': 1234,
+        }
+    }]
+    assert convert_results_to_spotify_tracks(results, 'Tidal') == [{
+        'id': 'sp1', 'name': 'Song', 'artists': ['A'], 'album': 'Alb',
+        'duration_ms': 1234,
+    }]
+
+
+def test_spotify_data_duration_defaults_to_zero():
+    results = [{'spotify_data': {'id': 'x', 'name': 'n', 'artists': [], 'album': 'a'}}]
+    out = convert_results_to_spotify_tracks(results, 'Deezer')
+    assert out[0]['duration_ms'] == 0
+
+
+def test_spotify_data_includes_track_and_disc_number_when_present():
+    results = [{'spotify_data': {
+        'id': 'x', 'name': 'n', 'artists': [], 'album': 'a',
+        'track_number': 5, 'disc_number': 2,
+    }}]
+    out = convert_results_to_spotify_tracks(results, 'Qobuz')
+    assert out[0]['track_number'] == 5
+    assert out[0]['disc_number'] == 2
+
+
+def test_spotify_data_omits_track_disc_when_absent_or_falsy():
+    # track_number/disc_number of 0 are falsy -> omitted, matching original.
+    results = [{'spotify_data': {
+        'id': 'x', 'name': 'n', 'artists': [], 'album': 'a',
+        'track_number': 0, 'disc_number': 0,
+    }}]
+    out = convert_results_to_spotify_tracks(results, 'YouTube')
+    assert 'track_number' not in out[0]
+    assert 'disc_number' not in out[0]
+
+
+# ---------------------------------------------------------------------------
+# spotify_track + status_class == 'found' (auto-discovery) shape
+# ---------------------------------------------------------------------------
+
+def test_auto_discovery_shape_full():
+    results = [{
+        'spotify_track': 'Track', 'status_class': 'found',
+        'spotify_id': 'id9', 'spotify_artist': 'Artist', 'spotify_album': 'Album',
+    }]
+    assert convert_results_to_spotify_tracks(results, 'ListenBrainz') == [{
+        'id': 'id9', 'name': 'Track', 'artists': ['Artist'], 'album': 'Album',
+        'duration_ms': 0,
+    }]
+
+
+def test_auto_discovery_defaults_when_fields_missing():
+    results = [{'spotify_track': 'T', 'status_class': 'found'}]
+    out = convert_results_to_spotify_tracks(results, 'Spotify Public')
+    assert out == [{
+        'id': 'unknown', 'name': 'T', 'artists': ['Unknown Artist'],
+        'album': 'Unknown Album', 'duration_ms': 0,
+    }]
+
+
+def test_auto_discovery_empty_artist_yields_unknown_artist():
+    results = [{
+        'spotify_track': 'T', 'status_class': 'found', 'spotify_artist': '',
+    }]
+    out = convert_results_to_spotify_tracks(results, 'Tidal')
+    assert out[0]['artists'] == ['Unknown Artist']
+
+
+# ---------------------------------------------------------------------------
+# skip / mixed / empty
+# ---------------------------------------------------------------------------
+
+def test_auto_discovery_requires_found_status():
+    # spotify_track present but status_class != 'found' -> skipped.
+    results = [{'spotify_track': 'T', 'status_class': 'not_found'}]
+    assert convert_results_to_spotify_tracks(results, 'Tidal') == []
+
+
+def test_result_matching_neither_shape_is_skipped():
+    results = [{'irrelevant': True}, {'spotify_track': 'T'}]  # 2nd has no status_class
+    assert convert_results_to_spotify_tracks(results, 'Tidal') == []
+
+
+def test_mixed_results_preserve_order():
+    results = [
+        {'spotify_data': {'id': '1', 'name': 'a', 'artists': [], 'album': ''}},
+        {'irrelevant': True},
+        {'spotify_track': 'b', 'status_class': 'found', 'spotify_id': '2'},
+    ]
+    out = convert_results_to_spotify_tracks(results, 'Tidal')
+    assert [t['id'] for t in out] == ['1', '2']
+
+
+def test_empty_input():
+    assert convert_results_to_spotify_tracks([], 'Tidal') == []
+
+
+def test_spotify_data_takes_precedence_over_auto_fields():
+    # A result carrying both shapes uses spotify_data (the if-branch wins),
+    # matching the original if/elif ordering.
+    results = [{
+        'spotify_data': {'id': 'D', 'name': 'd', 'artists': [], 'album': ''},
+        'spotify_track': 'IGNORED', 'status_class': 'found', 'spotify_id': 'A',
+    }]
+    out = convert_results_to_spotify_tracks(results, 'Tidal')
+    assert out[0]['id'] == 'D'
+
+
+# ---------------------------------------------------------------------------
+# cancel_sync
+# ---------------------------------------------------------------------------
+
+def test_cancel_sync_not_found_returns_404():
+    body, code = cancel_sync({}, 'missing', label='Tidal',
+                             not_found_message='Tidal playlist not found', **_cancel_infra())
+    assert code == 404
+    assert body == {"error": "Tidal playlist not found"}
+
+
+def test_cancel_sync_cancels_active_worker_and_reverts_state():
+    infra = _cancel_infra()
+    infra['sync_states']['sp1'] = {"status": "running"}
+    infra['active_sync_workers']['sp1'] = _FakeFuture()
+    states = {'pl': {'phase': 'syncing', 'sync_playlist_id': 'sp1',
+                     'sync_progress': {'done': 1}, 'last_accessed': 0}}
+
+    body, code = cancel_sync(states, 'pl', label='Tidal',
+                             not_found_message='nf', **infra)
+
+    assert code == 200
+    assert body == {"success": True, "message": "Tidal sync cancelled"}
+    # sync marked cancelled, worker removed
+    assert infra['sync_states']['sp1'] == {"status": "cancelled"}
+    assert 'sp1' not in infra['active_sync_workers']
+    # state reverted
+    assert states['pl']['phase'] == 'discovered'
+    assert states['pl']['sync_playlist_id'] is None
+    assert states['pl']['sync_progress'] == {}
+    assert states['pl']['last_accessed'] != 0  # touched
+
+
+def test_cancel_sync_worker_absent_from_active_map_is_safe():
+    infra = _cancel_infra()  # active_sync_workers empty
+    states = {'pl': {'phase': 'syncing', 'sync_playlist_id': 'sp1'}}
+    body, code = cancel_sync(states, 'pl', label='Deezer', not_found_message='nf', **infra)
+    assert code == 200
+    assert infra['sync_states']['sp1'] == {"status": "cancelled"}
+
+
+def test_cancel_sync_no_sync_in_progress_still_reverts():
+    infra = _cancel_infra()
+    states = {'pl': {'phase': 'discovered'}}  # no sync_playlist_id
+    body, code = cancel_sync(states, 'pl', label='Qobuz', not_found_message='nf', **infra)
+    assert code == 200
+    assert states['pl']['sync_playlist_id'] is None
+    assert states['pl']['sync_progress'] == {}
+    assert infra['sync_states'] == {}  # nothing cancelled
+
+
+def test_cancel_sync_label_in_message():
+    infra = _cancel_infra()
+    states = {'pl': {}}
+    body, _ = cancel_sync(states, 'pl', label='iTunes Link', not_found_message='nf', **infra)
+    assert body["message"] == "iTunes Link sync cancelled"
+
+
+def test_cancel_sync_exception_returns_500():
+    infra = _cancel_infra()
+    states = {'pl': object()}  # not subscriptable -> raises inside try
+    body, code = cancel_sync(states, 'pl', label='Tidal', not_found_message='nf', **infra)
+    assert code == 500
+    assert "error" in body
+
+
+# ---------------------------------------------------------------------------
+# delete_playlist_state
+# ---------------------------------------------------------------------------
+
+def test_delete_not_found_returns_404():
+    body, code = delete_playlist_state({}, 'missing', label='Tidal',
+                                       not_found_message='Tidal playlist not found')
+    assert code == 404
+    assert body == {"error": "Tidal playlist not found"}
+
+
+def test_delete_cancels_discovery_future_and_removes_state():
+    fut = _FakeFuture()
+    states = {'pl': {'discovery_future': fut}}
+    body, code = delete_playlist_state(states, 'pl', label='Tidal', not_found_message='nf')
+    assert code == 200
+    assert body == {"success": True, "message": "Playlist deleted"}
+    assert fut.cancelled is True
+    assert 'pl' not in states
+
+
+def test_delete_without_discovery_future_still_removes():
+    states = {'pl': {'phase': 'discovered'}}  # no discovery_future
+    body, code = delete_playlist_state(states, 'pl', label='Deezer', not_found_message='nf')
+    assert code == 200
+    assert 'pl' not in states
+
+
+def test_delete_falsy_discovery_future_not_cancelled():
+    states = {'pl': {'discovery_future': None}}
+    body, code = delete_playlist_state(states, 'pl', label='Qobuz', not_found_message='nf')
+    assert code == 200
+    assert 'pl' not in states
+
+
+def test_delete_exception_returns_500():
+    fut = object()  # no .cancel() -> AttributeError inside try
+    states = {'pl': {'discovery_future': fut}}
+    body, code = delete_playlist_state(states, 'pl', label='Tidal', not_found_message='nf')
+    assert code == 500
+    assert "error" in body
+    # state NOT deleted because the exception fired before del
+    assert 'pl' in states
+
+
+# ---------------------------------------------------------------------------
+# playlist-name accessors
+# ---------------------------------------------------------------------------
+
+class _Obj:
+    def __init__(self, name):
+        self.name = name
+
+
+def test_name_attr_or_unknown():
+    from core.discovery.endpoints import playlist_name_attr_or_unknown as g
+    assert g({'playlist': _Obj('My PL')}) == 'My PL'
+    assert g({'playlist': {'name': 'dict'}}) == 'Unknown Playlist'  # dict has no .name attr
+    assert g({}) == 'Unknown Playlist'
+    assert g({'playlist': None}) == 'Unknown Playlist'
+
+
+def test_name_strict():
+    from core.discovery.endpoints import playlist_name_strict as g
+    assert g({'playlist': {'name': 'X'}}) == 'X'
+    import pytest
+    with pytest.raises(KeyError):
+        g({})  # strict -> raises, matching originals
+
+
+def test_name_safe():
+    from core.discovery.endpoints import playlist_name_safe as g
+    assert g({'playlist': {'name': 'X'}}) == 'X'
+    assert g({}) == 'Unknown Playlist'
+    assert g({'playlist': {}}) == 'Unknown Playlist'
+
+
+# ---------------------------------------------------------------------------
+# get_sync_status
+# ---------------------------------------------------------------------------
+
+def _activity_recorder():
+    calls = []
+
+    def add_activity_item(user, action, desc, when):
+        calls.append((user, action, desc, when))
+
+    return calls, add_activity_item
+
+
+def _status_kwargs(infra, add_activity_item, *, not_found_message='nf',
+                   error_label='Tidal', activity_subject='Tidal playlist',
+                   name_getter=None):
+    return dict(
+        not_found_message=not_found_message, error_label=error_label,
+        activity_subject=activity_subject,
+        playlist_name_getter=name_getter or _pl_name_safe,
+        add_activity_item=add_activity_item,
+        sync_lock=infra['sync_lock'], sync_states=infra['sync_states'],
+    )
+
+
+def test_status_not_found():
+    infra = _cancel_infra()
+    calls, add = _activity_recorder()
+    body, code = get_sync_status({}, 'missing',
+                                 **_status_kwargs(infra, add, not_found_message='Tidal playlist not found'))
+    assert code == 404 and body == {"error": "Tidal playlist not found"}
+    assert calls == []
+
+
+def test_status_no_sync_in_progress():
+    infra = _cancel_infra()
+    calls, add = _activity_recorder()
+    states = {'pl': {'phase': 'discovered'}}  # no sync_playlist_id
+    body, code = get_sync_status(states, 'pl', **_status_kwargs(infra, add))
+    assert code == 404 and body == {"error": "No sync in progress"}
+
+
+def test_status_running_returns_shape_without_mutation_or_activity():
+    infra = _cancel_infra()
+    infra['sync_states']['sp1'] = {"status": "running", "progress": {"n": 2}}
+    calls, add = _activity_recorder()
+    states = {'pl': {'phase': 'syncing', 'sync_playlist_id': 'sp1'}}
+    body, code = get_sync_status(states, 'pl', **_status_kwargs(infra, add))
+    assert code == 200
+    assert body == {
+        'phase': 'syncing', 'sync_status': 'running',
+        'progress': {"n": 2}, 'complete': False, 'error': None,
+    }
+    assert states['pl']['phase'] == 'syncing'  # unchanged
+    assert calls == []
+
+
+def test_status_finished_sets_complete_and_posts_activity():
+    infra = _cancel_infra()
+    infra['sync_states']['sp1'] = {"status": "finished", "progress": {"done": 9}}
+    calls, add = _activity_recorder()
+    states = {'pl': {'phase': 'syncing', 'sync_playlist_id': 'sp1',
+                     'playlist': {'name': 'Mix'}}}
+    body, code = get_sync_status(states, 'pl', **_status_kwargs(
+        infra, add, activity_subject='Spotify Link playlist',
+        name_getter=_pl_name_strict))
+    assert code == 200
+    assert body['complete'] is True
+    assert states['pl']['phase'] == 'sync_complete'
+    assert states['pl']['sync_progress'] == {"done": 9}
+    assert calls == [("", "Sync Complete", "Spotify Link playlist 'Mix' synced successfully", "Now")]
+
+
+def test_status_error_reverts_and_posts_failed_activity():
+    infra = _cancel_infra()
+    infra['sync_states']['sp1'] = {"status": "error", "error": "boom"}
+    calls, add = _activity_recorder()
+    states = {'pl': {'phase': 'syncing', 'sync_playlist_id': 'sp1',
+                     'playlist': {'name': 'Mix'}}}
+    body, code = get_sync_status(states, 'pl', **_status_kwargs(
+        infra, add, activity_subject='YouTube playlist', name_getter=_pl_name_safe))
+    assert code == 200
+    assert body['error'] == "boom"
+    assert states['pl']['phase'] == 'discovered'
+    assert calls == [("", "Sync Failed", "YouTube playlist 'Mix' sync failed", "Now")]
+
+
+def test_status_strict_getter_missing_playlist_raises_500_after_partial_mutation():
+    infra = _cancel_infra()
+    infra['sync_states']['sp1'] = {"status": "finished", "progress": {}}
+    calls, add = _activity_recorder()
+    states = {'pl': {'phase': 'syncing', 'sync_playlist_id': 'sp1'}}  # no 'playlist'
+    body, code = get_sync_status(states, 'pl', **_status_kwargs(
+        infra, add, error_label='Deezer', name_getter=_pl_name_strict))
+    assert code == 500 and "error" in body
+    # phase was set to sync_complete BEFORE the strict getter raised (1:1).
+    assert states['pl']['phase'] == 'sync_complete'
+    assert calls == []  # activity never posted
+
+
+# ---------------------------------------------------------------------------
+# get_discovery_status
+# ---------------------------------------------------------------------------
+
+def test_discovery_status_not_found():
+    from core.discovery.endpoints import get_discovery_status
+    body, code = get_discovery_status({}, 'missing',
+                                      not_found_message='Tidal discovery not found',
+                                      error_label='Tidal')
+    assert code == 404 and body == {"error": "Tidal discovery not found"}
+
+
+def test_discovery_status_builds_response_and_bumps_access():
+    from core.discovery.endpoints import get_discovery_status
+    state = {
+        'phase': 'discovered', 'status': 'done', 'discovery_progress': 100,
+        'spotify_matches': 8, 'spotify_total': 10,
+        'discovery_results': [{'x': 1}], 'last_accessed': 0,
+    }
+    states = {'pl': state}
+    body, code = get_discovery_status(states, 'pl',
+                                      not_found_message='nf', error_label='Tidal')
+    assert code == 200
+    assert body == {
+        'phase': 'discovered', 'status': 'done', 'progress': 100,
+        'spotify_matches': 8, 'spotify_total': 10,
+        'results': [{'x': 1}], 'complete': True,
+    }
+    assert state['last_accessed'] != 0
+
+
+def test_discovery_status_complete_false_when_not_discovered():
+    from core.discovery.endpoints import get_discovery_status
+    state = {
+        'phase': 'discovering', 'status': 'running', 'discovery_progress': 40,
+        'spotify_matches': 2, 'spotify_total': 10, 'discovery_results': [],
+    }
+    body, code = get_discovery_status({'pl': state}, 'pl',
+                                      not_found_message='nf', error_label='Deezer')
+    assert code == 200 and body['complete'] is False
+
+
+def test_discovery_status_missing_field_raises_500():
+    from core.discovery.endpoints import get_discovery_status
+    # state missing 'status' -> strict access raises -> 500 (matches original)
+    states = {'pl': {'phase': 'x'}}
+    body, code = get_discovery_status(states, 'pl',
+                                      not_found_message='nf', error_label='Beatport')
+    assert code == 500 and "error" in body
+
+
+# ---------------------------------------------------------------------------
+# reset_playlist
+# ---------------------------------------------------------------------------
+
+def test_reset_not_found():
+    from core.discovery.endpoints import reset_playlist
+    body, code = reset_playlist({}, 'missing', label='Tidal',
+                                not_found_message='Tidal playlist not found')
+    assert code == 404 and body == {"error": "Tidal playlist not found"}
+
+
+def test_reset_clears_state_preserving_playlist_and_cancels_future():
+    from core.discovery.endpoints import reset_playlist
+    fut = _FakeFuture()
+    state = {
+        'playlist': {'name': 'keep me'},
+        'phase': 'discovered', 'status': 'done',
+        'discovery_results': [{'x': 1}], 'discovery_progress': 100,
+        'spotify_matches': 5, 'sync_playlist_id': 'sp', 'last_accessed': 0,
+        'converted_spotify_playlist_id': 'cv', 'download_process_id': 'dp',
+        'sync_progress': {'n': 1}, 'discovery_future': fut,
+    }
+    states = {'pl': state}
+    body, code = reset_playlist(states, 'pl', label='Tidal', not_found_message='nf')
+
+    assert code == 200
+    assert body == {"success": True, "message": "Playlist reset to fresh phase"}
+    assert fut.cancelled is True
+    # cleared
+    assert state['phase'] == 'fresh'
+    assert state['status'] == 'fresh'
+    assert state['discovery_results'] == []
+    assert state['discovery_progress'] == 0
+    assert state['spotify_matches'] == 0
+    assert state['sync_playlist_id'] is None
+    assert state['converted_spotify_playlist_id'] is None
+    assert state['download_process_id'] is None
+    assert state['sync_progress'] == {}
+    assert state['discovery_future'] is None
+    assert state['last_accessed'] != 0
+    # original playlist payload preserved
+    assert state['playlist'] == {'name': 'keep me'}
+
+
+def test_reset_without_discovery_future():
+    from core.discovery.endpoints import reset_playlist
+    state = {'phase': 'discovered'}  # no discovery_future key
+    body, code = reset_playlist({'pl': state}, 'pl', label='Deezer', not_found_message='nf')
+    assert code == 200
+    assert state['phase'] == 'fresh'
+
+
+def test_reset_exception_returns_500():
+    from core.discovery.endpoints import reset_playlist
+    fut = object()  # .cancel missing -> AttributeError in try
+    states = {'pl': {'discovery_future': fut}}
+    body, code = reset_playlist(states, 'pl', label='Qobuz', not_found_message='nf')
+    assert code == 500 and "error" in body
+
+
+# ---------------------------------------------------------------------------
+# get_playlist_states (bulk hydration list)
+# ---------------------------------------------------------------------------
+
+def test_playlist_states_builds_list_and_bumps_access():
+    from core.discovery.endpoints import get_playlist_states
+    s1 = {'phase': 'discovered', 'status': 'done', 'discovery_progress': 100,
+          'spotify_matches': 3, 'spotify_total': 5, 'discovery_results': [{'a': 1}],
+          'converted_spotify_playlist_id': 'cv', 'download_process_id': 'dp',
+          'last_accessed': 0}
+    states = {'k1': s1}
+    body, code = get_playlist_states(states, error_label='Tidal', info_log_label='Tidal')
+    assert code == 200
+    assert body == {"states": [{
+        'playlist_id': 'k1', 'phase': 'discovered', 'status': 'done',
+        'discovery_progress': 100, 'spotify_matches': 3, 'spotify_total': 5,
+        'discovery_results': [{'a': 1}], 'converted_spotify_playlist_id': 'cv',
+        'download_process_id': 'dp', 'last_accessed': s1['last_accessed'],
+    }]}
+    assert s1['last_accessed'] != 0  # bumped
+
+
+def test_playlist_states_empty():
+    from core.discovery.endpoints import get_playlist_states
+    body, code = get_playlist_states({}, error_label='Deezer', info_log_label='Deezer')
+    assert code == 200 and body == {"states": []}
+
+
+def test_playlist_states_optional_ids_default_none():
+    from core.discovery.endpoints import get_playlist_states
+    state = {'phase': 'fresh', 'status': 'fresh', 'discovery_progress': 0,
+             'spotify_matches': 0, 'spotify_total': 0, 'discovery_results': []}
+    body, _ = get_playlist_states({'k': state}, error_label='iTunes Link')
+    assert body['states'][0]['converted_spotify_playlist_id'] is None
+    assert body['states'][0]['download_process_id'] is None
+
+
+def test_playlist_states_missing_required_field_raises_500():
+    from core.discovery.endpoints import get_playlist_states
+    # state missing 'phase' -> strict access raises -> 500
+    body, code = get_playlist_states({'k': {'status': 'x'}}, error_label='Qobuz')
+    assert code == 500 and "error" in body
+
+
+# ---------------------------------------------------------------------------
+# start_sync
+# ---------------------------------------------------------------------------
+
+def _start_kwargs(infra, *, convert_fn=None, name='PL', image='img',
+                  activity_label='Tidal', error_label='Tidal',
+                  not_found_message='Tidal playlist not found',
+                  not_ready_message='Tidal playlist not ready for sync',
+                  sync_id_prefix='tidal'):
+    submitted = []
+
+    def submit(sync_playlist_id, playlist_name, tracks, image_url):
+        rec = (sync_playlist_id, playlist_name, tracks, image_url)
+        submitted.append(rec)
+        return f"future:{sync_playlist_id}"
+
+    calls, add = _activity_recorder()
+    kw = dict(
+        sync_id_prefix=sync_id_prefix, not_found_message=not_found_message,
+        not_ready_message=not_ready_message,
+        convert_fn=convert_fn or (lambda r: [{'id': 't'}]),
+        playlist_name_getter=lambda s: name, playlist_image_getter=lambda s: image,
+        activity_label=activity_label, error_label=error_label,
+        sync_lock=infra['sync_lock'], sync_states=infra['sync_states'],
+        active_sync_workers=infra['active_sync_workers'],
+        submit_sync_task=submit, add_activity_item=add,
+    )
+    return kw, submitted, calls
+
+
+def test_start_sync_not_found():
+    from core.discovery.endpoints import start_sync
+    infra = _cancel_infra()
+    kw, _, _ = _start_kwargs(infra)
+    body, code = start_sync({}, 'missing', **kw)
+    assert code == 404 and body == {"error": "Tidal playlist not found"}
+
+
+def test_start_sync_not_ready_phase():
+    from core.discovery.endpoints import start_sync
+    infra = _cancel_infra()
+    kw, _, _ = _start_kwargs(infra)
+    states = {'pl': {'phase': 'discovering'}}
+    body, code = start_sync(states, 'pl', **kw)
+    assert code == 400 and body == {"error": "Tidal playlist not ready for sync"}
+
+
+def test_start_sync_no_matches():
+    from core.discovery.endpoints import start_sync
+    infra = _cancel_infra()
+    kw, _, _ = _start_kwargs(infra, convert_fn=lambda r: [])
+    states = {'pl': {'phase': 'discovered', 'discovery_results': []}}
+    body, code = start_sync(states, 'pl', **kw)
+    assert code == 400 and body == {"error": "No Spotify matches found for sync"}
+
+
+def test_start_sync_happy_path():
+    from core.discovery.endpoints import start_sync
+    infra = _cancel_infra()
+    kw, submitted, calls = _start_kwargs(
+        infra, convert_fn=lambda r: [{'id': 'a'}, {'id': 'b'}],
+        name='My Mix', image='cover.jpg',
+        activity_label='Spotify Link', error_label='Spotify Public',
+        sync_id_prefix='spotify_public')
+    states = {'h1': {'phase': 'discovered', 'discovery_results': [1, 2]}}
+    body, code = start_sync(states, 'h1', **kw)
+
+    assert code == 200
+    assert body == {"success": True, "sync_playlist_id": "spotify_public_h1"}
+    # state mutated
+    assert states['h1']['phase'] == 'syncing'
+    assert states['h1']['sync_playlist_id'] == 'spotify_public_h1'
+    assert states['h1']['sync_progress'] == {}
+    # sync infra seeded + worker registered
+    assert infra['sync_states']['spotify_public_h1'] == {"status": "starting", "progress": {}}
+    assert infra['active_sync_workers']['spotify_public_h1'] == 'future:spotify_public_h1'
+    # submit got name/tracks/image
+    assert submitted == [('spotify_public_h1', 'My Mix', [{'id': 'a'}, {'id': 'b'}], 'cover.jpg')]
+    # activity uses activity_label (not error_label) + track count
+    assert calls == [("", "Spotify Link Sync Started", "'My Mix' - 2 tracks", "Now")]
+
+
+def test_start_sync_allows_resync_phases():
+    from core.discovery.endpoints import start_sync
+    for phase in ('sync_complete', 'download_complete'):
+        infra = _cancel_infra()
+        kw, _, _ = _start_kwargs(infra)
+        states = {'pl': {'phase': phase, 'discovery_results': [1]}}
+        body, code = start_sync(states, 'pl', **kw)
+        assert code == 200, phase
+
+
+def test_start_sync_exception_returns_500():
+    from core.discovery.endpoints import start_sync
+    infra = _cancel_infra()
+    def boom(r):
+        raise RuntimeError("convert blew up")
+    kw, _, _ = _start_kwargs(infra, convert_fn=boom)
+    states = {'pl': {'phase': 'discovered', 'discovery_results': [1]}}
+    body, code = start_sync(states, 'pl', **kw)
+    assert code == 500 and "error" in body
+
+
+# ---------------------------------------------------------------------------
+# first_artist extractors
+# ---------------------------------------------------------------------------
+
+def test_first_artist_str_or_obj():
+    from core.discovery.endpoints import first_artist_str_or_obj as g
+    assert g({'artists': ['A', 'B']}) == 'A'
+    assert g({'artists': [{'name': 'Obj'}]}) == 'Obj'
+    assert g({'artists': []}) == ''
+    assert g({}) == ''
+
+
+def test_first_artist_plain():
+    from core.discovery.endpoints import first_artist_plain as g
+    assert g({'artists': ['A', 'B']}) == 'A'
+    assert g({'artists': []}) == ''
+    assert g({}) == ''
+
+
+# ---------------------------------------------------------------------------
+# update_discovery_match
+# ---------------------------------------------------------------------------
+
+class _FakeCacheDB:
+    def __init__(self):
+        self.saved = []
+
+    def save_discovery_cache_match(self, *args):
+        self.saved.append(args)
+
+
+def _update_kwargs(*, json_data, cache_db=None, getter=None):
+    from core.discovery.endpoints import first_artist_plain
+    db = cache_db or _FakeCacheDB()
+    kw = dict(
+        source_log_label='tidal', error_label='Tidal',
+        original_track_key='tidal_track',
+        original_artist_getter=getter or first_artist_plain,
+        join_artist_names=lambda arts: ", ".join(arts),
+        extract_artist_name=lambda a: str(a),
+        build_fix_modal_spotify_data=lambda st: {'built': st['id']},
+        get_discovery_cache_key=lambda name, artist: (name.lower(), artist.lower()),
+        get_database=lambda: db,
+        get_active_discovery_source=lambda: 'spotify',
+    )
+    return (lambda: json_data), kw, db
+
+
+def test_update_match_missing_fields():
+    from core.discovery.endpoints import update_discovery_match
+    gj, kw, _ = _update_kwargs(json_data={'identifier': 'p'})  # missing track_index/spotify_track
+    body, code = update_discovery_match({}, gj, **kw)
+    assert code == 400 and body == {'error': 'Missing required fields'}
+
+
+def test_update_match_state_not_found():
+    from core.discovery.endpoints import update_discovery_match
+    gj, kw, _ = _update_kwargs(json_data={
+        'identifier': 'p', 'track_index': 0, 'spotify_track': {'id': 'x'}})
+    body, code = update_discovery_match({}, gj, **kw)
+    assert code == 404 and body == {'error': 'Discovery state not found'}
+
+
+def test_update_match_invalid_index():
+    from core.discovery.endpoints import update_discovery_match
+    gj, kw, _ = _update_kwargs(json_data={
+        'identifier': 'p', 'track_index': 5,
+        'spotify_track': {'id': 'x', 'name': 'n', 'artists': [], 'album': 'a'}})
+    states = {'p': {'discovery_results': []}}
+    body, code = update_discovery_match(states, gj, **kw)
+    assert code == 400 and body == {'error': 'Invalid track index'}
+
+
+def test_update_match_happy_path_full():
+    from core.discovery.endpoints import update_discovery_match
+    sp = {'id': 'sp9', 'name': 'New Song', 'artists': ['Art1', 'Art2'],
+          'album': 'Alb', 'duration_ms': 185000, 'image_url': 'cov.jpg'}
+    gj, kw, db = _update_kwargs(json_data={
+        'identifier': 'p', 'track_index': 0, 'spotify_track': sp})
+    result = {'status': 'not_found', 'tidal_track': {'name': 'Orig', 'artists': ['OrigArt']}}
+    states = {'p': {'discovery_results': [result], 'spotify_matches': 2}}
+
+    body, code = update_discovery_match(states, gj, **kw)
+
+    assert code == 200 and body['success'] is True
+    assert result['status'] == 'Found'
+    assert result['status_class'] == 'found'
+    assert result['spotify_track'] == 'New Song'
+    assert result['spotify_artist'] == 'Art1, Art2'
+    assert result['spotify_id'] == 'sp9'
+    assert result['duration'] == '3:05'
+    assert result['spotify_data'] == {'built': 'sp9'}
+    assert result['wing_it_fallback'] is False
+    assert result['manual_match'] is True
+    assert states['p']['spotify_matches'] == 3  # incremented (was not found)
+    # cache saved with normalized key + matched_data carrying image
+    assert len(db.saved) == 1
+    key0, key1, source, score, matched, oname, oartist = db.saved[0]
+    assert (key0, key1) == ('orig', 'origart')
+    assert source == 'spotify' and score == 1.0
+    assert matched['album'] == {'name': 'Alb', 'image_url': 'cov.jpg', 'images': [{'url': 'cov.jpg'}]}
+    assert oname == 'Orig' and oartist == 'OrigArt'
+
+
+def test_update_match_no_increment_when_already_found():
+    from core.discovery.endpoints import update_discovery_match
+    sp = {'id': 'x', 'name': 'n', 'artists': ['A'], 'album': 'a', 'duration_ms': 0}
+    gj, kw, _ = _update_kwargs(json_data={'identifier': 'p', 'track_index': 0, 'spotify_track': sp})
+    result = {'status': 'Found', 'tidal_track': {}}
+    states = {'p': {'discovery_results': [result], 'spotify_matches': 5}}
+    body, code = update_discovery_match(states, gj, **kw)
+    assert code == 200
+    assert states['p']['spotify_matches'] == 5  # unchanged
+    assert result['duration'] == '0:00'
+
+
+def test_update_match_cache_error_is_swallowed():
+    from core.discovery.endpoints import update_discovery_match
+    class _BoomDB:
+        def save_discovery_cache_match(self, *a):
+            raise RuntimeError("db down")
+    sp = {'id': 'x', 'name': 'n', 'artists': ['A'], 'album': 'a', 'duration_ms': 0}
+    gj, kw, _ = _update_kwargs(json_data={'identifier': 'p', 'track_index': 0, 'spotify_track': sp},
+                               cache_db=_BoomDB())
+    states = {'p': {'discovery_results': [{'status': 'x', 'tidal_track': {}}], 'spotify_matches': 0}}
+    body, code = update_discovery_match(states, gj, **kw)
+    assert code == 200 and body['success'] is True  # cache failure doesn't fail the request
+
+
+def test_update_match_get_json_raises_returns_500():
+    from core.discovery.endpoints import update_discovery_match
+    def boom():
+        raise ValueError("bad json")
+    _, kw, _ = _update_kwargs(json_data={})
+    body, code = update_discovery_match({}, boom, **kw)
+    assert code == 500 and 'error' in body
+
+
+# ---------------------------------------------------------------------------
+# update_playlist_phase
+# ---------------------------------------------------------------------------
+
+_PHASES = ['fresh', 'discovering', 'discovered', 'syncing', 'sync_complete', 'downloading', 'download_complete']
+
+
+def test_update_phase_not_found():
+    from core.discovery.endpoints import update_playlist_phase
+    body, code = update_playlist_phase({}, 'k', lambda: {'phase': 'fresh'},
+                                       not_found_message='Tidal playlist not found',
+                                       error_label='Tidal', valid_phases=_PHASES,
+                                       apply_extra_fields=False)
+    assert code == 404 and body == {"error": "Tidal playlist not found"}
+
+
+def test_update_phase_missing_phase():
+    from core.discovery.endpoints import update_playlist_phase
+    states = {'k': {'phase': 'discovered'}}
+    body, code = update_playlist_phase(states, 'k', lambda: {},
+                                       not_found_message='nf', error_label='Tidal',
+                                       valid_phases=_PHASES, apply_extra_fields=False)
+    assert code == 400 and body == {"error": "Phase not provided"}
+
+
+def test_update_phase_invalid():
+    from core.discovery.endpoints import update_playlist_phase
+    states = {'k': {'phase': 'discovered'}}
+    body, code = update_playlist_phase(states, 'k', lambda: {'phase': 'bogus'},
+                                       not_found_message='nf', error_label='Tidal',
+                                       valid_phases=_PHASES, apply_extra_fields=False)
+    assert code == 400 and 'Invalid phase' in body['error']
+
+
+def test_update_phase_happy_no_extra_fields():
+    from core.discovery.endpoints import update_playlist_phase
+    state = {'phase': 'download_complete', 'last_accessed': 0}
+    body, code = update_playlist_phase({'k': state}, 'k',
+                                       lambda: {'phase': 'discovered',
+                                                'download_process_id': 'dp'},
+                                       not_found_message='nf', error_label='Tidal',
+                                       valid_phases=_PHASES, apply_extra_fields=False)
+    assert code == 200
+    assert body == {"success": True, "message": "Phase updated to discovered",
+                    "old_phase": "download_complete", "new_phase": "discovered"}
+    assert state['phase'] == 'discovered'
+    assert state['last_accessed'] != 0
+    # apply_extra_fields=False -> download_process_id NOT applied (1:1 for Tidal/YouTube)
+    assert 'download_process_id' not in state
+
+
+def test_update_phase_applies_extra_fields_when_enabled():
+    from core.discovery.endpoints import update_playlist_phase
+    state = {'phase': 'discovered'}
+    body, code = update_playlist_phase({'k': state}, 'k',
+                                       lambda: {'phase': 'downloading',
+                                                'download_process_id': 'dp7',
+                                                'converted_spotify_playlist_id': 'cv7'},
+                                       not_found_message='nf', error_label='Deezer',
+                                       valid_phases=_PHASES, apply_extra_fields=True)
+    assert code == 200
+    assert state['download_process_id'] == 'dp7'
+    assert state['converted_spotify_playlist_id'] == 'cv7'
+
+
+def test_update_phase_youtube_parsed_allowed():
+    from core.discovery.endpoints import update_playlist_phase
+    yt_phases = ['fresh', 'parsed', 'discovering', 'discovered', 'syncing',
+                 'sync_complete', 'downloading', 'download_complete']
+    state = {'phase': 'discovered'}
+    body, code = update_playlist_phase({'k': state}, 'k', lambda: {'phase': 'parsed'},
+                                       not_found_message='nf', error_label='YouTube',
+                                       valid_phases=yt_phases, apply_extra_fields=False)
+    assert code == 200 and state['phase'] == 'parsed'
+
+
+def test_update_phase_exception_500():
+    from core.discovery.endpoints import update_playlist_phase
+    def boom():
+        raise ValueError('bad')
+    states = {'k': {'phase': 'discovered'}}
+    body, code = update_playlist_phase(states, 'k', boom, not_found_message='nf',
+                                       error_label='Tidal', valid_phases=_PHASES,
+                                       apply_extra_fields=False)
+    assert code == 500 and 'error' in body
+
+
+# ---------------------------------------------------------------------------
+# save_bubble_snapshot
+# ---------------------------------------------------------------------------
+
+class _SnapDB:
+    def __init__(self):
+        self.saved = []
+
+    def save_bubble_snapshot(self, kind, items, profile_id=None):
+        self.saved.append((kind, items, profile_id))
+
+
+def _snap_kwargs(db, *, payload_key='bubbles', no_data_error='No bubble data provided',
+                 snapshot_kind='artist_bubbles', success_noun='artist bubbles',
+                 log_subject='artist bubble snapshot', log_noun='artists'):
+    return dict(
+        payload_key=payload_key, no_data_error=no_data_error, snapshot_kind=snapshot_kind,
+        success_noun=success_noun, log_subject=log_subject, log_noun=log_noun,
+        get_database=lambda: db, get_current_profile_id=lambda: 7,
+    )
+
+
+def test_snapshot_missing_payload_key():
+    from core.discovery.endpoints import save_bubble_snapshot
+    db = _SnapDB()
+    body, code = save_bubble_snapshot(lambda: {'other': 1}, **_snap_kwargs(db))
+    assert code == 400 and body == {'success': False, 'error': 'No bubble data provided'}
+    assert db.saved == []
+
+
+def test_snapshot_none_body():
+    from core.discovery.endpoints import save_bubble_snapshot
+    db = _SnapDB()
+    body, code = save_bubble_snapshot(lambda: None,
+                                      **_snap_kwargs(db, payload_key='downloads',
+                                                     no_data_error='No download data provided'))
+    assert code == 400 and body['error'] == 'No download data provided'
+
+
+def test_snapshot_happy_path():
+    from core.discovery.endpoints import save_bubble_snapshot
+    db = _SnapDB()
+    items = [{'a': 1}, {'b': 2}, {'c': 3}]
+    body, code = save_bubble_snapshot(lambda: {'bubbles': items},
+                                      **_snap_kwargs(db, snapshot_kind='search_bubbles',
+                                                     success_noun='search bubbles'))
+    assert code == 200
+    assert body['success'] is True
+    assert body['message'] == 'Snapshot saved with 3 search bubbles'
+    assert isinstance(body['timestamp'], str) and body['timestamp']
+    # persisted with kind + profile id
+    assert db.saved == [('search_bubbles', items, 7)]
+
+
+def test_snapshot_discover_downloads_key():
+    from core.discovery.endpoints import save_bubble_snapshot
+    db = _SnapDB()
+    body, code = save_bubble_snapshot(lambda: {'downloads': [1, 2]},
+                                      **_snap_kwargs(db, payload_key='downloads',
+                                                     no_data_error='No download data provided',
+                                                     snapshot_kind='discover_downloads',
+                                                     success_noun='downloads',
+                                                     log_subject='discover download snapshot',
+                                                     log_noun='downloads'))
+    assert code == 200
+    assert body['message'] == 'Snapshot saved with 2 downloads'
+    assert db.saved[0][0] == 'discover_downloads'
+
+
+def test_snapshot_exception_returns_500():
+    from core.discovery.endpoints import save_bubble_snapshot
+    class _BoomDB:
+        def save_bubble_snapshot(self, *a, **k):
+            raise RuntimeError('db down')
+    body, code = save_bubble_snapshot(lambda: {'bubbles': [1]}, **_snap_kwargs(_BoomDB()))
+    assert code == 500 and body == {'success': False, 'error': 'db down'}