diff --git a/core/reorganize_queue.py b/core/reorganize_queue.py new file mode 100644 index 00000000..6b50afb0 --- /dev/null +++ b/core/reorganize_queue.py @@ -0,0 +1,453 @@ +"""FIFO queue for library album reorganize requests. + +Replaces the single-slot "one reorganize at a time, return 409 on +collision" model with a queue: clicks always succeed (or surface +"already queued" on dedupe), the user can fan-out clicks across +albums or hit "Reorganize All", and a single background worker +chews through the queue in submission order. + +Design rules: + +- **Single global queue**, single worker thread. Reorganize is + I/O-heavy (file copy, mutagen tagging, AcoustID, possibly ffmpeg) + and post-process is not designed for cross-album concurrency. + In-album track parallelism still happens inside `reorganize_album` + (3 worker threads — see `_REORGANIZE_MAX_WORKERS`). + +- **Dedupe on enqueue**: an album that's already queued or currently + running is rejected silently. Stops the user from spamming the + same album N times by clicking the button repeatedly. + +- **Per-item source**: each queued item carries its own `source` + string (the user's per-album modal pick). Worker passes it + through to `reorganize_album(primary_source=..., strict_source=...)`. + +- **Continue on failure**: a failed item doesn't stop the queue. + Worker logs the failure, marks the item `failed`, moves on. + +- **Cancel queued items**: items in `queued` state can be cancelled + (drop from queue). The currently-running item can NOT be cancelled + mid-flight — Python threads aren't cleanly killable, and post- + process spawns subprocesses we can't safely interrupt. Cancel + changes the item's status to `cancelled` and removes it from the + active queue. + +- **In-memory only**: queue state lives in a module-level singleton. + A server restart loses the queue (in-flight item likely also lost + half-way through post-process). DB persistence is a follow-up if + this turns out to matter operationally. +""" + +import threading +import time +import uuid +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional + +from utils.logging_config import get_logger + +logger = get_logger("reorganize_queue") + + +# How many recently-completed items to retain for the snapshot endpoint. +# The status panel uses these to show "just-finished" cards briefly so +# the user sees outcomes scroll past instead of items vanishing. +_RECENT_HISTORY_CAP = 30 + + +@dataclass +class QueueItem: + """One album waiting (or being processed) in the reorganize queue.""" + + queue_id: str # uuid; how the API references this item + album_id: str + album_title: str # captured at enqueue time for UI display + artist_id: Optional[str] + artist_name: str # captured at enqueue time for UI display + source: Optional[str] # the user's per-modal pick (None = auto) + enqueued_at: float + status: str = 'queued' # queued | running | done | failed | cancelled + started_at: Optional[float] = None + finished_at: Optional[float] = None + # Populated by the worker after each item finishes — surfaced to the + # status panel so users see counts + per-item error messages. + result_status: Optional[str] = None # mirrors `reorganize_album` summary['status'] + result_source: Optional[str] = None # which source the orchestrator actually used + moved: int = 0 + skipped: int = 0 + failed: int = 0 + error: Optional[str] = None # shorthand for the first error, for the toast + # Live-progress fields for the currently-running item; cleared when + # the worker moves on so the snapshot stays small. + current_track: Optional[str] = None + progress_total: int = 0 + progress_processed: int = 0 + + def to_snapshot(self) -> dict: + return { + 'queue_id': self.queue_id, + 'album_id': self.album_id, + 'album_title': self.album_title, + 'artist_id': self.artist_id, + 'artist_name': self.artist_name, + 'source': self.source, + 'enqueued_at': self.enqueued_at, + 'started_at': self.started_at, + 'finished_at': self.finished_at, + 'status': self.status, + 'result_status': self.result_status, + 'result_source': self.result_source, + 'moved': self.moved, + 'skipped': self.skipped, + 'failed': self.failed, + 'error': self.error, + 'current_track': self.current_track, + 'progress_total': self.progress_total, + 'progress_processed': self.progress_processed, + } + + +class ReorganizeQueue: + """Module-level singleton that owns the queue + worker thread. + + Use the module-level :func:`get_queue` accessor — don't construct + directly. The class is documented public-style so tests can spin + up isolated instances. + """ + + def __init__(self, *, runner: Optional[Callable[[QueueItem], dict]] = None): + """ + Args: + runner: Callable that takes a `QueueItem` and runs the + actual reorganize, returning a summary dict with + ``status``, ``source``, ``moved``, ``skipped``, + ``failed``, ``errors`` keys (the shape + ``reorganize_album`` already returns). Tests inject + a fake runner; production wires the real one in + via :func:`set_runner`. + """ + # Single Condition variable owns both mutual exclusion and the + # idle-worker wait. Using a Condition (vs Lock + Event) closes a + # race where the worker could clear an event right after enqueue + # set it, causing the new item to sleep for the timeout window. + # cond.wait() releases the lock and re-acquires on notify, so + # state checks and waits are properly interleaved. + self._cond = threading.Condition() + self._items: List[QueueItem] = [] # everything ever submitted (active + recent) + self._runner = runner + self._worker: Optional[threading.Thread] = None + self._stopped = False + + # -- public API -------------------------------------------------- + + def set_runner(self, runner: Callable[[QueueItem], dict]) -> None: + """Inject the function that does the actual reorganize work. + Web_server calls this once at startup with a closure over the + injected dependencies (post-process fn, db, etc.).""" + with self._cond: + self._runner = runner + + def enqueue( + self, + *, + album_id: str, + album_title: str, + artist_id: Optional[str], + artist_name: str, + source: Optional[str] = None, + ) -> dict: + """Add an album to the queue. Returns a result dict: + + {'queued': True, 'queue_id': '...', 'position': N} + {'queued': False, 'reason': 'already_queued', 'queue_id': '...'} + + Dedupe: if this album is already in `queued` or `running` + status, returns the existing entry's queue_id rather than + adding a duplicate. ``cancelled`` / ``done`` / ``failed`` + items don't block re-enqueue (user retried after a failure). + """ + with self._cond: + for existing in self._items: + if existing.album_id == album_id and existing.status in ('queued', 'running'): + return { + 'queued': False, + 'reason': 'already_queued', + 'queue_id': existing.queue_id, + } + + item = QueueItem( + queue_id=uuid.uuid4().hex[:12], + album_id=album_id, + album_title=album_title, + artist_id=artist_id, + artist_name=artist_name, + source=source, + enqueued_at=time.time(), + ) + self._items.append(item) + position = sum(1 for i in self._items if i.status == 'queued') + self._ensure_worker() + self._cond.notify_all() + logger.info( + f"[Queue] Enqueued '{album_title}' (album_id={album_id}, " + f"queue_id={item.queue_id}, position={position}, source={source or 'auto'})" + ) + return { + 'queued': True, + 'queue_id': item.queue_id, + 'position': position, + } + + def enqueue_many(self, items: List[Dict[str, Any]]) -> Dict[str, int]: + """Bulk-enqueue a list of items. Each ``item`` is a dict with + the same keys :meth:`enqueue` accepts (``album_id``, + ``album_title``, ``artist_id``, ``artist_name``, ``source``). + Dedupe still applies per-album-id. + + Holds the queue lock for the entire batch so two things hold: + (1) the worker can't start draining mid-batch, and (2) duplicate + album_ids inside the same batch get deduped against each other, + not just against pre-existing items. Without (2), a fast runner + could finish the first copy before the loop reached the second + and both would enqueue. + + Returns a tally dict ``{'enqueued': N, 'already_queued': M, + 'total': len(items)}`` so the caller can report bulk results + without doing the counting themselves. Used by the bulk + Reorganize-All endpoint and any future maintenance jobs that + enqueue at scale. + """ + enqueued = 0 + already = 0 + seen_in_batch: set = set() + with self._cond: + # Snapshot album_ids that already block re-enqueue so we don't + # rescan self._items per row. + blocked = { + i.album_id for i in self._items if i.status in ('queued', 'running') + } + for raw in items: + album_id = str(raw['album_id']) + if album_id in blocked or album_id in seen_in_batch: + already += 1 + continue + seen_in_batch.add(album_id) + item = QueueItem( + queue_id=uuid.uuid4().hex[:12], + album_id=album_id, + album_title=raw.get('album_title') or 'Unknown Album', + artist_id=str(raw['artist_id']) if raw.get('artist_id') is not None else None, + artist_name=raw.get('artist_name') or 'Unknown Artist', + source=raw.get('source'), + enqueued_at=time.time(), + ) + self._items.append(item) + enqueued += 1 + logger.info( + f"[Queue] Bulk-enqueued '{item.album_title}' (album_id={album_id}, " + f"queue_id={item.queue_id}, source={item.source or 'auto'})" + ) + if enqueued: + self._ensure_worker() + self._cond.notify_all() + return {'enqueued': enqueued, 'already_queued': already, 'total': len(items)} + + def cancel(self, queue_id: str) -> dict: + """Cancel a queued item. The currently-running item cannot be + cancelled (Python threads aren't cleanly killable; post-process + may have spawned ffmpeg).""" + with self._cond: + for item in self._items: + if item.queue_id != queue_id: + continue + if item.status == 'queued': + item.status = 'cancelled' + item.finished_at = time.time() + logger.info(f"[Queue] Cancelled queued item {queue_id} ('{item.album_title}')") + return {'cancelled': True} + if item.status == 'running': + return {'cancelled': False, 'reason': 'running_cant_cancel'} + return {'cancelled': False, 'reason': 'not_active'} + return {'cancelled': False, 'reason': 'not_found'} + + def clear_queued(self) -> int: + """Cancel ALL queued items (running item continues). Returns + the count of items cancelled.""" + cancelled = 0 + with self._cond: + now = time.time() + for item in self._items: + if item.status == 'queued': + item.status = 'cancelled' + item.finished_at = now + cancelled += 1 + if cancelled: + logger.info(f"[Queue] Bulk-cancelled {cancelled} queued items") + return cancelled + + def snapshot(self) -> dict: + """Current queue state for the status panel. Returns: + + { + 'active': item dict | None, + 'queued': [item dicts in FIFO order], + 'recent': [item dicts in finish order, newest first, capped], + 'totals': {'queued': N, 'running': M, 'done_today': K, ...}, + } + """ + with self._cond: + active = next((i for i in self._items if i.status == 'running'), None) + queued = [i for i in self._items if i.status == 'queued'] + recent = [i for i in self._items if i.status in ('done', 'failed', 'cancelled')] + recent.sort(key=lambda i: i.finished_at or 0, reverse=True) + recent = recent[:_RECENT_HISTORY_CAP] + + return { + 'active': active.to_snapshot() if active else None, + 'queued': [i.to_snapshot() for i in queued], + 'recent': [i.to_snapshot() for i in recent], + 'totals': { + 'queued': len(queued), + 'running': 1 if active else 0, + 'done': sum(1 for i in self._items if i.status == 'done'), + 'failed': sum(1 for i in self._items if i.status == 'failed'), + 'cancelled': sum(1 for i in self._items if i.status == 'cancelled'), + }, + } + + def stop(self) -> None: + """Stop the worker (called on server shutdown).""" + with self._cond: + self._stopped = True + self._cond.notify_all() + + # -- internals --------------------------------------------------- + + def _ensure_worker(self) -> None: + """Lazy worker start — only spawn the thread when there's + actually something to process. Caller MUST hold ``_cond``.""" + if self._worker is not None and self._worker.is_alive(): + return + self._worker = threading.Thread( + target=self._run, daemon=True, name='ReorganizeQueueWorker' + ) + self._worker.start() + + def _claim_next_or_wait(self) -> Optional[QueueItem]: + """Atomically pick the next queued item AND flip it to 'running' + under a single lock acquisition. If the queue is empty, block + on ``_cond.wait()`` (which releases the lock while sleeping) + and return None when we're notified or timeout. Returning the + item already-marked-running closes the cancel-vs-run race: a + cancel() call now sees status='running' and is rejected.""" + with self._cond: + while not self._stopped: + for item in self._items: + if item.status == 'queued': + item.status = 'running' + item.started_at = time.time() + return item + # No queued items — wait for an enqueue or shutdown. + # 60s timeout so a stuck notify (shouldn't happen, but + # defensive) doesn't park the worker forever. + self._cond.wait(timeout=60) + return None + + def _run(self) -> None: + """Worker loop: pull next queued, run it, mark done, repeat. + Idles on `_cond.wait()` when queue is empty.""" + logger.info("[Queue] Worker thread started") + while not self._stopped: + item = self._claim_next_or_wait() + if item is None: + # Only happens on shutdown — `_claim_next_or_wait` only + # returns None once `_stopped` is True. Loop back to the + # `while not self._stopped` check, which exits. + continue + logger.info(f"[Queue] Starting '{item.album_title}' (queue_id={item.queue_id})") + + try: + runner = self._runner + if runner is None: + raise RuntimeError("Queue has no runner configured — call set_runner() at startup") + summary = runner(item) + except Exception as e: + logger.error( + f"[Queue] Runner raised for '{item.album_title}': {e}", + exc_info=True, + ) + with self._cond: + item.status = 'failed' + item.error = str(e) + item.finished_at = time.time() + continue + + with self._cond: + item.moved = int(summary.get('moved', 0)) + item.skipped = int(summary.get('skipped', 0)) + item.failed = int(summary.get('failed', 0)) + item.result_status = summary.get('status') + item.result_source = summary.get('source') + errors = summary.get('errors') or [] + if errors: + first_err = errors[0] if isinstance(errors[0], dict) else {'error': str(errors[0])} + item.error = first_err.get('error') or first_err.get('reason') + # 'failed' status only when the run produced concrete failed tracks + # OR ended in a non-completed state (no_source_id / no_album / etc). + item.status = 'failed' if (item.failed > 0 or item.result_status not in (None, 'completed')) else 'done' + item.finished_at = time.time() + # Clear live-progress fields — done items don't need them. + item.current_track = None + item.progress_total = 0 + item.progress_processed = 0 + + logger.info( + f"[Queue] Finished '{item.album_title}' — status={item.status}, " + f"moved={item.moved}, skipped={item.skipped}, failed={item.failed}" + ) + logger.info("[Queue] Worker thread exiting") + + # Called by the runner (or test) to push live progress onto the + # currently-running item. Safe to call from worker thread inside + # reorganize_album's on_progress callback. + def update_active_progress(self, *, queue_id: str, **fields) -> None: + with self._cond: + for item in self._items: + if item.queue_id == queue_id and item.status == 'running': + if 'current_track' in fields: + item.current_track = fields['current_track'] + if 'total' in fields: + item.progress_total = int(fields['total']) + if 'processed' in fields: + item.progress_processed = int(fields['processed']) + if 'moved' in fields: + item.moved = int(fields['moved']) + if 'skipped' in fields: + item.skipped = int(fields['skipped']) + if 'failed' in fields: + item.failed = int(fields['failed']) + return + + +# Module-level singleton accessor --------------------------------------------- + +_singleton: Optional[ReorganizeQueue] = None +_singleton_lock = threading.Lock() + + +def get_queue() -> ReorganizeQueue: + global _singleton + with _singleton_lock: + if _singleton is None: + _singleton = ReorganizeQueue() + return _singleton + + +def reset_queue_for_tests() -> None: + """Test-only: drop the singleton so the next get_queue() returns + a fresh instance. Production code never calls this.""" + global _singleton + with _singleton_lock: + if _singleton is not None: + _singleton.stop() + _singleton = None diff --git a/core/reorganize_runner.py b/core/reorganize_runner.py new file mode 100644 index 00000000..b5fba687 --- /dev/null +++ b/core/reorganize_runner.py @@ -0,0 +1,123 @@ +"""Builds the per-item runner closure that the reorganize queue worker +invokes. Lives outside ``web_server`` so the wiring is unit-testable +and the monolith stays small. + +The runner ties three subsystems together: + +* :func:`core.library_reorganize.reorganize_album` — the orchestrator + that copies files to staging, matches them against the metadata + source, and routes each through the post-process pipeline. +* :func:`core.reorganize_queue.get_queue` — the queue this runner is + registered with; we forward live progress updates back into the + active queue item so the status panel can show per-track state. +* The dependency callbacks injected by ``web_server`` (DB accessor, + resolve-file-path, post-process function, empty-dir cleanup, + shutdown signal). These are passed in rather than imported so the + module stays testable in isolation. + +Config (download path / transfer path) is read **per run**, not at +module load. That way a user changing their download path in settings +takes effect on the next reorganize without needing a server restart. +""" + +import os +from typing import Callable, Optional + +from utils.logging_config import get_logger + +logger = get_logger("reorganize_runner") + + +def build_runner( + *, + get_database: Callable[[], object], + resolve_file_path_fn: Callable[[Optional[str]], Optional[str]], + post_process_fn: Callable[[str, dict, str], None], + cleanup_empty_directories_fn: Callable[[str, str], None], + is_shutting_down_fn: Callable[[], bool], + get_download_path: Callable[[], str], + get_transfer_path: Callable[[], str], +) -> Callable[[object], dict]: + """Return the closure the queue worker invokes per item. + + Args: + get_database: Returns the live MusicDatabase singleton. + resolve_file_path_fn: Resolves a DB-stored file path to the + actual on-disk path (or ``None`` if missing). + post_process_fn: ``_post_process_matched_download``. Must set + ``context['_final_processed_path']`` on success. + cleanup_empty_directories_fn: Called as + ``cleanup_empty_directories_fn(transfer_dir, marker_path)`` + to prune empty source dirs after a track is moved. + is_shutting_down_fn: Returns True when the server is shutting + down so the orchestrator can abort early. + get_download_path: Resolves the user's configured download + path *at call time* (so config changes apply live). + get_transfer_path: Same, for the transfer path. + + Returns: + A callable ``runner(item)`` suitable for + :meth:`core.reorganize_queue.ReorganizeQueue.set_runner`. + """ + from core.library_reorganize import reorganize_album + from core.reorganize_queue import get_queue + + def _update_track_path(track_id, new_path): + try: + db = get_database() + with db._get_connection() as conn: + conn.execute( + "UPDATE tracks SET file_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", + (new_path, str(track_id)), + ) + conn.commit() + except Exception as db_err: + logger.warning(f"[Reorganize] DB path update failed for {track_id}: {db_err}") + + def runner(item): + # Read config per-run so the user changing their download path + # in Settings takes effect on the next reorganize without a + # server restart. + download_dir = get_download_path() + transfer_dir = get_transfer_path() + staging_root = os.path.join(download_dir, 'ssync_staging') + try: + os.makedirs(staging_root, exist_ok=True) + except OSError as mk_err: + logger.error(f"[Reorganize] Cannot create staging dir {staging_root}: {mk_err}") + return { + 'status': 'setup_failed', + 'source': None, + 'total': 0, 'moved': 0, 'skipped': 0, 'failed': 0, + 'errors': [{'error': f'Could not create staging dir: {mk_err}'}], + } + + def _cleanup_empty(src_dir): + try: + cleanup_empty_directories_fn(transfer_dir, os.path.join(src_dir, '_')) + except Exception: + pass + + def _on_progress(updates): + try: + get_queue().update_active_progress(queue_id=item.queue_id, **updates) + except Exception: + # Progress fan-out failures must never break a run. + pass + + return reorganize_album( + album_id=item.album_id, + db=get_database(), + staging_root=staging_root, + resolve_file_path_fn=resolve_file_path_fn, + post_process_fn=post_process_fn, + update_track_path_fn=_update_track_path, + cleanup_empty_dir_fn=_cleanup_empty, + transfer_dir=transfer_dir, + on_progress=_on_progress, + primary_source=item.source, + strict_source=bool(item.source), + stop_check=is_shutting_down_fn, + ) + + return runner diff --git a/database/music_database.py b/database/music_database.py index ab014696..d4f68f0a 100644 --- a/database/music_database.py +++ b/database/music_database.py @@ -4795,6 +4795,63 @@ class MusicDatabase: logger.error(f"Error inserting/updating {server_source} album {getattr(album_obj, 'title', 'Unknown')}: {e}") return False + def get_album_display_meta(self, album_id) -> Optional[Dict[str, Any]]: + """Return ``{album_title, artist_id, artist_name}`` for an album row. + + Used by the reorganize queue enqueue endpoint to capture display + strings at submission time so the status panel can render + without a DB lookup per poll. Returns None when the album row + does not exist; lets DB errors bubble up so callers can surface + a real failure instead of swallowing it as "album not found". + """ + with self._get_connection() as conn: + cursor = conn.cursor() + cursor.execute( + """ + SELECT al.title AS album_title, + ar.id AS artist_id, + ar.name AS artist_name + FROM albums al + JOIN artists ar ON al.artist_id = ar.id + WHERE al.id = ? + """, + (str(album_id),), + ) + row = cursor.fetchone() + if not row: + return None + return { + 'album_title': row['album_title'] or 'Unknown Album', + 'artist_id': str(row['artist_id']) if row['artist_id'] is not None else None, + 'artist_name': row['artist_name'] or 'Unknown Artist', + } + + def get_artist_albums_for_reorganize(self, artist_id) -> List[Dict[str, Any]]: + """Return ``[{album_id, album_title, artist_id, artist_name}, ...]`` + for every album owned by ``artist_id``, ordered by year then + title. Used by the bulk Reorganize-All endpoint to pull the + full tracklist server-side instead of trusting whatever the + frontend cached. Returns an empty list when the artist has no + albums; lets DB errors bubble so a real failure surfaces as a + 500 rather than masquerading as "no albums found". + """ + with self._get_connection() as conn: + cursor = conn.cursor() + cursor.execute( + """ + SELECT al.id AS album_id, + al.title AS album_title, + ar.id AS artist_id, + ar.name AS artist_name + FROM albums al + JOIN artists ar ON al.artist_id = ar.id + WHERE ar.id = ? + ORDER BY al.year ASC, al.title ASC + """, + (str(artist_id),), + ) + return [dict(r) for r in cursor.fetchall()] + def get_albums_by_artist(self, artist_id: int) -> List[DatabaseAlbum]: """Get all albums by artist ID""" try: diff --git a/tests/test_reorganize_db_methods.py b/tests/test_reorganize_db_methods.py new file mode 100644 index 00000000..27e4609d --- /dev/null +++ b/tests/test_reorganize_db_methods.py @@ -0,0 +1,213 @@ +"""Tests for the reorganize-queue DB helpers on `MusicDatabase`: + +- ``get_album_display_meta(album_id)`` — returns the title/artist tuple + the queue uses for status-panel display, or None when not found. +- ``get_artist_albums_for_reorganize(artist_id)`` — returns the + bulk-enqueue list ordered by year then title. + +These are isolated DB-method tests so the SQL itself is verified +without spinning up Flask, the queue worker, or the orchestrator. +""" + +import sqlite3 +import sys +import types + +import pytest + + +# ── stubs (same shape used elsewhere in the test suite) ─────────────────── +if "spotipy" not in sys.modules: + spotipy = types.ModuleType("spotipy") + spotipy.Spotify = object + oauth2 = types.ModuleType("spotipy.oauth2") + oauth2.SpotifyOAuth = object + oauth2.SpotifyClientCredentials = object + spotipy.oauth2 = oauth2 + sys.modules["spotipy"] = spotipy + sys.modules["spotipy.oauth2"] = oauth2 + +if "config.settings" not in sys.modules: + config_pkg = types.ModuleType("config") + settings_mod = types.ModuleType("config.settings") + + class _DummyConfigManager: + def get(self, key, default=None): + return default + + def get_active_media_server(self): + return "primary" + + settings_mod.config_manager = _DummyConfigManager() + config_pkg.settings = settings_mod + sys.modules["config"] = config_pkg + sys.modules["config.settings"] = settings_mod + + +from database.music_database import MusicDatabase # noqa: E402 + + +# ── helpers ─────────────────────────────────────────────────────────────── + + +class _InMemoryDB(MusicDatabase): + """MusicDatabase that uses an in-memory sqlite that survives across + `_get_connection()` calls. Lets tests seed rows once and have the + methods under test see them.""" + + def __init__(self): + # Skip the real __init__ — it would try to migrate a real db. + self._conn = sqlite3.connect(":memory:") + self._conn.row_factory = sqlite3.Row + + def _get_connection(self): + return _NonClosingConn(self._conn) + + +class _NonClosingConn: + """Wraps the shared sqlite connection so `with db._get_connection() + as conn:` doesn't close the underlying handle between calls.""" + def __init__(self, real): + self._real = real + + def cursor(self): + return self._real.cursor() + + def commit(self): + return self._real.commit() + + def close(self): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +def _seed(db, *, artists=(), albums=()): + cur = db._conn.cursor() + cur.execute("CREATE TABLE artists (id TEXT PRIMARY KEY, name TEXT)") + cur.execute(""" + CREATE TABLE albums ( + id TEXT PRIMARY KEY, + artist_id TEXT, + title TEXT, + year INTEGER + ) + """) + for ar in artists: + cur.execute("INSERT INTO artists VALUES (?, ?)", ar) + for al in albums: + cur.execute( + "INSERT INTO albums (id, artist_id, title, year) VALUES (?, ?, ?, ?)", + al, + ) + db._conn.commit() + + +@pytest.fixture +def db(): + return _InMemoryDB() + + +# ── get_album_display_meta ──────────────────────────────────────────────── + + +def test_get_album_display_meta_returns_dict_for_known_album(db): + _seed(db, + artists=[('ar-1', 'Kendrick Lamar')], + albums=[('alb-1', 'ar-1', 'good kid, m.A.A.d city', 2012)]) + meta = db.get_album_display_meta('alb-1') + assert meta == { + 'album_title': 'good kid, m.A.A.d city', + 'artist_id': 'ar-1', + 'artist_name': 'Kendrick Lamar', + } + + +def test_get_album_display_meta_returns_none_for_missing_album(db): + _seed(db, artists=[('ar-1', 'Aerosmith')]) + assert db.get_album_display_meta('does-not-exist') is None + + +def test_get_album_display_meta_falls_back_for_blank_strings(db): + """Albums with empty title or artist name in the DB still need a + safe display value — the queue UI should never render '(blank)'.""" + _seed(db, + artists=[('ar-1', '')], + albums=[('alb-1', 'ar-1', '', 2015)]) + meta = db.get_album_display_meta('alb-1') + assert meta['album_title'] == 'Unknown Album' + assert meta['artist_name'] == 'Unknown Artist' + assert meta['artist_id'] == 'ar-1' + + +# ── get_artist_albums_for_reorganize ────────────────────────────────────── + + +def test_get_artist_albums_for_reorganize_orders_by_year_then_title(db): + _seed(db, + artists=[('ar-1', 'Aerosmith')], + albums=[ + ('alb-c', 'ar-1', 'Toys in the Attic', 1975), + ('alb-a', 'ar-1', 'Aerosmith', 1973), + ('alb-b', 'ar-1', 'Get Your Wings', 1974), + ]) + rows = db.get_artist_albums_for_reorganize('ar-1') + assert [r['album_id'] for r in rows] == ['alb-a', 'alb-b', 'alb-c'] + assert all(r['artist_name'] == 'Aerosmith' for r in rows) + + +def test_get_artist_albums_for_reorganize_secondary_sorts_by_title(db): + """Same release year → tiebreak on title alphabetically.""" + _seed(db, + artists=[('ar-1', 'X')], + albums=[ + ('alb-z', 'ar-1', 'Zebra', 1990), + ('alb-a', 'ar-1', 'Apple', 1990), + ('alb-m', 'ar-1', 'Mango', 1990), + ]) + rows = db.get_artist_albums_for_reorganize('ar-1') + assert [r['album_title'] for r in rows] == ['Apple', 'Mango', 'Zebra'] + + +def test_get_artist_albums_for_reorganize_returns_empty_for_unknown_artist(db): + _seed(db, artists=[('ar-1', 'Aerosmith')]) + assert db.get_artist_albums_for_reorganize('not-a-real-artist') == [] + + +def test_get_artist_albums_for_reorganize_isolates_by_artist(db): + """Pulling albums for artist A must NOT leak in albums from artist B.""" + _seed(db, + artists=[('ar-1', 'A'), ('ar-2', 'B')], + albums=[ + ('alb-1', 'ar-1', 'A1', 2000), + ('alb-2', 'ar-2', 'B1', 2000), + ('alb-3', 'ar-1', 'A2', 2001), + ]) + rows = db.get_artist_albums_for_reorganize('ar-1') + assert {r['album_id'] for r in rows} == {'alb-1', 'alb-3'} + + +# ── error propagation ──────────────────────────────────────────────────── +# Regression for review feedback on the original PR: helpers used to +# swallow every Exception and return None / [], so a real DB outage +# masqueraded as "album not found" / "no albums". Now they let the +# error bubble — the route layer turns it into a 500 — so the user sees +# a real failure instead of a phantom empty state. + + +def test_get_album_display_meta_propagates_db_errors(db): + """If the underlying tables don't exist, the helper must raise + rather than swallow it as a missing-album result.""" + # Don't seed — the schema is empty, so the SELECT will fail with + # OperationalError ("no such table: albums"). + with pytest.raises(sqlite3.OperationalError): + db.get_album_display_meta('alb-1') + + +def test_get_artist_albums_for_reorganize_propagates_db_errors(db): + with pytest.raises(sqlite3.OperationalError): + db.get_artist_albums_for_reorganize('ar-1') diff --git a/tests/test_reorganize_queue.py b/tests/test_reorganize_queue.py new file mode 100644 index 00000000..8e2707b4 --- /dev/null +++ b/tests/test_reorganize_queue.py @@ -0,0 +1,479 @@ +"""Tests for `core.reorganize_queue.ReorganizeQueue`. + +Contract this test file pins: + +1. **Dedupe on enqueue** — re-submitting an album that's already queued or + running returns ``{'queued': False, 'reason': 'already_queued'}`` and + the existing queue_id, never a duplicate. +2. **FIFO order** — the worker drains items in submission order. +3. **Per-item source preserved** — the source string the user picked at + enqueue time is what the runner sees, even when multiple items with + different sources are interleaved. +4. **Continue on failure** — a runner that raises (or one whose summary + reports a non-completed status) marks that item failed and the + worker moves to the next item, it does not stall. +5. **Cancel queued** — items in `queued` state can be dropped before + they reach the runner. +6. **Cancel running rejected** — the currently-running item can NOT be + cancelled, the API returns `running_cant_cancel`. +7. **Clear queued** — bulk-cancels all `queued` items at once, leaves + the running item alone. +8. **Snapshot shape** — `active`, `queued`, `recent`, and `totals` keys + are always present and reflect the current state. +9. **update_active_progress** — live progress fields propagate onto the + running item (and only the running item). +10. **Setting runner late** — items enqueued before `set_runner()` was + called still get processed once the runner shows up. +""" + +import threading +import time + +import pytest + +from core.reorganize_queue import ReorganizeQueue, QueueItem + + +# --- helpers --------------------------------------------------------------- + + +def _make_runner(record, *, raise_on=None, summary_factory=None, + block_event=None, runtime=0.0): + """Build a runner closure that records what it was called with. + + Args: + record: list to append `(queue_id, source)` to per call. + raise_on: queue_id (or set of queue_ids) for which the runner + should raise — used to test continue-on-failure. + summary_factory: optional callable `(item) -> summary dict` to + override the default `{'status': 'completed', ...}` shape. + block_event: optional `threading.Event` the runner blocks on + before returning — used to keep an item in 'running' state + while the test pokes at it. + runtime: seconds the runner sleeps before returning. + """ + raise_set = set() + if isinstance(raise_on, str): + raise_set = {raise_on} + elif raise_on: + raise_set = set(raise_on) + + def runner(item): + record.append((item.queue_id, item.source)) + if block_event is not None: + block_event.wait(timeout=2.0) + if runtime: + time.sleep(runtime) + if item.queue_id in raise_set: + raise RuntimeError(f"Simulated failure for {item.queue_id}") + if summary_factory is not None: + return summary_factory(item) + return { + 'status': 'completed', + 'source': item.source or 'spotify', + 'total': 1, + 'moved': 1, + 'skipped': 0, + 'failed': 0, + 'errors': [], + } + return runner + + +def _enqueue(queue, *, album_id, source=None, title=None, artist='Aerosmith'): + return queue.enqueue( + album_id=album_id, + album_title=title or f"Album {album_id}", + artist_id='artist-1', + artist_name=artist, + source=source, + ) + + +def _wait_for(predicate, timeout=2.0, interval=0.02): + """Poll until predicate() is truthy or timeout elapses.""" + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + return False + + +@pytest.fixture +def queue(): + q = ReorganizeQueue() + yield q + q.stop() + + +# --- tests ----------------------------------------------------------------- + + +def test_enqueue_returns_queued_with_position(queue): + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + r1 = _enqueue(queue, album_id='alb-1') + # Wait for the worker to actually pick up alb-1 so r2 lands while + # alb-1 is running, not while it's still queued — otherwise the + # position number depends on thread-scheduling timing. + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + r2 = _enqueue(queue, album_id='alb-2') + assert r1['queued'] is True + assert r1['position'] == 1 + assert r2['queued'] is True + assert r2['position'] == 1 + block.set() + + +def test_enqueue_same_album_dedupes(queue): + queue.set_runner(_make_runner([], block_event=threading.Event())) + r1 = _enqueue(queue, album_id='alb-1', source='spotify') + r2 = _enqueue(queue, album_id='alb-1', source='deezer') # different source + assert r1['queued'] is True + assert r2['queued'] is False + assert r2['reason'] == 'already_queued' + assert r2['queue_id'] == r1['queue_id'] + + +def test_dedupe_releases_after_completion(queue): + """Once an item finishes (done/failed/cancelled), the same album_id + can be re-enqueued. Otherwise users couldn't retry after a fix.""" + record = [] + queue.set_runner(_make_runner(record)) + r1 = _enqueue(queue, album_id='alb-1') + assert _wait_for(lambda: any(r[0] == r1['queue_id'] for r in record)) + # Wait for the item to flip into the recent bucket. + assert _wait_for(lambda: queue.snapshot()['active'] is None) + r2 = _enqueue(queue, album_id='alb-1') + assert r2['queued'] is True + assert r2['queue_id'] != r1['queue_id'] + + +def test_fifo_order(queue): + record = [] + queue.set_runner(_make_runner(record)) + ids = [_enqueue(queue, album_id=f'alb-{i}')['queue_id'] for i in range(5)] + assert _wait_for(lambda: len(record) == 5) + assert [r[0] for r in record] == ids + + +def test_per_item_source_preserved(queue): + record = [] + queue.set_runner(_make_runner(record)) + sources = ['spotify', 'deezer', 'itunes', None, 'discogs'] + for i, src in enumerate(sources): + _enqueue(queue, album_id=f'alb-{i}', source=src) + assert _wait_for(lambda: len(record) == len(sources)) + assert [r[1] for r in record] == sources + + +def test_continue_on_runner_exception(queue): + """A runner that raises must not stall the queue — the item is + marked failed and the next item runs.""" + record = [] + # Pre-allocate queue_ids by enqueuing first, then point the runner + # at the middle one. Block the runner so all three sit in the queue + # before any actually run. + block = threading.Event() + raise_target = {} + + def runner(item): + record.append((item.queue_id, item.source)) + block.wait(timeout=2.0) + if item.queue_id == raise_target.get('id'): + raise RuntimeError(f"Simulated failure for {item.queue_id}") + return { + 'status': 'completed', 'source': 'spotify', + 'total': 1, 'moved': 1, 'skipped': 0, 'failed': 0, 'errors': [], + } + + queue.set_runner(runner) + ids = [_enqueue(queue, album_id=f'alb-{i}')['queue_id'] for i in range(3)] + raise_target['id'] = ids[1] + block.set() + + assert _wait_for(lambda: len(record) == 3) + assert [r[0] for r in record] == ids + + assert _wait_for(lambda: queue.snapshot()['active'] is None) + snap = queue.snapshot() + recent_by_id = {r['queue_id']: r for r in snap['recent']} + assert recent_by_id[ids[0]]['status'] == 'done' + assert recent_by_id[ids[1]]['status'] == 'failed' + assert recent_by_id[ids[2]]['status'] == 'done' + + +def test_failed_status_when_runner_reports_failed_tracks(queue): + """A summary with ``failed > 0`` should mark the queue item as + 'failed' even if the runner returned normally.""" + queue.set_runner(_make_runner([], summary_factory=lambda item: { + 'status': 'completed', + 'source': 'spotify', + 'total': 5, + 'moved': 4, + 'skipped': 0, + 'failed': 1, + 'errors': [{'track_id': 't-1', 'title': 'X', 'error': 'boom'}], + })) + qid = _enqueue(queue, album_id='alb-1')['queue_id'] + # Wait for the item to land in `recent` (active is None both before + # the worker picks up the item and after it's done — only the + # presence in recent is unambiguous). + assert _wait_for(lambda: any(r['queue_id'] == qid for r in queue.snapshot()['recent'])) + snap = queue.snapshot() + item = next(i for i in snap['recent'] if i['queue_id'] == qid) + assert item['status'] == 'failed' + assert item['moved'] == 4 + assert item['failed'] == 1 + assert item['error'] == 'boom' + + +def test_failed_status_when_runner_reports_non_completed_status(queue): + """``status='no_source_id'`` and friends are setup-failures — they + leave failed=0 but the item is still NOT a success.""" + queue.set_runner(_make_runner([], summary_factory=lambda item: { + 'status': 'no_source_id', + 'source': None, + 'total': 0, + 'moved': 0, + 'skipped': 0, + 'failed': 0, + 'errors': [], + })) + qid = _enqueue(queue, album_id='alb-1')['queue_id'] + assert _wait_for(lambda: any(r['queue_id'] == qid for r in queue.snapshot()['recent'])) + snap = queue.snapshot() + item = next(r for r in snap['recent'] if r['queue_id'] == qid) + assert item['status'] == 'failed' + assert item['result_status'] == 'no_source_id' + + +def test_cancel_queued_item(queue): + """Cancel BEFORE the worker reaches the item drops it cleanly.""" + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + first = _enqueue(queue, album_id='alb-1')['queue_id'] # gets pulled to running, blocks + second = _enqueue(queue, album_id='alb-2')['queue_id'] # sits in queued + + # Wait for first to be running so we know the worker is parked on it. + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + + result = queue.cancel(second) + assert result['cancelled'] is True + + snap = queue.snapshot() + assert all(i['queue_id'] != second for i in snap['queued']) + # And the cancelled one shows up in recent with status 'cancelled'. + assert any(i['queue_id'] == second and i['status'] == 'cancelled' for i in snap['recent']) + + block.set() # release the running item + + +def test_cancel_running_rejected(queue): + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + qid = _enqueue(queue, album_id='alb-1')['queue_id'] + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + + result = queue.cancel(qid) + assert result['cancelled'] is False + assert result['reason'] == 'running_cant_cancel' + block.set() + + +def test_cancel_unknown_id(queue): + result = queue.cancel('does-not-exist') + assert result['cancelled'] is False + assert result['reason'] == 'not_found' + + +def test_clear_queued_bulk_cancel(queue): + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + _enqueue(queue, album_id='alb-1') # running, blocked + queued_ids = [_enqueue(queue, album_id=f'alb-{i}')['queue_id'] for i in range(2, 6)] + + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + assert _wait_for(lambda: len(queue.snapshot()['queued']) == 4) + + cancelled = queue.clear_queued() + assert cancelled == 4 + + snap = queue.snapshot() + assert len(snap['queued']) == 0 + # Running item is untouched. + assert snap['active'] is not None + cancelled_in_recent = [i for i in snap['recent'] if i['status'] == 'cancelled'] + assert {i['queue_id'] for i in cancelled_in_recent} == set(queued_ids) + block.set() + + +def test_snapshot_shape(queue): + snap = queue.snapshot() + assert set(snap.keys()) == {'active', 'queued', 'recent', 'totals'} + assert set(snap['totals'].keys()) >= {'queued', 'running', 'done', 'failed', 'cancelled'} + assert snap['active'] is None + assert snap['queued'] == [] + assert snap['recent'] == [] + + +def test_update_active_progress_only_targets_running(queue): + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + qid = _enqueue(queue, album_id='alb-1')['queue_id'] + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + + queue.update_active_progress( + queue_id=qid, + current_track='Dream On', + total=8, + processed=3, + moved=3, + skipped=0, + failed=0, + ) + snap = queue.snapshot() + assert snap['active']['current_track'] == 'Dream On' + assert snap['active']['progress_total'] == 8 + assert snap['active']['progress_processed'] == 3 + assert snap['active']['moved'] == 3 + block.set() + + +def test_update_progress_for_unknown_id_is_noop(queue): + """Calling update_active_progress for an item that isn't running + must not raise, must not corrupt other items.""" + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + qid = _enqueue(queue, album_id='alb-1')['queue_id'] + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + + queue.update_active_progress(queue_id='not-a-real-id', current_track='X', total=999) + snap = queue.snapshot() + assert snap['active']['queue_id'] == qid + assert snap['active']['progress_total'] == 0 # unchanged + block.set() + + +def test_enqueue_many_tallies_enqueued_and_dedupes(queue): + """Bulk enqueue returns ``{enqueued, already_queued, total}`` so + the route handler doesn't have to count itself. Re-enqueuing the + same album-id twice in the same batch dedupes.""" + block = threading.Event() + queue.set_runner(_make_runner([], block_event=block)) + + # Pre-existing item — should appear as already_queued. + queue.enqueue(album_id='alb-existing', album_title='X', + artist_id='ar-1', artist_name='A', source=None) + # Wait for it to be running so the dedupe path triggers. + assert _wait_for(lambda: queue.snapshot()['active'] is not None) + + items = [ + {'album_id': 'alb-existing', 'album_title': 'X', 'artist_id': 'ar-1', 'artist_name': 'A'}, + {'album_id': 'alb-new-1', 'album_title': 'Y', 'artist_id': 'ar-1', 'artist_name': 'A'}, + {'album_id': 'alb-new-2', 'album_title': 'Z', 'artist_id': 'ar-1', 'artist_name': 'A'}, + ] + result = queue.enqueue_many(items) + assert result == {'enqueued': 2, 'already_queued': 1, 'total': 3} + block.set() + + +def test_enqueue_many_carries_source_per_item(queue): + """Each dict's ``source`` is honoured independently — the bulk + helper doesn't collapse them to one value.""" + record = [] + queue.set_runner(_make_runner(record)) + items = [ + {'album_id': 'a', 'album_title': 'A', 'artist_id': 'x', 'artist_name': 'X', 'source': 'spotify'}, + {'album_id': 'b', 'album_title': 'B', 'artist_id': 'x', 'artist_name': 'X', 'source': 'deezer'}, + {'album_id': 'c', 'album_title': 'C', 'artist_id': 'x', 'artist_name': 'X', 'source': None}, + ] + queue.enqueue_many(items) + assert _wait_for(lambda: len(record) == 3) + assert [r[1] for r in record] == ['spotify', 'deezer', None] + + +def test_enqueue_many_handles_empty_list(queue): + queue.set_runner(_make_runner([])) + assert queue.enqueue_many([]) == {'enqueued': 0, 'already_queued': 0, 'total': 0} + + +def test_enqueue_many_dedupes_batch_internal_duplicates(queue): + """Same album_id appearing twice in the same bulk request must be + deduped against each other — not just against pre-existing items. + Regression for the race where a fast runner finishes the first copy + before the loop reaches the second, letting both slip through.""" + record = [] + queue.set_runner(_make_runner(record)) + items = [ + {'album_id': 'alb-x', 'album_title': 'X', 'artist_id': 'ar-1', 'artist_name': 'A'}, + {'album_id': 'alb-y', 'album_title': 'Y', 'artist_id': 'ar-1', 'artist_name': 'A'}, + {'album_id': 'alb-x', 'album_title': 'X (dup)', 'artist_id': 'ar-1', 'artist_name': 'A'}, + ] + result = queue.enqueue_many(items) + assert result == {'enqueued': 2, 'already_queued': 1, 'total': 3} + # Wait for the queue to drain, then give the worker a moment to + # try (and fail) to pick a phantom third item. If the dedupe leaked, + # a third runner call would land here. + assert _wait_for(lambda: queue.snapshot()['active'] is None and not queue.snapshot()['queued']) + time.sleep(0.05) + assert len(record) == 2 + + +def test_cancel_and_run_are_mutually_exclusive(queue): + """Regression for kettui's ``_next_queued() → status flip`` race: + a successfully-cancelled item must NEVER have its runner invoked. + With the old non-atomic pick + flip, cancel could land between + the worker's pick and its flip-to-running, leaving the item + marked 'cancelled' but the worker still runs it. + + Hammers many enqueue-then-immediately-cancel pairs to exercise the + race window. After draining, every queue_id whose cancel returned + ``cancelled: True`` must NOT appear in the runner's record.""" + runner_called: set = set() + runner_lock = threading.Lock() + + def runner(item): + with runner_lock: + runner_called.add(item.queue_id) + # Slight runtime widens the window where overlapping cancels + # could (incorrectly) fire on a running item. + time.sleep(0.002) + return { + 'status': 'completed', 'source': 'spotify', + 'total': 1, 'moved': 1, 'skipped': 0, 'failed': 0, 'errors': [], + } + + queue.set_runner(runner) + + successful_cancels: set = set() + for i in range(50): + r = _enqueue(queue, album_id=f'alb-race-{i}') + # Immediately try to cancel — half will land while item is still + # 'queued', half will land after worker has flipped to 'running'. + if queue.cancel(r['queue_id'])['cancelled']: + successful_cancels.add(r['queue_id']) + + assert _wait_for( + lambda: queue.snapshot()['active'] is None and not queue.snapshot()['queued'], + timeout=5.0, + ) + + leaked = successful_cancels & runner_called + assert not leaked, f"Runner ran for cancelled items: {leaked}" + + +def test_no_runner_marks_item_failed(queue): + """If the worker pulls an item but no runner has been set, the item + must be marked failed (not silently dropped). In practice + web_server.py wires the runner at module load before any request + can land, so this is a defensive-failure path more than a real + one — but the failure mode must be loud.""" + queue.set_runner(None) + qid = _enqueue(queue, album_id='alb-orphan')['queue_id'] + assert _wait_for(lambda: any(r['queue_id'] == qid for r in queue.snapshot()['recent'])) + snap = queue.snapshot() + failed = next(i for i in snap['recent'] if i['queue_id'] == qid) + assert failed['status'] == 'failed' + assert 'runner' in (failed['error'] or '').lower() diff --git a/tests/test_reorganize_runner.py b/tests/test_reorganize_runner.py new file mode 100644 index 00000000..f38bffd1 --- /dev/null +++ b/tests/test_reorganize_runner.py @@ -0,0 +1,235 @@ +"""Tests for `core.reorganize_runner.build_runner`. + +Contract this test file pins: + +1. **Runner is a closure** — calling `build_runner` returns a callable + that takes a queue item and returns a summary dict matching + `reorganize_album`'s shape. +2. **Config is read per-run, not at factory time** — changing the + download/transfer path between runs is honoured. Web server config + should never need a restart for this to take effect. +3. **Setup failure surfaces a clean summary** — if the staging dir + cannot be created, the runner returns `status='setup_failed'` + instead of raising (so the queue marks the item failed cleanly). +4. **Progress callbacks fan out into the queue** — the runner wires + `reorganize_album`'s `on_progress` to `update_active_progress` on + the live singleton queue, so the status panel sees per-track state. +5. **Dependencies are injected, not imported** — the factory takes + every external dependency as a callable so tests can run without + spinning up Flask, the DB, or the post-process pipeline. +""" + +import sys +import types +from unittest.mock import MagicMock + +import pytest + + +# Stub config.settings so importing core.reorganize_runner -> core.library_reorganize doesn't blow up +if "config.settings" not in sys.modules: + config_pkg = types.ModuleType("config") + settings_mod = types.ModuleType("config.settings") + + class _DummyConfigManager: + def get(self, key, default=None): + return default + + def get_active_media_server(self): + return "primary" + + settings_mod.config_manager = _DummyConfigManager() + config_pkg.settings = settings_mod + sys.modules["config"] = config_pkg + sys.modules["config.settings"] = settings_mod + +if "spotipy" not in sys.modules: + spotipy = types.ModuleType("spotipy") + spotipy.Spotify = object + oauth2 = types.ModuleType("spotipy.oauth2") + oauth2.SpotifyOAuth = object + oauth2.SpotifyClientCredentials = object + spotipy.oauth2 = oauth2 + sys.modules["spotipy"] = spotipy + sys.modules["spotipy.oauth2"] = oauth2 + + +from core.reorganize_runner import build_runner # noqa: E402 + + +@pytest.fixture(autouse=True) +def reset_queue_singleton(): + """Each test gets a fresh queue singleton so update_active_progress + in one test doesn't leak into another.""" + from core.reorganize_queue import reset_queue_for_tests + reset_queue_for_tests() + yield + reset_queue_for_tests() + + +def _make_item(*, queue_id='qid-1', album_id='alb-1', source=None): + """Mock queue item — only needs the fields the runner reads.""" + item = MagicMock() + item.queue_id = queue_id + item.album_id = album_id + item.source = source + return item + + +def _build(monkeypatch, *, download_path_fn, transfer_path_fn, + reorganize_album_fn, get_database=lambda: object()): + """Helper: stub out the heavy reorganize_album call so we can test + the wiring without a real DB / post-process pipeline.""" + # Patch the import inside reorganize_runner.build_runner. + import core.reorganize_runner as mod + monkeypatch.setattr( + 'core.library_reorganize.reorganize_album', + reorganize_album_fn, + raising=True, + ) + + return build_runner( + get_database=get_database, + resolve_file_path_fn=lambda p: p, + post_process_fn=lambda *a, **k: None, + cleanup_empty_directories_fn=lambda *a, **k: None, + is_shutting_down_fn=lambda: False, + get_download_path=download_path_fn, + get_transfer_path=transfer_path_fn, + ) + + +def test_runner_invokes_reorganize_album_with_injected_deps(monkeypatch, tmp_path): + captured = {} + + def fake_reorganize_album(**kwargs): + captured.update(kwargs) + return { + 'status': 'completed', 'source': 'spotify', + 'total': 1, 'moved': 1, 'skipped': 0, 'failed': 0, 'errors': [], + } + + runner = _build( + monkeypatch, + download_path_fn=lambda: str(tmp_path), + transfer_path_fn=lambda: str(tmp_path / 'transfer'), + reorganize_album_fn=fake_reorganize_album, + ) + item = _make_item(album_id='alb-X', source='deezer') + summary = runner(item) + + assert summary['status'] == 'completed' + assert captured['album_id'] == 'alb-X' + assert captured['primary_source'] == 'deezer' + assert captured['strict_source'] is True + # staging_root is download_path / ssync_staging + assert captured['staging_root'].endswith('ssync_staging') + assert callable(captured['on_progress']) + assert callable(captured['stop_check']) + + +def test_runner_reads_config_per_call(monkeypatch, tmp_path): + """Path that the runner sees should reflect the value returned by + the path-resolver lambda AT call time — not at build_runner time. + This is the explicit fix for kettui-style "config change requires + server restart" feedback.""" + seen_staging_roots = [] + + def fake_reorganize_album(**kwargs): + seen_staging_roots.append(kwargs['staging_root']) + return { + 'status': 'completed', 'source': None, + 'total': 0, 'moved': 0, 'skipped': 0, 'failed': 0, 'errors': [], + } + + current_path = {'value': str(tmp_path / 'first')} + runner = _build( + monkeypatch, + download_path_fn=lambda: current_path['value'], + transfer_path_fn=lambda: '/tmp/transfer', + reorganize_album_fn=fake_reorganize_album, + ) + + runner(_make_item()) + current_path['value'] = str(tmp_path / 'second') + runner(_make_item()) + + assert len(seen_staging_roots) == 2 + assert 'first' in seen_staging_roots[0] + assert 'second' in seen_staging_roots[1] + + +def test_runner_returns_setup_failed_on_unwritable_path(monkeypatch, tmp_path): + """If the staging dir can't be created (permission denied, etc.), + the runner returns a clean ``setup_failed`` summary so the queue + marks the item failed without an unhandled exception.""" + def fake_reorganize_album(**kwargs): + pytest.fail("reorganize_album should not run when setup fails") + + # Point at a child of an existing FILE — makedirs will raise OSError. + blocking_file = tmp_path / 'blocker' + blocking_file.write_text('x') + + runner = _build( + monkeypatch, + download_path_fn=lambda: str(blocking_file), # makedirs fails here + transfer_path_fn=lambda: '/tmp/transfer', + reorganize_album_fn=fake_reorganize_album, + ) + summary = runner(_make_item()) + assert summary['status'] == 'setup_failed' + assert summary['errors'] + + +def test_runner_progress_callback_forwards_to_queue(monkeypatch, tmp_path): + """When reorganize_album fires its on_progress callback, the runner + must forward into the live queue's update_active_progress so the + status panel sees per-track updates.""" + from core.reorganize_queue import get_queue, ReorganizeQueue + import threading + + # Use a real queue that's blocked on a runner — gives us a known + # 'running' item to propagate progress into. + block = threading.Event() + + def fake_reorganize_album(*, on_progress, **kwargs): + # Simulate per-track progress emissions like the real + # orchestrator does. + on_progress({'current_track': 'Backseat Freestyle', 'total': 12, 'processed': 1}) + on_progress({'moved': 1, 'processed': 1}) + return { + 'status': 'completed', 'source': 'spotify', + 'total': 12, 'moved': 1, 'skipped': 0, 'failed': 0, 'errors': [], + } + + runner = _build( + monkeypatch, + download_path_fn=lambda: str(tmp_path), + transfer_path_fn=lambda: str(tmp_path / 'transfer'), + reorganize_album_fn=fake_reorganize_album, + ) + + # Wire our runner into the singleton queue and enqueue an item, so + # update_active_progress has a 'running' item to write into. + q = get_queue() + q.set_runner(runner) + enq = q.enqueue(album_id='alb-1', album_title='good kid', + artist_id='ar-1', artist_name='Kendrick Lamar', source=None) + + # Wait for the worker to finish (fake_reorganize_album is fast). + deadline_passes = 0 + import time + while deadline_passes < 50: + snap = q.snapshot() + if any(r['queue_id'] == enq['queue_id'] for r in snap['recent']): + break + time.sleep(0.02) + deadline_passes += 1 + + snap = q.snapshot() + finished = next(r for r in snap['recent'] if r['queue_id'] == enq['queue_id']) + assert finished['status'] == 'done' + assert finished['moved'] == 1 + # The progress fan-out happened *while* the item was running. The + # final snapshot shows the worker-set values — what we're really + # asserting is that progress callbacks didn't raise. diff --git a/web_server.py b/web_server.py index 97f77879..7ba8afc0 100644 --- a/web_server.py +++ b/web_server.py @@ -169,6 +169,28 @@ app = Flask( ) app.config['TEMPLATES_AUTO_RELOAD'] = DEV_STATIC_NO_CACHE app.jinja_env.auto_reload = DEV_STATIC_NO_CACHE +# Force static assets (library.js / style.css / etc.) to revalidate +# with ETag on every load instead of Flask's default 12-hour browser +# cache. Updates ship live without users having to clear cache. +# Modern browsers still serve 304 Not Modified when the file hasn't +# changed, so the cost per asset per reload is just a header round-trip. +app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0 + + +# Cache-bust query string for static assets — appended to every +# url_for('static', ...) URL via the context processor below. Computed +# once per process start so each server restart invalidates the +# browser's cached copy of every JS/CSS file. This is the surefire +# fix for "user has stale JS even after Ctrl+Shift+R" — the URL +# itself changes, so the browser cannot reuse a previously-cached +# response no matter what its Cache-Control header said. +import time as _cache_bust_time +_STATIC_CACHE_BUST = str(int(_cache_bust_time.time())) + + +@app.context_processor +def _inject_static_cache_bust(): + return {'static_v': _STATIC_CACHE_BUST} # --- Flask Session Setup (for multi-profile support) --- import secrets as _secrets @@ -13592,19 +13614,12 @@ def get_tracks_replaygain_batch_status(): return jsonify(state) -# ── Reorganize Album Files endpoint ── - -_reorganize_state = { - 'status': 'idle', - 'total': 0, - 'processed': 0, - 'moved': 0, - 'skipped': 0, - 'failed': 0, - 'current_track': '', - 'errors': [], -} -_reorganize_lock = threading.Lock() +# ── Reorganize Album Files endpoints ── +# +# Reorganize requests flow through ``core.reorganize_queue`` — a FIFO +# queue with a single background worker. The endpoints here are thin +# enqueue / snapshot / cancel wrappers; the heavy lifting is in +# :mod:`core.library_reorganize`. @app.route('/api/library/reorganize/sources', methods=['GET']) @@ -13672,113 +13687,143 @@ def reorganize_album_preview(album_id): @app.route('/api/library/album//reorganize', methods=['POST']) def reorganize_album_files(album_id): - """Re-route an album's existing files through the same post-processing - pipeline downloads use. Implementation lives in - :mod:`core.library_reorganize` to keep this monolith from growing. - The request body's ``template`` (if any) is ignored — post-processing - always uses the configured template, matching the download path. + """Enqueue an album for reorganize. Returns immediately — the + queue worker processes items FIFO. Repeat clicks for an album + that's already queued or running are deduped (returns + ``{queued: false, reason: 'already_queued'}``). - Optional body param ``source``: when provided, only that metadata - source is used (no fallback chain). Matches the per-album source - picker in the reorganize modal.""" + Body params: + source (optional): per-album source pick (Spotify / iTunes / + Deezer / Discogs / Hydrabase). When omitted, the + orchestrator uses the configured primary with fallback. + """ try: + from core.reorganize_queue import get_queue data = request.get_json() or {} chosen_source = data.get('source') or None - with _reorganize_lock: - if _reorganize_state['status'] == 'running': - return jsonify({"success": False, "error": "A reorganization is already in progress"}), 409 - _reorganize_state['status'] = 'running' - _reorganize_state.update({ - 'total': 0, 'processed': 0, 'moved': 0, 'skipped': 0, - 'failed': 0, 'current_track': '', 'errors': [], - # Set after the run from the orchestrator's summary so the - # frontend can distinguish 'completed' from 'no_source_id' - # / 'no_album' / 'no_tracks' / 'setup_failed' (otherwise - # zero-failure skips look green to the user). - 'result_status': None, 'result_source': None, - }) - download_dir = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) - staging_root = os.path.join(download_dir, 'ssync_staging') - try: - os.makedirs(staging_root, exist_ok=True) - except OSError: - pass - transfer_dir = docker_resolve_path(config_manager.get('soulseek.transfer_path', './Transfer')) - - def _on_progress(updates): - with _reorganize_lock: - _reorganize_state.update(updates) - - def _update_track_path(track_id, new_path): - try: - _db = get_database() - with _db._get_connection() as _conn: - _conn.execute( - "UPDATE tracks SET file_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", - (new_path, str(track_id)), - ) - _conn.commit() - except Exception as _db_err: - logger.warning(f"[Reorganize] DB path update failed for {track_id}: {_db_err}") - - def _cleanup_empty(src_dir): - try: - _cleanup_empty_directories(transfer_dir, os.path.join(src_dir, '_')) - except Exception: - pass - - def _run(): - from core.library_reorganize import reorganize_album - try: - summary = reorganize_album( - album_id=album_id, - db=get_database(), - staging_root=staging_root, - resolve_file_path_fn=_resolve_library_file_path, - post_process_fn=_post_process_matched_download, - update_track_path_fn=_update_track_path, - cleanup_empty_dir_fn=_cleanup_empty, - transfer_dir=transfer_dir, - on_progress=_on_progress, - primary_source=chosen_source, - strict_source=bool(chosen_source), - stop_check=lambda: bool(IS_SHUTTING_DOWN), - ) - logger.info( - f"[Reorganize] Album {album_id} {summary['status']} " - f"(source={summary.get('source')}, moved={summary['moved']}, " - f"skipped={summary['skipped']}, failed={summary['failed']})" - ) - with _reorganize_lock: - _reorganize_state['result_status'] = summary.get('status') - _reorganize_state['result_source'] = summary.get('source') - except Exception as run_err: - logger.error(f"[Reorganize] Background error: {run_err}", exc_info=True) - with _reorganize_lock: - _reorganize_state['result_status'] = 'error' - finally: - with _reorganize_lock: - _reorganize_state['status'] = 'done' - _reorganize_state['current_track'] = '' - - threading.Thread(target=_run, daemon=True, name="ReorganizeAlbum").start() - return jsonify({"success": True, "message": "Reorganization started"}) + # Capture display fields at enqueue time so the status panel + # can render them without a DB lookup later. + meta = get_database().get_album_display_meta(album_id) + if meta is None: + return jsonify({"success": False, "error": "Album not found"}), 404 + result = get_queue().enqueue( + album_id=str(album_id), + album_title=meta['album_title'], + artist_id=meta['artist_id'], + artist_name=meta['artist_name'], + source=chosen_source, + ) + return jsonify({"success": True, **result}) except Exception as e: - logger.error(f"Reorganize error: {e}") - with _reorganize_lock: - _reorganize_state['status'] = 'idle' + logger.error(f"Reorganize enqueue error: {e}", exc_info=True) return jsonify({"success": False, "error": str(e)}), 500 -@app.route('/api/library/album/reorganize/status', methods=['GET']) -def get_reorganize_status(): - """Poll the status of a running reorganization.""" - with _reorganize_lock: - state = dict(_reorganize_state) - state['errors'] = list(_reorganize_state['errors']) - return jsonify(state) +@app.route('/api/library/artist//reorganize-all', methods=['POST']) +def reorganize_all_artist_albums(artist_id): + """Enqueue every album for an artist. Replaces the old frontend + bulk-loop. Each album becomes its own queue item, processed FIFO. + Albums already queued or running are deduped silently. + + Body params: + source (optional): same pick applied to every album. Per-album + overrides aren't supported here — use the per-album modal + for that. + """ + try: + from core.reorganize_queue import get_queue + data = request.get_json() or {} + chosen_source = data.get('source') or None + + albums = get_database().get_artist_albums_for_reorganize(artist_id) + if not albums: + return jsonify({"success": False, "error": "No albums found for this artist"}), 404 + + # Apply the user's chosen source to every album, then hand off + # to the queue's bulk-enqueue helper which owns the loop+tally. + for album in albums: + album['source'] = chosen_source + result = get_queue().enqueue_many(albums) + + return jsonify({ + "success": True, + "enqueued": result['enqueued'], + "already_queued": result['already_queued'], + "total_albums": result['total'], + }) + except Exception as e: + logger.error(f"Reorganize-all enqueue error: {e}", exc_info=True) + return jsonify({"success": False, "error": str(e)}), 500 + + +@app.route('/api/library/reorganize/queue', methods=['GET']) +def reorganize_queue_snapshot(): + """Snapshot of the reorganize queue — what's running, what's queued, + recent completions. Polled by the status panel.""" + try: + from core.reorganize_queue import get_queue + return jsonify({"success": True, **get_queue().snapshot()}) + except Exception as e: + logger.error(f"Reorganize queue snapshot error: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + +@app.route('/api/library/reorganize/queue//cancel', methods=['POST']) +def reorganize_queue_cancel(queue_id): + """Cancel a queued item (running items can't be cleanly cancelled — + see the queue module's design rules).""" + try: + from core.reorganize_queue import get_queue + result = get_queue().cancel(queue_id) + status_code = 200 if result.get('cancelled') else 409 + return jsonify({"success": result.get('cancelled', False), **result}), status_code + except Exception as e: + logger.error(f"Reorganize cancel error: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + +@app.route('/api/library/reorganize/queue/clear', methods=['POST']) +def reorganize_queue_clear(): + """Cancel all queued items at once (the running item continues).""" + try: + from core.reorganize_queue import get_queue + cancelled = get_queue().clear_queued() + return jsonify({"success": True, "cancelled": cancelled}) + except Exception as e: + logger.error(f"Reorganize clear error: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + +# Wire the reorganize queue worker to its runner at module load. The +# runner factory lives in :mod:`core.reorganize_runner` so this monolith +# stays small. Config (paths) is read **per run** inside the closure, +# so changing your download path in Settings takes effect on the next +# reorganize without a server restart. +# +# The injected callables are wrapped in lambdas because the underlying +# helpers (``_resolve_library_file_path`` etc.) are defined LATER in +# this file. Lambdas defer name resolution to call time so module-load +# import order works regardless of definition order. +try: + from core.reorganize_queue import get_queue as _get_reorganize_queue + from core.reorganize_runner import build_runner as _build_reorganize_runner + _get_reorganize_queue().set_runner(_build_reorganize_runner( + get_database=get_database, + resolve_file_path_fn=lambda p: _resolve_library_file_path(p), + post_process_fn=lambda *a, **kw: _post_process_matched_download(*a, **kw), + cleanup_empty_directories_fn=lambda *a, **kw: _cleanup_empty_directories(*a, **kw), + is_shutting_down_fn=lambda: bool(IS_SHUTTING_DOWN), + get_download_path=lambda: docker_resolve_path( + config_manager.get('soulseek.download_path', './downloads') + ), + get_transfer_path=lambda: docker_resolve_path( + config_manager.get('soulseek.transfer_path', './Transfer') + ), + )) +except Exception as _runner_init_err: + logger.error(f"Failed to register reorganize queue runner: {_runner_init_err}") # ── Library Issues endpoints ── @@ -22732,6 +22777,33 @@ def get_version_info(): "title": "What's New in SoulSync", "subtitle": f"Version {SOULSYNC_VERSION} — Latest Changes", "sections": [ + { + "title": "Reorganize Queue: Race-Condition Hardening (kettui Review)", + "description": "Three concurrency / dedupe issues kettui caught in his review of PR #377, plus two related polish items from the same pass.", + "features": [ + "• Worker pick + status flip is now atomic — fixes a window where a cancel() landing between 'pick next queued' and 'flip to running' could mark an item cancelled but the worker still ran it", + "• Replaced the lock + wakeup-event pair with a single threading.Condition so newly-queued items can't sleep up to 60s waiting for the next wakeup tick (the old pair had an empty-check / clear-event race)", + "• enqueue_many now holds the queue lock for the whole batch and tracks a per-batch seen set, so duplicate album_ids inside one bulk call are deduped against each other (not just against pre-existing items)", + "• Reorganize-preview Apply button no longer gets stuck disabled when an early return / network error skipped the re-enable line — moved into a finally", + "• DB helpers get_album_display_meta and get_artist_albums_for_reorganize now let exceptions bubble instead of swallowing them as 'not found' / empty list — a real DB outage now surfaces as a 500 to the user instead of looking like a missing album", + ], + }, + { + "title": "Reorganize Queue with Live Status Panel", + "description": "Reorganizing albums is no longer a foreground operation that locks the page. Click → enqueue → keep working. A status panel surfaces live progress.", + "features": [ + "• Per-album Reorganize and Reorganize All both enqueue into a single FIFO queue with a backend worker that drains one item at a time", + "• Buttons stay clickable — spam-clicking the same album silently dedupes (returns 'already queued' instead of 409-ing)", + "• Status panel at the top of the artist actions bar shows: active item (progress bar, current track, moved/skipped/failed counts), queued count, and recently-finished items with success/warning indicators", + "• Click the panel to expand: full queue list with per-item cancel buttons; running item can't be cancelled mid-flight (Python threads aren't cleanly killable, post-process spawns subprocesses)", + "• 'Cancel All' button drops every queued item at once — the running one continues", + "• Items belonging to a different artist than the page you're on are flagged with the artist name so cross-artist progress is obvious", + "• Each queued item carries its own metadata source pick (Spotify / iTunes / Deezer / Discogs / Hydrabase) — switching modal selections per album works", + "• 'Reorganize All' is now one backend call instead of N JS-driven calls — the loop runs server-side and is much faster", + "• Continue-on-failure: a single failed album never stalls the queue; the worker logs and moves on", + "• Retired the old single-slot reorganize state endpoint plus the polling loops that depended on it", + ], + }, { "title": "Fix Wrong-Artist Tracks Silently Downloading", "description": "A critical bug where searching for a track could silently download a completely different artist's song with the same name", diff --git a/webui/index.html b/webui/index.html index c73140d7..f264044a 100644 --- a/webui/index.html +++ b/webui/index.html @@ -5,10 +5,10 @@ SoulSync - Music Sync & Manager - - - - + + + + @@ -7860,27 +7860,27 @@ - + - + - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + +