speedup on tagging

frontend/src/api.rs

@@ -176,6 +176,56 @@ pub async fn add_album(
     post_json(&format!("{BASE}/albums"), &body).await
 }
 
+pub async fn watch_track(
+    artist: Option<&str>,
+    title: &str,
+    mbid: &str,
+) -> Result<WatchTrackResponse, ApiError> {
+    let body = serde_json::json!({"artist": artist, "title": title, "mbid": mbid}).to_string();
+    post_json(&format!("{BASE}/tracks/watch"), &body).await
+}
+
+pub async fn unwatch_artist(id: i32) -> Result<serde_json::Value, ApiError> {
+    let resp = Request::delete(&format!("{BASE}/artists/{id}/watch"))
+        .send()
+        .await
+        .map_err(|e| ApiError(e.to_string()))?;
+    if !resp.ok() {
+        return Err(ApiError(format!("HTTP {}", resp.status())));
+    }
+    resp.json().await.map_err(|e| ApiError(e.to_string()))
+}
+
+pub async fn unwatch_album(mbid: &str) -> Result<serde_json::Value, ApiError> {
+    let resp = Request::delete(&format!("{BASE}/albums/{mbid}/watch"))
+        .send()
+        .await
+        .map_err(|e| ApiError(e.to_string()))?;
+    if !resp.ok() {
+        return Err(ApiError(format!("HTTP {}", resp.status())));
+    }
+    resp.json().await.map_err(|e| ApiError(e.to_string()))
+}
+
+pub async fn unwatch_track(mbid: &str) -> Result<serde_json::Value, ApiError> {
+    let body = serde_json::json!({"mbid": mbid}).to_string();
+    let resp = Request::delete(&format!("{BASE}/tracks/watch"))
+        .header("Content-Type", "application/json")
+        .body(&body)
+        .map_err(|e| ApiError(e.to_string()))?
+        .send()
+        .await
+        .map_err(|e| ApiError(e.to_string()))?;
+    if !resp.ok() {
+        return Err(ApiError(format!("HTTP {}", resp.status())));
+    }
+    resp.json().await.map_err(|e| ApiError(e.to_string()))
+}
+
+pub async fn delete_artist(id: i32) -> Result<(), ApiError> {
+    delete(&format!("{BASE}/artists/{id}")).await
+}
+
 // --- Downloads ---
 pub async fn get_downloads(status: Option<&str>) -> Result<Vec<DownloadItem>, ApiError> {
     let mut url = format!("{BASE}/downloads/queue");

frontend/src/components/navbar.rs

@@ -13,6 +13,7 @@ pub struct Props {
 #[function_component(Navbar)]
 pub fn navbar(props: &Props) -> Html {
     let route = use_route::<Route>();
+    let sidebar_open = use_state(|| false);
 
     let link = |to: Route, label: &str| {
         let active = route.as_ref() == Some(&to);
@@ -24,16 +25,56 @@ pub fn navbar(props: &Props) -> Html {
 
     let on_logout = {
         let cb = props.on_logout.clone();
+        let sidebar_open = sidebar_open.clone();
         Callback::from(move |e: MouseEvent| {
             e.prevent_default();
+            sidebar_open.set(false);
             cb.emit(());
         })
     };
 
+    let toggle = {
+        let sidebar_open = sidebar_open.clone();
+        Callback::from(move |_: MouseEvent| {
+            sidebar_open.set(!*sidebar_open);
+        })
+    };
+
+    let close_overlay = {
+        let sidebar_open = sidebar_open.clone();
+        Callback::from(move |_: MouseEvent| {
+            sidebar_open.set(false);
+        })
+    };
+
+    // Close sidebar when any nav link is clicked
+    let on_nav_click = {
+        let sidebar_open = sidebar_open.clone();
+        Callback::from(move |_: MouseEvent| {
+            sidebar_open.set(false);
+        })
+    };
+
+    let sidebar_class = if *sidebar_open {
+        "sidebar open"
+    } else {
+        "sidebar"
+    };
+    let overlay_class = if *sidebar_open {
+        "sidebar-overlay open"
+    } else {
+        "sidebar-overlay"
+    };
+
     html! {
-        <div class="sidebar">
+        <>
+        <button class="hamburger" onclick={toggle}>
+            { "\u{2630}" }
+        </button>
+        <div class={overlay_class} onclick={close_overlay}></div>
+        <div class={sidebar_class}>
             <h1>{ "Shanty" }</h1>
-            <nav>
+            <nav onclick={on_nav_click}>
                 { link(Route::Dashboard, "Dashboard") }
                 { link(Route::Search, "Search") }
                 { link(Route::Library, "Library") }
@@ -48,5 +89,6 @@ pub fn navbar(props: &Props) -> Html {
                 <a href="#" class="text-sm" onclick={on_logout}>{ "Logout" }</a>
             </div>
         </div>
+        </>
     }
 }

frontend/src/pages/album.rs

@@ -80,7 +80,7 @@ pub fn album_page(props: &Props) -> Html {
                         </tr>
                     </thead>
                     <tbody>
-                        { for d.tracks.iter().map(|t| {
+                        { for d.tracks.iter().enumerate().map(|(idx, t)| {
                             let duration = t.duration_ms
                                 .map(&fmt_duration)
                                 .unwrap_or_default();
@@ -88,6 +88,7 @@ pub fn album_page(props: &Props) -> Html {
                             let track_key = t.recording_mbid.clone();
                             let is_active = active_lyrics.as_ref() == Some(&track_key);
                             let cached = lyrics_cache.get(&track_key).cloned();
+                            let has_status = t.status.is_some();
 
                             let on_lyrics_click = {
                                 let active = active_lyrics.clone();
@@ -117,6 +118,52 @@ pub fn album_page(props: &Props) -> Html {
                                 })
                             };
 
+                            let on_watch_click = {
+                                let detail = detail.clone();
+                                let title = t.title.clone();
+                                let mbid = t.recording_mbid.clone();
+                                let artist = d.artist.clone();
+                                Callback::from(move |_: MouseEvent| {
+                                    let detail = detail.clone();
+                                    let title = title.clone();
+                                    let mbid = mbid.clone();
+                                    let artist = artist.clone();
+                                    let idx = idx;
+                                    wasm_bindgen_futures::spawn_local(async move {
+                                        if let Ok(resp) = api::watch_track(artist.as_deref(), &title, &mbid).await {
+                                            if let Some(ref d) = *detail {
+                                                let mut updated = d.clone();
+                                                if let Some(track) = updated.tracks.get_mut(idx) {
+                                                    track.status = Some(resp.status);
+                                                }
+                                                detail.set(Some(updated));
+                                            }
+                                        }
+                                    });
+                                })
+                            };
+
+                            let on_unwatch_click = {
+                                let detail = detail.clone();
+                                let mbid = t.recording_mbid.clone();
+                                Callback::from(move |_: MouseEvent| {
+                                    let detail = detail.clone();
+                                    let mbid = mbid.clone();
+                                    let idx = idx;
+                                    wasm_bindgen_futures::spawn_local(async move {
+                                        if api::unwatch_track(&mbid).await.is_ok() {
+                                            if let Some(ref d) = *detail {
+                                                let mut updated = d.clone();
+                                                if let Some(track) = updated.tracks.get_mut(idx) {
+                                                    track.status = None;
+                                                }
+                                                detail.set(Some(updated));
+                                            }
+                                        }
+                                    });
+                                })
+                            };
+
                             html! {
                                 <>
                                 <tr>
@@ -130,7 +177,18 @@ pub fn album_page(props: &Props) -> Html {
                                             <span class="text-muted text-sm">{ "\u{2014}" }</span>
                                         }
                                     </td>
-                                    <td>
+                                    <td class="actions">
+                                        if !has_status {
+                                            <button class="btn btn-sm"
+                                                onclick={on_watch_click}>
+                                                { "Watch" }
+                                            </button>
+                                        } else {
+                                            <button class="btn btn-sm btn-secondary"
+                                                onclick={on_unwatch_click}>
+                                                { "Unwatch" }
+                                            </button>
+                                        }
                                         <button class="btn btn-sm btn-secondary"
                                             onclick={on_lyrics_click}>
                                             { if is_active { "Hide" } else { "Lyrics" } }

frontend/src/pages/artist.rs

@@ -106,9 +106,40 @@ pub fn artist_page(props: &Props) -> Html {
     };
 
     let watch_all_btn = {
-        let artist_status = d.artist_status.clone();
+        let is_watched = d.artist_status == "owned"
-        let show = artist_status != "owned";
+            || d.artist_status == "partial"
-        if show {
+            || d.artist_status == "wanted";
+        if is_watched {
+            // Unwatch All
+            let artist_id_num = d.artist.id;
+            let artist_name = d.artist.name.clone();
+            let message = message.clone();
+            let error = error.clone();
+            let fetch = fetch.clone();
+            let artist_id = id.clone();
+            html! {
+                <button class="btn btn-sm btn-secondary"
+                    onclick={Callback::from(move |_: MouseEvent| {
+                        let artist_name = artist_name.clone();
+                        let message = message.clone();
+                        let error = error.clone();
+                        let fetch = fetch.clone();
+                        let artist_id = artist_id.clone();
+                        wasm_bindgen_futures::spawn_local(async move {
+                            match api::unwatch_artist(artist_id_num).await {
+                                Ok(_) => {
+                                    message.set(Some(format!("Unwatched {artist_name}")));
+                                    fetch.emit(artist_id);
+                                }
+                                Err(e) => error.set(Some(e.0)),
+                            }
+                        });
+                    })}>
+                    { "Unwatch All" }
+                </button>
+            }
+        } else {
+            // Watch All
             let artist_name = d.artist.name.clone();
             let artist_mbid = d.artist.musicbrainz_id.clone();
             let message = message.clone();
@@ -140,8 +171,6 @@ pub fn artist_page(props: &Props) -> Html {
                     { "Watch All" }
                 </button>
             }
-        } else {
-            html! {}
         }
     };
 
@@ -191,6 +220,33 @@ pub fn artist_page(props: &Props) -> Html {
         }
     };
 
+    let remove_btn = {
+        let artist_id_num = d.artist.id;
+        if artist_id_num > 0 {
+            let error = error.clone();
+            html! {
+                <button class="btn btn-sm btn-danger"
+                    onclick={Callback::from(move |_: MouseEvent| {
+                        let error = error.clone();
+                        wasm_bindgen_futures::spawn_local(async move {
+                            if let Err(e) = api::delete_artist(artist_id_num).await {
+                                error.set(Some(e.0));
+                                return;
+                            }
+                            // Navigate back to library
+                            if let Some(window) = web_sys::window() {
+                                let _ = window.location().set_href("/library");
+                            }
+                        });
+                    })}>
+                    { "Remove" }
+                </button>
+            }
+        } else {
+            html! {}
+        }
+    };
+
     html! {
         <div>
             if let Some(ref banner) = d.artist_banner {
@@ -249,6 +305,7 @@ pub fn artist_page(props: &Props) -> Html {
                     <div class="flex gap-1">
                         { watch_all_btn }
                         { monitor_btn }
+                        { remove_btn }
                     </div>
                 </div>
                 if let Some(ref bio) = d.artist_bio {
@@ -279,7 +336,7 @@ pub fn artist_page(props: &Props) -> Html {
                 <p class="text-muted">{ "No releases found on MusicBrainz." }</p>
             }
 
-            // Group albums by type
+            // Group albums by type (primary credit only)
             { for ["Album", "EP", "Single"].iter().map(|release_type| {
                 let type_albums: Vec<_> = d.albums.iter()
                     .filter(|a| a.release_type.as_deref().unwrap_or("Album") == *release_type)
@@ -315,7 +372,7 @@ pub fn artist_page(props: &Props) -> Html {
 
                             let tc = album.track_count;
 
-                            // Watch button for unwatched albums
+                            // Watch/Unwatch toggle for albums
                             let watch_btn = if is_unwatched {
                                 let artist_name = d.artist.name.clone();
                                 let album_title = album.title.clone();
@@ -351,7 +408,34 @@ pub fn artist_page(props: &Props) -> Html {
                                     </button>
                                 }
                             } else {
-                                html! {}
+                                let album_title = album.title.clone();
+                                let album_mbid = album.mbid.clone();
+                                let message = message.clone();
+                                let error = error.clone();
+                                let fetch = fetch.clone();
+                                let artist_id = id.clone();
+                                html! {
+                                    <button class="btn btn-sm btn-secondary"
+                                        onclick={Callback::from(move |_: MouseEvent| {
+                                            let album_title = album_title.clone();
+                                            let album_mbid = album_mbid.clone();
+                                            let message = message.clone();
+                                            let error = error.clone();
+                                            let fetch = fetch.clone();
+                                            let artist_id = artist_id.clone();
+                                            wasm_bindgen_futures::spawn_local(async move {
+                                                match api::unwatch_album(&album_mbid).await {
+                                                    Ok(_) => {
+                                                        message.set(Some(format!("Unwatched '{album_title}'")));
+                                                        fetch.emit(artist_id);
+                                                    }
+                                                    Err(e) => error.set(Some(e.0)),
+                                                }
+                                            });
+                                        })}>
+                                        { "Unwatch" }
+                                    </button>
+                                }
                             };
 
                             html! {
@@ -397,6 +481,56 @@ pub fn artist_page(props: &Props) -> Html {
                     </div>
                 }
             })}
+
+            // Featured releases (collapsible, pre-collapsed)
+            { for ["Album", "EP", "Single"].iter().map(|release_type| {
+                let featured: Vec<_> = d.featured_albums.iter()
+                    .filter(|a| a.release_type.as_deref().unwrap_or("Album") == *release_type)
+                    .collect();
+                if featured.is_empty() {
+                    return html! {};
+                }
+                html! {
+                    <details class="mb-2">
+                        <summary class="text-muted" style="cursor: pointer;">
+                            { format!("Featured {}s ({})", release_type, featured.len()) }
+                        </summary>
+                        <table>
+                            <thead>
+                                <tr>
+                                    <th style="width: 60px;"></th>
+                                    <th>{ "Title" }</th>
+                                    <th>{ "Date" }</th>
+                                </tr>
+                            </thead>
+                            <tbody>
+                                { for featured.iter().map(|album| {
+                                    let cover_url = format!("https://coverartarchive.org/release/{}/front-250", album.mbid);
+                                    html! {
+                                        <tr style="opacity: 0.6;">
+                                            <td>
+                                                <img class="album-art" src={cover_url}
+                                                    loading="lazy"
+                                                    onerror={Callback::from(|e: web_sys::Event| {
+                                                        if let Some(el) = e.target_dyn_into::<web_sys::HtmlElement>() {
+                                                            el.set_attribute("style", "display:none").ok();
+                                                        }
+                                                    })} />
+                                            </td>
+                                            <td>
+                                                <Link<Route> to={Route::Album { mbid: album.mbid.clone() }}>
+                                                    { &album.title }
+                                                </Link<Route>>
+                                            </td>
+                                            <td class="text-muted">{ album.date.as_deref().unwrap_or("") }</td>
+                                        </tr>
+                                    }
+                                })}
+                            </tbody>
+                        </table>
+                    </details>
+                }
+            })}
         </div>
     }
 }

frontend/src/pages/dashboard.rs

@@ -34,6 +34,8 @@ pub fn dashboard() -> Html {
     let status = use_state(|| None::<Status>);
     let error = use_state(|| None::<String>);
     let message = use_state(|| None::<String>);
+    let pipeline_was_active = use_state(|| false);
+    let pipeline_complete = use_state(|| false);
 
     // Fetch status function
     let fetch_status = {
@@ -240,72 +242,182 @@ pub fn dashboard() -> Html {
         .iter()
         .any(|t| t.status == "Pending" || t.status == "Running");
 
-    // Pre-compute scheduled task rows
+    // Skip callbacks for scheduler
-    let scheduled_rows = {
+    let on_skip_pipeline = {
-        let mut rows = Vec::new();
+        let message = message.clone();
-        if let Some(ref sched) = s.scheduled {
+        let error = error.clone();
-            if let Some(ref next) = sched.next_pipeline {
+        let fetch = fetch_status.clone();
-                let on_skip = {
+        Callback::from(move |_: MouseEvent| {
-                    let message = message.clone();
+            let message = message.clone();
-                    let error = error.clone();
+            let error = error.clone();
-                    let fetch = fetch_status.clone();
+            let fetch = fetch.clone();
-                    Callback::from(move |_: MouseEvent| {
+            wasm_bindgen_futures::spawn_local(async move {
-                        let message = message.clone();
+                match api::skip_scheduled_pipeline().await {
-                        let error = error.clone();
+                    Ok(_) => {
-                        let fetch = fetch.clone();
+                        message.set(Some("Next pipeline run skipped".into()));
-                        wasm_bindgen_futures::spawn_local(async move {
+                        fetch.emit(());
-                            match api::skip_scheduled_pipeline().await {
+                    }
-                                Ok(_) => {
+                    Err(e) => error.set(Some(e.0)),
-                                    message.set(Some("Next pipeline run skipped".into()));
+                }
-                                    fetch.emit(());
+            });
-                                }
+        })
-                                Err(e) => error.set(Some(e.0)),
+    };
-                            }
+    let on_skip_monitor = {
-                        });
+        let message = message.clone();
-                    })
+        let error = error.clone();
-                };
+        let fetch = fetch_status.clone();
-                rows.push(html! {
+        Callback::from(move |_: MouseEvent| {
-                    <tr>
+            let message = message.clone();
-                        <td>{ "Auto Pipeline" }</td>
+            let error = error.clone();
-                        <td><span class="badge badge-pending">{ "Scheduled" }</span></td>
+            let fetch = fetch.clone();
-                        <td class="text-sm text-muted">{ format!("Next run: {}", format_next_run(next)) }</td>
+            wasm_bindgen_futures::spawn_local(async move {
-                        <td><button class="btn btn-sm btn-danger" onclick={on_skip}>{ "Skip" }</button></td>
+                match api::skip_scheduled_monitor().await {
-                    </tr>
+                    Ok(_) => {
-                });
+                        message.set(Some("Next monitor check skipped".into()));
-            }
+                        fetch.emit(());
-            if let Some(ref next) = sched.next_monitor {
+                    }
-                let on_skip = {
+                    Err(e) => error.set(Some(e.0)),
-                    let message = message.clone();
+                }
-                    let error = error.clone();
+            });
-                    let fetch = fetch_status.clone();
+        })
-                    Callback::from(move |_: MouseEvent| {
+    };
-                        let message = message.clone();
+
-                        let error = error.clone();
+    let pipeline_progress_html = if let Some(ref wq) = s.work_queue {
-                        let fetch = fetch.clone();
+        let active = wq.download.pending
-                        wasm_bindgen_futures::spawn_local(async move {
+            + wq.download.running
-                            match api::skip_scheduled_monitor().await {
+            + wq.index.pending
-                                Ok(_) => {
+            + wq.index.running
-                                    message.set(Some("Next monitor check skipped".into()));
+            + wq.tag.pending
-                                    fetch.emit(());
+            + wq.tag.running
-                                }
+            + wq.organize.pending
-                                Err(e) => error.set(Some(e.0)),
+            + wq.organize.running
-                            }
+            + wq.enrich.pending
-                        });
+            + wq.enrich.running;
-                    })
+
-                };
+        // Track pipeline active→inactive transition
-                rows.push(html! {
+        if active > 0 {
-                    <tr>
+            if !*pipeline_was_active {
-                        <td>{ "Monitor Check" }</td>
+                pipeline_was_active.set(true);
-                        <td><span class="badge badge-pending">{ "Scheduled" }</span></td>
+                pipeline_complete.set(false);
-                        <td class="text-sm text-muted">{ format!("Next run: {}", format_next_run(next)) }</td>
+            }
-                        <td><button class="btn btn-sm btn-danger" onclick={on_skip}>{ "Skip" }</button></td>
+        } else if *pipeline_was_active {
-                    </tr>
+            pipeline_was_active.set(false);
-                });
+            pipeline_complete.set(true);
-            }
+        }
-        }
+
-        rows
+        if active > 0 {
+            html! {
+                <div class="card">
+                    <h3>{ "Pipeline Progress" }</h3>
+                    <table>
+                        <thead>
+                            <tr><th>{ "Step" }</th><th>{ "Pending" }</th><th>{ "Running" }</th><th>{ "Done" }</th><th>{ "Failed" }</th></tr>
+                        </thead>
+                        <tbody>
+                            { for [("Download", &wq.download), ("Index", &wq.index), ("Tag", &wq.tag), ("Organize", &wq.organize), ("Enrich", &wq.enrich)].iter().map(|(name, c)| {
+                                html! {
+                                    <tr>
+                                        <td>{ name }</td>
+                                        <td>{ c.pending }</td>
+                                        <td>{ if c.running > 0 { html! { <span class="badge badge-accent">{ c.running }</span> } } else { html! { { "0" } } } }</td>
+                                        <td>{ c.completed }</td>
+                                        <td>{ if c.failed > 0 { html! { <span class="badge badge-danger">{ c.failed }</span> } } else { html! { { "0" } } } }</td>
+                                    </tr>
+                                }
+                            })}
+                        </tbody>
+                    </table>
+                </div>
+            }
+        } else if *pipeline_complete {
+            html! {
+                <div class="card" style="border-color: var(--success);">
+                    <p style="color: var(--success);">{ "Pipeline run complete!" }</p>
+                </div>
+            }
+        } else {
+            html! {}
+        }
+    } else if *pipeline_complete {
+        html! {
+            <div class="card" style="border-color: var(--success);">
+                <p style="color: var(--success);">{ "Pipeline run complete!" }</p>
+            </div>
+        }
+    } else {
+        html! {}
+    };
+
+    let scheduled_jobs_html = {
+        let next_pipeline = s
+            .scheduled
+            .as_ref()
+            .and_then(|sc| sc.next_pipeline.as_ref());
+        let next_monitor = s.scheduled.as_ref().and_then(|sc| sc.next_monitor.as_ref());
+        let pipeline_next_str = next_pipeline
+            .map(|n| format_next_run(n))
+            .unwrap_or_default();
+        let monitor_next_str = next_monitor.map(|n| format_next_run(n)).unwrap_or_default();
+        let pipeline_last = s
+            .scheduler
+            .as_ref()
+            .and_then(|sc| sc.get("pipeline"))
+            .and_then(|j| j.get("last_result"))
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+        let monitor_last = s
+            .scheduler
+            .as_ref()
+            .and_then(|sc| sc.get("monitor"))
+            .and_then(|j| j.get("last_result"))
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+
+        html! {
+            <div class="card">
+                <h3>{ "Scheduled Jobs" }</h3>
+                <table>
+                    <thead>
+                        <tr><th>{ "Job" }</th><th>{ "Status" }</th><th>{ "Next Run" }</th><th>{ "Last Result" }</th><th></th></tr>
+                    </thead>
+                    <tbody>
+                        <tr>
+                            <td>{ "Auto Pipeline" }</td>
+                            <td>{ if next_pipeline.is_some() {
+                                html! { <span class="badge badge-pending">{ "Scheduled" }</span> }
+                            } else {
+                                html! { <span class="text-muted text-sm">{ "Idle" }</span> }
+                            }}</td>
+                            <td class="text-sm text-muted">{ pipeline_next_str }</td>
+                            <td class="text-sm text-muted">{ pipeline_last }</td>
+                            <td>{ if next_pipeline.is_some() {
+                                html! { <button class="btn btn-sm btn-danger" onclick={on_skip_pipeline}>{ "Skip" }</button> }
+                            } else {
+                                html! {}
+                            }}</td>
+                        </tr>
+                        <tr>
+                            <td>{ "Monitor Check" }</td>
+                            <td>{ if next_monitor.is_some() {
+                                html! { <span class="badge badge-pending">{ "Scheduled" }</span> }
+                            } else {
+                                html! { <span class="text-muted text-sm">{ "Idle" }</span> }
+                            }}</td>
+                            <td class="text-sm text-muted">{ monitor_next_str }</td>
+                            <td class="text-sm text-muted">{ monitor_last }</td>
+                            <td>{ if next_monitor.is_some() {
+                                html! { <button class="btn btn-sm btn-danger" onclick={on_skip_monitor}>{ "Skip" }</button> }
+                            } else {
+                                html! {}
+                            }}</td>
+                        </tr>
+                    </tbody>
+                </table>
+            </div>
+        }
     };
-    let has_scheduled = !scheduled_rows.is_empty();
 
     html! {
         <div>
@@ -365,8 +477,14 @@ pub fn dashboard() -> Html {
                 </div>
             </div>
 
-            // Background Tasks (always show if there are tasks or scheduled items)
+            // Work Queue Progress
-            if !s.tasks.is_empty() || has_scheduled {
+            { pipeline_progress_html }
+
+            // Scheduled Jobs (always visible)
+            { scheduled_jobs_html }
+
+            // Background Tasks (one-off tasks like MB import)
+            if !s.tasks.is_empty() {
                 <div class="card">
                     <h3>{ "Background Tasks" }</h3>
                     <table class="tasks-table">
@@ -374,7 +492,6 @@ pub fn dashboard() -> Html {
                             <tr><th>{ "Type" }</th><th>{ "Status" }</th><th>{ "Progress" }</th><th>{ "Result" }</th></tr>
                         </thead>
                         <tbody>
-                            { for scheduled_rows.into_iter() }
                             { for s.tasks.iter().map(|t| {
                                 let progress_html = if let Some(ref p) = t.progress {
                                     if p.total > 0 {
@@ -417,7 +534,7 @@ pub fn dashboard() -> Html {
                             <tr><th>{ "Query" }</th><th>{ "Status" }</th><th>{ "Error" }</th></tr>
                         </thead>
                         <tbody>
-                            { for s.queue.items.iter().map(|item| html! {
+                            { for s.queue.items.iter().take(10).map(|item| html! {
                                 <tr>
                                     <td>{ &item.query }</td>
                                     <td><StatusBadge status={item.status.clone()} /></td>
@@ -426,6 +543,11 @@ pub fn dashboard() -> Html {
                             })}
                         </tbody>
                     </table>
+                    if s.queue.items.len() > 10 {
+                        <p class="text-sm text-muted mt-1">
+                            { format!("and {} more...", s.queue.items.len() - 10) }
+                        </p>
+                    }
                 </div>
             } else if s.queue.pending > 0 || s.queue.downloading > 0 {
                 <div class="card">
@@ -445,7 +567,7 @@ pub fn dashboard() -> Html {
                                     <tr><th>{ "Title" }</th><th>{ "Artist" }</th><th>{ "Album" }</th><th>{ "MBID" }</th></tr>
                                 </thead>
                                 <tbody>
-                                    { for tagging.items.iter().map(|t| html! {
+                                    { for tagging.items.iter().take(10).map(|t| html! {
                                         <tr>
                                             <td>{ t.title.as_deref().unwrap_or("Unknown") }</td>
                                             <td>{ t.artist.as_deref().unwrap_or("") }</td>
@@ -461,6 +583,11 @@ pub fn dashboard() -> Html {
                                     })}
                                 </tbody>
                             </table>
+                            if tagging.items.len() > 10 {
+                                <p class="text-sm text-muted mt-1">
+                                    { format!("and {} more...", tagging.items.len() - 10) }
+                                </p>
+                            }
                         }
                     </div>
                 }

frontend/src/pages/library.rs

@@ -10,16 +10,25 @@ pub fn library_page() -> Html {
     let artists = use_state(|| None::<Vec<ArtistListItem>>);
     let error = use_state(|| None::<String>);
 
-    {
+    let fetch_artists = {
         let artists = artists.clone();
         let error = error.clone();
-        use_effect_with((), move |_| {
+        Callback::from(move |_: ()| {
+            let artists = artists.clone();
+            let error = error.clone();
             wasm_bindgen_futures::spawn_local(async move {
                 match api::list_artists(200, 0).await {
                     Ok(a) => artists.set(Some(a)),
                     Err(e) => error.set(Some(e.0)),
                 }
             });
+        })
+    };
+
+    {
+        let fetch = fetch_artists.clone();
+        use_effect_with((), move |_| {
+            fetch.emit(());
         });
     }
 
@@ -49,10 +58,25 @@ pub fn library_page() -> Html {
                             <th>{ "Owned" }</th>
                             <th>{ "Watched" }</th>
                             <th>{ "Tracks" }</th>
+                            <th></th>
                         </tr>
                     </thead>
                     <tbody>
-                        { for artists.iter().map(|a| html! {
+                        { for artists.iter().map(|a| {
+                            let artist_id = a.id;
+                            let error = error.clone();
+                            let fetch = fetch_artists.clone();
+                            let on_remove = Callback::from(move |_: MouseEvent| {
+                                let error = error.clone();
+                                let fetch = fetch.clone();
+                                wasm_bindgen_futures::spawn_local(async move {
+                                    match api::delete_artist(artist_id).await {
+                                        Ok(_) => fetch.emit(()),
+                                        Err(e) => error.set(Some(e.0)),
+                                    }
+                                });
+                            });
+                            html! {
                             <tr>
                                 <td>
                                     <Link<Route> to={Route::Artist { id: a.id.to_string() }}>
@@ -64,8 +88,8 @@ pub fn library_page() -> Html {
                                         <span style="color: var(--success);" title="Monitored">{ "\u{2713}" }</span>
                                     }
                                 </td>
-                                <td>
+                                if a.enriched {
-                                    if a.total_items > 0 {
+                                    <td>
                                         <span class="text-sm" style={
                                             if a.total_owned >= a.total_watched && a.total_watched > 0 { "color: var(--success);" }
                                             else if a.total_owned > 0 { "color: var(--warning);" }
@@ -73,24 +97,32 @@ pub fn library_page() -> Html {
                                         }>
                                             { format!("{}/{}", a.total_owned, a.total_watched) }
                                         </span>
-                                    }
+                                    </td>
-                                </td>
+                                    <td>
-                                <td>
-                                    if a.total_items > 0 {
                                         <span class="text-sm" style={
                                             if a.total_watched > 0 { "color: var(--accent);" }
                                             else { "color: var(--text-muted);" }
                                         }>
                                             { format!("{}/{}", a.total_watched, a.total_items) }
                                         </span>
-                                    }
+                                    </td>
-                                </td>
+                                } else {
+                                    <td colspan="2" class="text-sm text-muted loading">
+                                        { "Awaiting artist enrichment..." }
+                                    </td>
+                                }
                                 <td class="text-muted text-sm">
-                                    if a.total_items > 0 {
+                                    if a.enriched && a.total_items > 0 {
                                         { a.total_items }
                                     }
                                 </td>
+                                <td>
+                                    <button class="btn btn-sm btn-danger" onclick={on_remove}>
+                                        { "Remove" }
+                                    </button>
+                                </td>
                             </tr>
+                            }
                         })}
                     </tbody>
                 </table>

frontend/src/pages/settings.rs

@@ -453,6 +453,18 @@ pub fn settings_page() -> Html {
                                 }
                             })} />
                     </div>
+                    <div class="form-group">
+                        <label>{ "Concurrency" }</label>
+                        <input type="number" min="1" max="16" value={c.tagging.concurrency.to_string()}
+                            oninput={let config = config.clone(); Callback::from(move |e: InputEvent| {
+                                let input: HtmlInputElement = e.target_unchecked_into();
+                                if let Ok(v) = input.value().parse::<usize>() {
+                                    let mut cfg = (*config).clone().unwrap();
+                                    cfg.tagging.concurrency = v;
+                                    config.set(Some(cfg));
+                                }
+                            })} />
+                    </div>
                 </div>
 
                 // Downloads
@@ -615,9 +627,11 @@ pub fn settings_page() -> Html {
                         <p class="text-sm" style="margin:0;">
                             <strong style="color: var(--warning);">{ "Heads up: " }</strong>
                             { "This downloads ~24 GB of data and builds a ~10 GB local database. " }
-                            { "The initial import can take 3-6 hours depending on your hardware. " }
+                            { "The initial import can take 12-24 hours depending on your hardware. " }
                             { "Total disk usage: ~35 GB (downloads + database). " }
-                            { "After the initial import, the database is automatically refreshed weekly to stay current." }
+                            { "After the initial import, the database is automatically refreshed weekly to stay current. " }
+                            <strong>{ "SSD storage is recommended" }</strong>
+                            { " for best performance. HDDs will still be faster than the API, but lookups may take a few seconds instead of being instant." }
                         </p>
                     </div>
                     {
@@ -837,6 +851,27 @@ pub fn settings_page() -> Html {
                     </div>
                 </div>
 
+                // Logging
+                <div class="card">
+                    <h3>{ "Logging" }</h3>
+                    <div class="form-group">
+                        <label>{ "Log Level" }</label>
+                        <select onchange={let config = config.clone(); Callback::from(move |e: Event| {
+                            let select: HtmlSelectElement = e.target_unchecked_into();
+                            let mut cfg = (*config).clone().unwrap();
+                            cfg.log_level = select.value();
+                            config.set(Some(cfg));
+                        })}>
+                            { for [("error", "Error"), ("warn", "Warning"), ("info", "Info"), ("debug", "Debug"), ("trace", "Trace")].iter().map(|(v, label)| html! {
+                                <option value={*v} selected={c.log_level == *v}>{ label }</option>
+                            })}
+                        </select>
+                        <p class="text-muted text-sm" style="margin-top: 0.25rem;">
+                            { "Requires restart to take effect. CLI flags (-v, -vv) override this setting." }
+                        </p>
+                    </div>
+                </div>
+
                 <button type="submit" class="btn btn-primary">{ "Save Settings" }</button>
             </form>
         </div>

frontend/src/types.rs

@@ -30,6 +30,8 @@ pub struct ArtistListItem {
     pub musicbrainz_id: Option<String>,
     #[serde(default)]
     pub monitored: bool,
+    #[serde(default)]
+    pub enriched: bool,
     pub total_watched: usize,
     pub total_owned: usize,
     pub total_items: usize,
@@ -54,6 +56,8 @@ pub struct FullAlbumInfo {
 pub struct FullArtistDetail {
     pub artist: Artist,
     pub albums: Vec<FullAlbumInfo>,
+    #[serde(default)]
+    pub featured_albums: Vec<FullAlbumInfo>,
     pub artist_status: String,
     #[serde(default)]
     pub total_available_tracks: u32,
@@ -120,6 +124,8 @@ pub struct Track {
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct MbAlbumDetail {
     pub mbid: String,
+    #[serde(default)]
+    pub artist: Option<String>,
     pub tracks: Vec<MbAlbumTrack>,
 }
 
@@ -225,7 +231,10 @@ pub struct TaskRef {
 
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct PipelineRef {
+    #[serde(default)]
     pub task_ids: Vec<String>,
+    #[serde(default)]
+    pub pipeline_id: Option<String>,
 }
 
 // --- Status ---
@@ -246,6 +255,10 @@ pub struct Status {
     pub tasks: Vec<TaskInfo>,
     #[serde(default)]
     pub scheduled: Option<ScheduledTasks>,
+    #[serde(default)]
+    pub work_queue: Option<WorkQueueStats>,
+    #[serde(default)]
+    pub scheduler: Option<serde_json::Value>,
 }
 
 #[derive(Debug, Clone, PartialEq, Deserialize)]
@@ -254,6 +267,24 @@ pub struct ScheduledTasks {
     pub next_monitor: Option<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
+pub struct WorkQueueCounts {
+    pub pending: u64,
+    pub running: u64,
+    pub completed: u64,
+    pub failed: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+pub struct WorkQueueStats {
+    pub download: WorkQueueCounts,
+    pub index: WorkQueueCounts,
+    pub tag: WorkQueueCounts,
+    pub organize: WorkQueueCounts,
+    #[serde(default)]
+    pub enrich: WorkQueueCounts,
+}
+
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct LibrarySummary {
     pub total_items: u64,
@@ -282,6 +313,14 @@ pub struct AddSummary {
     pub errors: u64,
 }
 
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+pub struct WatchTrackResponse {
+    pub id: i32,
+    pub status: String,
+    pub name: String,
+    pub artist_name: Option<String>,
+}
+
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct SyncStats {
     pub found: u64,
@@ -402,6 +441,12 @@ pub struct AppConfig {
     pub scheduling: SchedulingConfigFe,
     #[serde(default)]
     pub musicbrainz: MusicBrainzConfigFe,
+    #[serde(default = "default_log_level")]
+    pub log_level: String,
+}
+
+fn default_log_level() -> String {
+    "info".into()
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
@@ -420,6 +465,12 @@ pub struct TaggingConfigFe {
     pub write_tags: bool,
     #[serde(default)]
     pub confidence: f64,
+    #[serde(default = "default_tag_concurrency")]
+    pub concurrency: usize,
+}
+
+fn default_tag_concurrency() -> usize {
+    4
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]

frontend/style.css

@@ -344,3 +344,59 @@ tr[draggable="true"]:active { cursor: grabbing; }
     flex-direction: column;
     gap: 0.25rem;
 }
+
+/* Hamburger menu button — hidden on desktop */
+.hamburger {
+    display: none;
+    position: fixed;
+    top: 0.75rem;
+    left: 0.75rem;
+    z-index: 101;
+    background: var(--bg-secondary);
+    border: 1px solid var(--border);
+    border-radius: var(--radius);
+    color: var(--text-primary);
+    font-size: 1.4rem;
+    padding: 0.25rem 0.6rem;
+    cursor: pointer;
+    line-height: 1;
+}
+
+/* Sidebar overlay — hidden on desktop */
+.sidebar-overlay {
+    display: none;
+    position: fixed;
+    inset: 0;
+    background: rgba(0, 0, 0, 0.5);
+    z-index: 99;
+}
+
+/* Mobile responsive */
+@media (max-width: 768px) {
+    .hamburger { display: block; }
+    .sidebar-overlay.open { display: block; }
+    .sidebar {
+        transform: translateX(-100%);
+        z-index: 100;
+        transition: transform 0.2s ease;
+    }
+    .sidebar.open { transform: translateX(0); }
+    .main-content {
+        margin-left: 0;
+        padding: 1rem;
+        padding-top: 3.5rem;
+    }
+    .stats-grid { grid-template-columns: repeat(2, 1fr); }
+    table { display: block; overflow-x: auto; }
+    .form-group input[type="text"],
+    .form-group input[type="number"],
+    .form-group input[type="password"],
+    .form-group select {
+        width: 100% !important;
+        min-width: 0 !important;
+    }
+    .tab-bar { overflow-x: auto; }
+    .album-art-lg { width: 120px; height: 120px; }
+    .album-header { flex-direction: column; }
+    .artist-photo { width: 80px; height: 80px; }
+}

src/cookie_refresh.rs

@@ -1,59 +1,19 @@
-//! Background task that periodically refreshes YouTube cookies via headless Firefox.
+//! YouTube cookie refresh via headless Firefox.
 
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 use std::process::Stdio;
-use std::sync::Arc;
-use std::time::Duration;
 
 use tokio::process::Command;
-use tokio::sync::RwLock;
 
-use crate::config::AppConfig;
+/// Run a headless cookie refresh. Returns success message or error.
+pub async fn run_refresh() -> Result<String, String> {
+    let profile_dir = shanty_config::data_dir().join("firefox-profile");
+    let cookies_path = shanty_config::data_dir().join("cookies.txt");
 
-/// Spawn the cookie refresh background loop.
+    if !profile_dir.exists() {
-///
+        return Err(format!("no Firefox profile at {}", profile_dir.display()));
-/// This task runs forever, sleeping for `cookie_refresh_hours` between refreshes.
+    }
-/// It reads the current config on each iteration so changes take effect without restart.
-pub fn spawn(config: Arc<RwLock<AppConfig>>) {
-    tokio::spawn(async move {
-        loop {
-            let (enabled, hours) = {
-                let cfg = config.read().await;
-                (
-                    cfg.download.cookie_refresh_enabled,
-                    cfg.download.cookie_refresh_hours.max(1),
-                )
-            };
 
-            // Sleep for the configured interval
-            tokio::time::sleep(Duration::from_secs(u64::from(hours) * 3600)).await;
-
-            if !enabled {
-                continue;
-            }
-
-            let profile_dir = shanty_config::data_dir().join("firefox-profile");
-            let cookies_path = shanty_config::data_dir().join("cookies.txt");
-
-            if !profile_dir.exists() {
-                tracing::warn!(
-                    "cookie refresh skipped: no Firefox profile at {}",
-                    profile_dir.display()
-                );
-                continue;
-            }
-
-            tracing::info!("starting cookie refresh");
-
-            match run_refresh(&profile_dir, &cookies_path).await {
-                Ok(msg) => tracing::info!("cookie refresh complete: {msg}"),
-                Err(e) => tracing::error!("cookie refresh failed: {e}"),
-            }
-        }
-    });
-}
-
-async fn run_refresh(profile_dir: &Path, cookies_path: &Path) -> Result<String, String> {
     let script = find_script()?;
 
     let output = Command::new("python3")

src/lib.rs

@@ -11,7 +11,8 @@ pub mod error;
 pub mod mb_update;
 pub mod monitor;
 pub mod pipeline;
-pub mod pipeline_scheduler;
 pub mod routes;
+pub mod scheduler;
 pub mod state;
 pub mod tasks;
+pub mod workers;

src/main.rs

@@ -14,6 +14,7 @@ use shanty_web::config::AppConfig;
 use shanty_web::routes;
 use shanty_web::state::AppState;
 use shanty_web::tasks::TaskManager;
+use shanty_web::workers::WorkerManager;
 
 #[derive(Parser)]
 #[command(name = "shanty-web", about = "Shanty web interface backend")]
@@ -35,18 +36,27 @@ struct Cli {
 async fn main() -> anyhow::Result<()> {
     let cli = Cli::parse();
 
+    let mut config = AppConfig::load(cli.config.as_deref());
+
     let filter = match cli.verbose {
-        0 => "info,shanty_web=info",
+        0 => {
-        1 => "info,shanty_web=debug",
+            let level = config.log_level.to_lowercase();
-        _ => "debug,shanty_web=trace",
+            match level.as_str() {
+                "error" => "error".to_string(),
+                "warn" => "warn".to_string(),
+                "debug" => "debug,shanty_web=debug".to_string(),
+                "trace" => "trace,shanty_web=trace".to_string(),
+                _ => "info,shanty_web=info".to_string(),
+            }
+        }
+        1 => "info,shanty_web=debug".to_string(),
+        _ => "debug,shanty_web=trace".to_string(),
     };
     tracing_subscriber::fmt()
         .with_env_filter(
-            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(filter)),
+            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(&filter)),
         )
         .init();
-
-    let mut config = AppConfig::load(cli.config.as_deref());
     if let Some(port) = cli.port {
         config.web.port = port;
     }
@@ -85,21 +95,13 @@ async fn main() -> anyhow::Result<()> {
         config: std::sync::Arc::new(tokio::sync::RwLock::new(config)),
         config_path,
         tasks: TaskManager::new(),
+        workers: WorkerManager::new(),
         firefox_login: tokio::sync::Mutex::new(None),
-        scheduler: tokio::sync::Mutex::new(shanty_web::state::SchedulerInfo {
-            next_pipeline: None,
-            next_monitor: None,
-            skip_pipeline: false,
-            skip_monitor: false,
-        }),
     });
 
-    // Start background cookie refresh task
+    // Start work queue workers and unified scheduler
-    shanty_web::cookie_refresh::spawn(state.config.clone());
+    WorkerManager::spawn_all(state.clone());
-
+    shanty_web::scheduler::spawn(state.clone());
-    // Start pipeline and monitor schedulers
-    shanty_web::pipeline_scheduler::spawn(state.clone());
-    shanty_web::monitor::spawn(state.clone());
     shanty_web::mb_update::spawn(state.clone());
 
     // Resolve static files directory relative to the binary location

src/monitor.rs

@@ -2,7 +2,6 @@
 //! and automatically adds them to the watchlist.
 
 use std::collections::HashSet;
-use std::time::Duration;
 
 use actix_web::web;
 use serde::Serialize;
@@ -197,68 +196,3 @@ pub async fn check_monitored_artists(
 
     Ok(stats)
 }
-
-/// Spawn the monitor scheduler background loop.
-///
-/// Sleeps for the configured interval, then checks monitored artists if enabled.
-/// Reads config each iteration so changes take effect without restart.
-pub fn spawn(state: web::Data<AppState>) {
-    tokio::spawn(async move {
-        loop {
-            let (enabled, hours) = {
-                let cfg = state.config.read().await;
-                (
-                    cfg.scheduling.monitor_enabled,
-                    cfg.scheduling.monitor_interval_hours.max(1),
-                )
-            };
-
-            let sleep_secs = u64::from(hours) * 3600;
-
-            // Update next-run time
-            {
-                let mut sched = state.scheduler.lock().await;
-                sched.next_monitor = if enabled {
-                    Some(
-                        (chrono::Utc::now() + chrono::Duration::seconds(sleep_secs as i64))
-                            .naive_utc(),
-                    )
-                } else {
-                    None
-                };
-            }
-
-            tokio::time::sleep(Duration::from_secs(sleep_secs)).await;
-
-            if !enabled {
-                continue;
-            }
-
-            // Check if this run was skipped
-            {
-                let mut sched = state.scheduler.lock().await;
-                sched.next_monitor = None;
-                if sched.skip_monitor {
-                    sched.skip_monitor = false;
-                    tracing::info!("scheduled monitor check skipped (user cancelled)");
-                    continue;
-                }
-            }
-
-            tracing::info!("scheduled monitor check starting");
-            match check_monitored_artists(&state).await {
-                Ok(stats) => {
-                    tracing::info!(
-                        artists_checked = stats.artists_checked,
-                        new_releases = stats.new_releases_found,
-                        tracks_added = stats.tracks_added,
-                        "scheduled monitor check complete"
-                    );
-                }
-                Err(e) => {
-                    tracing::error!(error = %e, "scheduled monitor check failed");
-                }
-            }
-        }
-    });
-}

src/pipeline.rs

@@ -1,218 +1,81 @@
-//! Shared pipeline logic used by both the API endpoint and the scheduler.
+//! Pipeline trigger logic. Creates work queue items that flow through the
+//! Download → Tag → Organize pipeline concurrently via typed workers.
 
 use actix_web::web;
+
+use shanty_db::entities::download_queue::DownloadStatus;
+use shanty_db::entities::work_queue::WorkTaskType;
 use shanty_db::queries;
 
-use crate::routes::artists::enrich_all_watched_artists;
+use crate::error::ApiError;
 use crate::state::AppState;
 
-/// Register and spawn the full 6-step pipeline. Returns the task IDs immediately.
+/// Trigger the full pipeline: sync wanted items to download queue, then create
-///
+/// work queue items for each pending download. Returns a pipeline ID for tracking.
-/// Steps: sync, download, index, tag, organize, enrich.
+pub async fn trigger_pipeline(state: &web::Data<AppState>) -> Result<String, ApiError> {
-pub fn spawn_pipeline(state: &web::Data<AppState>) -> Vec<String> {
+    let pipeline_id = uuid::Uuid::new_v4().to_string();
-    let sync_id = state.tasks.register_pending("sync");
+    let conn = state.db.conn();
-    let download_id = state.tasks.register_pending("download");
-    let index_id = state.tasks.register_pending("index");
-    let tag_id = state.tasks.register_pending("tag");
-    let organize_id = state.tasks.register_pending("organize");
-    let enrich_id = state.tasks.register_pending("enrich");
 
-    let task_ids = vec![
+    // Clear completed/failed items from previous pipeline runs (not standalone tasks)
-        sync_id.clone(),
+    let _ = queries::work_queue::clear_all_pipelines(conn).await;
-        download_id.clone(),
-        index_id.clone(),
-        tag_id.clone(),
-        organize_id.clone(),
-        enrich_id.clone(),
-    ];
 
-    let state = state.clone();
+    // Step 1: Sync wanted items to download queue (fast, just DB inserts)
-    tokio::spawn(async move {
+    let sync_stats = shanty_dl::sync_wanted_to_queue(conn, false).await?;
-        run_pipeline_inner(
+    tracing::info!(
-            &state,
+        enqueued = sync_stats.enqueued,
-            &sync_id,
+        skipped = sync_stats.skipped,
-            &download_id,
+        pipeline_id = %pipeline_id,
-            &index_id,
+        "pipeline sync complete"
-            &tag_id,
+    );
-            &organize_id,
+
-            &enrich_id,
+    // Step 2: Create Download work items for each pending download_queue entry
+    let pending = queries::downloads::list(conn, Some(DownloadStatus::Pending)).await?;
+    for dl_item in &pending {
+        let payload = serde_json::json!({"download_queue_id": dl_item.id});
+        queries::work_queue::enqueue(
+            conn,
+            WorkTaskType::Download,
+            &payload.to_string(),
+            Some(&pipeline_id),
         )
-        .await;
+        .await?;
+    }
+
+    if !pending.is_empty() {
+        state.workers.notify(WorkTaskType::Download);
+    }
+
+    // Step 3: Scan library for existing files (import pipeline)
+    let index_payload = serde_json::json!({"scan_all": true});
+    queries::work_queue::enqueue(
+        conn,
+        WorkTaskType::Index,
+        &index_payload.to_string(),
+        Some(&pipeline_id),
+    )
+    .await?;
+    state.workers.notify(WorkTaskType::Index);
+
+    tracing::info!(
+        download_items = pending.len(),
+        pipeline_id = %pipeline_id,
+        "pipeline work items created (including library scan)"
+    );
+
+    Ok(pipeline_id)
+}
+
+/// Trigger the pipeline and return immediately (for API endpoints).
+pub fn spawn_pipeline(state: &web::Data<AppState>) -> String {
+    let state = state.clone();
+    let pipeline_id = uuid::Uuid::new_v4().to_string();
+
+    tokio::spawn(async move {
+        match trigger_pipeline(&state).await {
+            Ok(id) => tracing::info!(pipeline_id = %id, "pipeline triggered"),
+            Err(e) => tracing::error!(error = %e, "pipeline trigger failed"),
+        }
     });
 
-    task_ids
+    pipeline_id
-}
-
-/// Run the pipeline without registering tasks (for the scheduler, which logs instead).
-pub async fn run_pipeline(state: &web::Data<AppState>) -> Vec<String> {
-    let sync_id = state.tasks.register_pending("sync");
-    let download_id = state.tasks.register_pending("download");
-    let index_id = state.tasks.register_pending("index");
-    let tag_id = state.tasks.register_pending("tag");
-    let organize_id = state.tasks.register_pending("organize");
-    let enrich_id = state.tasks.register_pending("enrich");
-
-    let task_ids = vec![
-        sync_id.clone(),
-        download_id.clone(),
-        index_id.clone(),
-        tag_id.clone(),
-        organize_id.clone(),
-        enrich_id.clone(),
-    ];
-
-    run_pipeline_inner(
-        state,
-        &sync_id,
-        &download_id,
-        &index_id,
-        &tag_id,
-        &organize_id,
-        &enrich_id,
-    )
-    .await;
-
-    task_ids
-}
-
-async fn run_pipeline_inner(
-    state: &web::Data<AppState>,
-    sync_id: &str,
-    download_id: &str,
-    index_id: &str,
-    tag_id: &str,
-    organize_id: &str,
-    enrich_id: &str,
-) {
-    let cfg = state.config.read().await.clone();
-
-    // Step 1: Sync
-    state.tasks.start(sync_id);
-    state
-        .tasks
-        .update_progress(sync_id, 0, 0, "Syncing watchlist to download queue...");
-    match shanty_dl::sync_wanted_to_queue(state.db.conn(), false).await {
-        Ok(stats) => state.tasks.complete(sync_id, format!("{stats}")),
-        Err(e) => state.tasks.fail(sync_id, e.to_string()),
-    }
-
-    // Step 2: Download
-    state.tasks.start(download_id);
-    let cookies = cfg.download.cookies_path.clone();
-    let format: shanty_dl::AudioFormat = cfg
-        .download
-        .format
-        .parse()
-        .unwrap_or(shanty_dl::AudioFormat::Opus);
-    let source: shanty_dl::SearchSource = cfg
-        .download
-        .search_source
-        .parse()
-        .unwrap_or(shanty_dl::SearchSource::YouTubeMusic);
-    let rate = if cookies.is_some() {
-        cfg.download.rate_limit_auth
-    } else {
-        cfg.download.rate_limit
-    };
-    let backend = shanty_dl::YtDlpBackend::new(rate, source, cookies.clone());
-    let backend_config = shanty_dl::BackendConfig {
-        output_dir: cfg.download_path.clone(),
-        format,
-        cookies_path: cookies,
-    };
-    let task_state = state.clone();
-    let progress_tid = download_id.to_string();
-    let on_progress: shanty_dl::ProgressFn = Box::new(move |current, total, msg| {
-        task_state
-            .tasks
-            .update_progress(&progress_tid, current, total, msg);
-    });
-    match shanty_dl::run_queue_with_progress(
-        state.db.conn(),
-        &backend,
-        &backend_config,
-        false,
-        Some(on_progress),
-    )
-    .await
-    {
-        Ok(stats) => {
-            let _ = queries::cache::purge_prefix(state.db.conn(), "artist_totals:").await;
-            state.tasks.complete(download_id, format!("{stats}"));
-        }
-        Err(e) => state.tasks.fail(download_id, e.to_string()),
-    }
-
-    // Step 3: Index
-    state.tasks.start(index_id);
-    state
-        .tasks
-        .update_progress(index_id, 0, 0, "Scanning library...");
-    let scan_config = shanty_index::ScanConfig {
-        root: cfg.library_path.clone(),
-        dry_run: false,
-        concurrency: cfg.indexing.concurrency,
-    };
-    match shanty_index::run_scan(state.db.conn(), &scan_config).await {
-        Ok(stats) => state.tasks.complete(index_id, format!("{stats}")),
-        Err(e) => state.tasks.fail(index_id, e.to_string()),
-    }
-
-    // Step 4: Tag
-    state.tasks.start(tag_id);
-    state
-        .tasks
-        .update_progress(tag_id, 0, 0, "Tagging tracks...");
-    match shanty_tag::MusicBrainzClient::new() {
-        Ok(mb) => {
-            let tag_config = shanty_tag::TagConfig {
-                dry_run: false,
-                write_tags: cfg.tagging.write_tags,
-                confidence: cfg.tagging.confidence,
-            };
-            match shanty_tag::run_tagging(state.db.conn(), &mb, &tag_config, None).await {
-                Ok(stats) => state.tasks.complete(tag_id, format!("{stats}")),
-                Err(e) => state.tasks.fail(tag_id, e.to_string()),
-            }
-        }
-        Err(e) => state.tasks.fail(tag_id, e.to_string()),
-    }
-
-    // Step 5: Organize
-    state.tasks.start(organize_id);
-    state
-        .tasks
-        .update_progress(organize_id, 0, 0, "Organizing files...");
-    let org_config = shanty_org::OrgConfig {
-        target_dir: cfg.library_path.clone(),
-        format: cfg.organization_format.clone(),
-        dry_run: false,
-        copy: false,
-    };
-    match shanty_org::organize_from_db(state.db.conn(), &org_config).await {
-        Ok(stats) => {
-            let promoted = queries::wanted::promote_downloaded_to_owned(state.db.conn())
-                .await
-                .unwrap_or(0);
-            let msg = if promoted > 0 {
-                format!("{stats} — {promoted} items marked as owned")
-            } else {
-                format!("{stats}")
-            };
-            state.tasks.complete(organize_id, msg);
-        }
-        Err(e) => state.tasks.fail(organize_id, e.to_string()),
-    }
-
-    // Step 6: Enrich
-    state.tasks.start(enrich_id);
-    state
-        .tasks
-        .update_progress(enrich_id, 0, 0, "Refreshing artist data...");
-    match enrich_all_watched_artists(state).await {
-        Ok(count) => state
-            .tasks
-            .complete(enrich_id, format!("{count} artists refreshed")),
-        Err(e) => state.tasks.fail(enrich_id, e.to_string()),
-    }
 }

src/pipeline_scheduler.rs

@@ -1,59 +0,0 @@
-//! Background task that runs the full pipeline on a configurable interval.
-
-use std::time::Duration;
-
-use actix_web::web;
-use chrono::Utc;
-
-use crate::state::AppState;
-
-/// Spawn the pipeline scheduler background loop.
-///
-/// Sleeps for the configured interval, then runs the full pipeline if enabled.
-/// Reads config each iteration so changes take effect without restart.
-pub fn spawn(state: web::Data<AppState>) {
-    tokio::spawn(async move {
-        loop {
-            let (enabled, hours) = {
-                let cfg = state.config.read().await;
-                (
-                    cfg.scheduling.pipeline_enabled,
-                    cfg.scheduling.pipeline_interval_hours.max(1),
-                )
-            };
-
-            let sleep_secs = u64::from(hours) * 3600;
-
-            // Update next-run time
-            {
-                let mut sched = state.scheduler.lock().await;
-                sched.next_pipeline = if enabled {
-                    Some((Utc::now() + chrono::Duration::seconds(sleep_secs as i64)).naive_utc())
-                } else {
-                    None
-                };
-            }
-
-            tokio::time::sleep(Duration::from_secs(sleep_secs)).await;
-
-            if !enabled {
-                continue;
-            }
-
-            // Check if this run was skipped
-            {
-                let mut sched = state.scheduler.lock().await;
-                sched.next_pipeline = None;
-                if sched.skip_pipeline {
-                    sched.skip_pipeline = false;
-                    tracing::info!("scheduled pipeline skipped (user cancelled)");
-                    continue;
-                }
-            }
-
-            tracing::info!("scheduled pipeline starting");
-            let task_ids = crate::pipeline::run_pipeline(&state).await;
-            tracing::info!(?task_ids, "scheduled pipeline complete");
-        }
-    });
-}

src/routes/albums.rs

@@ -44,6 +44,7 @@ pub fn configure(cfg: &mut web::ServiceConfig) {
             .route(web::get().to(list_albums))
             .route(web::post().to(add_album)),
     )
+    .service(web::resource("/albums/{mbid}/watch").route(web::delete().to(unwatch_album)))
     .service(web::resource("/albums/{mbid}").route(web::get().to(get_album)));
 }
 
@@ -69,19 +70,32 @@ async fn get_album(
     let mbid = path.into_inner();
 
     // Try fetching as a release first
-    let mb_tracks = match state.mb_client.get_release_tracks(&mbid).await {
+    let (mb_tracks, _release_mbid) = match state.mb_client.get_release_tracks(&mbid).await {
-        Ok(tracks) => tracks,
+        Ok(tracks) => (tracks, mbid.clone()),
         Err(_) => {
             // Probably a release-group MBID. Browse releases for this group.
-            let release_mbid = resolve_release_from_group(&state, &mbid).await?;
+            let resolved = resolve_release_from_group(&state, &mbid).await?;
-            state
+            let tracks = state
                 .mb_client
-                .get_release_tracks(&release_mbid)
+                .get_release_tracks(&resolved)
                 .await
-                .map_err(|e| ApiError::Internal(format!("MusicBrainz error: {e}")))?
+                .map_err(|e| ApiError::Internal(format!("MusicBrainz error: {e}")))?;
+            (tracks, resolved)
         }
     };
 
+    // Get the album artist from the release's recording credits
+    let album_artist = if let Some(first_track) = mb_tracks.first() {
+        state
+            .mb_client
+            .get_recording(&first_track.recording_mbid)
+            .await
+            .ok()
+            .map(|r| r.artist)
+    } else {
+        None
+    };
+
     // Get all wanted items to check local status
     let all_wanted = queries::wanted::list(state.db.conn(), None, None).await?;
 
@@ -112,6 +126,7 @@ async fn get_album(
 
     Ok(HttpResponse::Ok().json(serde_json::json!({
         "mbid": mbid,
+        "artist": album_artist,
         "tracks": tracks,
     })))
 }
@@ -180,3 +195,34 @@ async fn add_album(
         "errors": summary.errors,
     })))
 }
+
+async fn unwatch_album(
+    state: web::Data<AppState>,
+    session: Session,
+    path: web::Path<String>,
+) -> Result<HttpResponse, ApiError> {
+    auth::require_auth(&session)?;
+    let mbid = path.into_inner();
+    let conn = state.db.conn();
+
+    // Get the album's tracks from MB to find their recording MBIDs
+    let tracks = match state.mb_client.get_release_tracks(&mbid).await {
+        Ok(t) => t,
+        Err(_) => {
+            // Try as release-group
+            let release_mbid = resolve_release_from_group(&state, &mbid).await?;
+            state
+                .mb_client
+                .get_release_tracks(&release_mbid)
+                .await
+                .map_err(|e| ApiError::Internal(format!("MusicBrainz error: {e}")))?
+        }
+    };
+
+    let mut removed = 0u64;
+    for track in &tracks {
+        removed += queries::wanted::remove_by_mbid(conn, &track.recording_mbid).await?;
+    }
+
+    Ok(HttpResponse::Ok().json(serde_json::json!({"removed": removed})))
+}

src/routes/artists.rs

@@ -34,6 +34,7 @@ struct ArtistListItem {
     name: String,
     musicbrainz_id: Option<String>,
     monitored: bool,
+    enriched: bool,
     total_watched: usize,
     total_owned: usize,
     total_items: usize,
@@ -78,6 +79,7 @@ pub fn configure(cfg: &mut web::ServiceConfig) {
             .route(web::post().to(set_monitored))
             .route(web::delete().to(unset_monitored)),
     )
+    .service(web::resource("/artists/{id}/watch").route(web::delete().to(unwatch_artist)))
     .service(
         web::resource("/artists/{id}")
             .route(web::get().to(get_artist))
@@ -92,15 +94,9 @@ async fn list_artists(
 ) -> Result<HttpResponse, ApiError> {
     auth::require_auth(&session)?;
     let artists = queries::artists::list(state.db.conn(), query.limit, query.offset).await?;
-    let wanted = queries::wanted::list(state.db.conn(), None, None).await?;
 
     let mut items: Vec<ArtistListItem> = Vec::new();
     for a in &artists {
-        let artist_wanted: Vec<_> = wanted
-            .iter()
-            .filter(|w| w.artist_id == Some(a.id))
-            .collect();
-
         // Check if we have cached artist-level totals from a prior detail page load
         let cache_key = format!("artist_totals:{}", a.id);
         let cached_totals: Option<(u32, u32, u32)> =
@@ -110,17 +106,12 @@ async fn list_artists(
                 None
             };
 
+        let enriched = cached_totals.is_some();
         let (total_watched, total_owned, total_items) =
             if let Some((avail, watched, owned)) = cached_totals {
                 (watched as usize, owned as usize, avail as usize)
             } else {
-                // Fall back to wanted item counts
+                (0, 0, 0)
-                let total_items = artist_wanted.len();
-                let total_owned = artist_wanted
-                    .iter()
-                    .filter(|w| w.status == WantedStatus::Owned)
-                    .count();
-                (total_items, total_owned, total_items)
             };
 
         items.push(ArtistListItem {
@@ -128,6 +119,7 @@ async fn list_artists(
             name: a.name.clone(),
             musicbrainz_id: a.musicbrainz_id.clone(),
             monitored: a.monitored,
+            enriched,
             total_watched,
             total_owned,
             total_items,
@@ -376,14 +368,18 @@ pub async fn enrich_artist(
     .await;
     tracing::debug!(mbid = %mbid, has_photo = artist_photo.is_some(), has_bio = artist_bio.is_some(), has_banner = artist_banner.is_some(), "artist enrichment data");
 
-    // Fetch release groups and filter by allowed secondary types
+    // Fetch release groups and split into primary vs featured
     let all_release_groups = state
         .search
         .get_release_groups(&mbid)
         .await
         .map_err(|e| ApiError::Internal(e.to_string()))?;
     let allowed = state.config.read().await.allowed_secondary_types.clone();
-    let release_groups: Vec<_> = all_release_groups
+
+    let (primary_rgs, featured_rgs): (Vec<_>, Vec<_>) =
+        all_release_groups.into_iter().partition(|rg| !rg.featured);
+
+    let release_groups: Vec<_> = primary_rgs
         .into_iter()
         .filter(|rg| {
             if rg.secondary_types.is_empty() {
@@ -395,6 +391,31 @@ pub async fn enrich_artist(
         })
         .collect();
 
+    // Featured release groups — just pass through with type filtering
+    let featured_albums: Vec<FullAlbumInfo> = featured_rgs
+        .iter()
+        .filter(|rg| {
+            if rg.secondary_types.is_empty() {
+                true
+            } else {
+                rg.secondary_types.iter().all(|st| allowed.contains(st))
+            }
+        })
+        .map(|rg| FullAlbumInfo {
+            mbid: rg.first_release_id.clone().unwrap_or_else(|| rg.id.clone()),
+            title: rg.title.clone(),
+            release_type: rg.primary_type.clone(),
+            date: rg.first_release_date.clone(),
+            track_count: 0,
+            local_album_id: None,
+            watched_tracks: 0,
+            owned_tracks: 0,
+            downloaded_tracks: 0,
+            total_local_tracks: 0,
+            status: "featured".to_string(),
+        })
+        .collect();
+
     // Get all wanted items for this artist
     let all_wanted = queries::wanted::list(state.db.conn(), None, None).await?;
     let artist_wanted: Vec<_> = all_wanted
@@ -609,6 +630,7 @@ pub async fn enrich_artist(
     Ok(serde_json::json!({
         "artist": artist,
         "albums": albums,
+        "featured_albums": featured_albums,
         "artist_status": artist_status,
         "total_available_tracks": total_available_tracks,
         "total_watched_tracks": total_artist_watched,
@@ -777,11 +799,13 @@ async fn add_artist(
     if body.name.is_none() && body.mbid.is_none() {
         return Err(ApiError::BadRequest("provide name or mbid".into()));
     }
+    let allowed = state.config.read().await.allowed_secondary_types.clone();
     let summary = shanty_watch::add_artist(
         state.db.conn(),
         body.name.as_deref(),
         body.mbid.as_deref(),
         &state.mb_client,
+        &allowed,
         Some(user_id),
     )
     .await?;
@@ -807,12 +831,31 @@ async fn delete_artist(
     session: Session,
     path: web::Path<i32>,
 ) -> Result<HttpResponse, ApiError> {
-    auth::require_admin(&session)?;
+    auth::require_auth(&session)?;
     let id = path.into_inner();
-    queries::artists::delete(state.db.conn(), id).await?;
+    let conn = state.db.conn();
+
+    // Cascade: remove wanted items, tracks (DB only), albums, cache, then artist
+    queries::wanted::remove_by_artist(conn, id).await?;
+    queries::tracks::delete_by_artist(conn, id).await?;
+    queries::albums::delete_by_artist(conn, id).await?;
+    let _ = queries::cache::purge_prefix(conn, &format!("artist_totals:{id}")).await;
+    queries::artists::delete(conn, id).await?;
+
     Ok(HttpResponse::NoContent().finish())
 }
 
+async fn unwatch_artist(
+    state: web::Data<AppState>,
+    session: Session,
+    path: web::Path<i32>,
+) -> Result<HttpResponse, ApiError> {
+    auth::require_auth(&session)?;
+    let id = path.into_inner();
+    let removed = queries::wanted::remove_by_artist(state.db.conn(), id).await?;
+    Ok(HttpResponse::Ok().json(serde_json::json!({"removed": removed})))
+}
+
 async fn set_monitored(
     state: web::Data<AppState>,
     session: Session,

src/routes/system.rs

@@ -3,12 +3,12 @@ use actix_web::{HttpResponse, web};
 use serde::Deserialize;
 
 use shanty_db::entities::download_queue::DownloadStatus;
+use shanty_db::entities::work_queue::WorkTaskType;
 use shanty_db::queries;
 
 use crate::auth;
 use crate::config::AppConfig;
 use crate::error::ApiError;
-use crate::routes::artists::enrich_all_watched_artists;
 use crate::state::AppState;
 
 pub fn configure(cfg: &mut web::ServiceConfig) {
@@ -38,13 +38,13 @@ async fn get_status(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_auth(&session)?;
-    let summary = shanty_watch::library_summary(state.db.conn()).await?;
+    let conn = state.db.conn();
-    let pending_items =
+
-        queries::downloads::list(state.db.conn(), Some(DownloadStatus::Pending)).await?;
+    let summary = shanty_watch::library_summary(conn).await?;
+    let pending_items = queries::downloads::list(conn, Some(DownloadStatus::Pending)).await?;
     let downloading_items =
-        queries::downloads::list(state.db.conn(), Some(DownloadStatus::Downloading)).await?;
+        queries::downloads::list(conn, Some(DownloadStatus::Downloading)).await?;
-    let failed_items =
+    let failed_items = queries::downloads::list(conn, Some(DownloadStatus::Failed)).await?;
-        queries::downloads::list(state.db.conn(), Some(DownloadStatus::Failed)).await?;
     let tasks = state.tasks.list();
 
     let mut queue_items = Vec::new();
@@ -52,15 +52,40 @@ async fn get_status(
     queue_items.extend(pending_items.iter().cloned());
     queue_items.extend(failed_items.iter().take(5).cloned());
 
-    let needs_tagging = queries::tracks::get_needing_metadata(state.db.conn()).await?;
+    let needs_tagging = queries::tracks::get_needing_metadata(conn).await?;
 
-    // Scheduled task info
+    // Work queue counts
-    let sched = state.scheduler.lock().await;
+    let work_queue = queries::work_queue::counts_all(conn).await.ok();
-    let scheduled_tasks = serde_json::json!({
+
-        "next_pipeline": sched.next_pipeline,
+    // Scheduler state from DB
-        "next_monitor": sched.next_monitor,
+    let scheduler_jobs = queries::scheduler_state::list_all(conn)
-    });
+        .await
-    drop(sched);
+        .unwrap_or_default();
+    let scheduler_json: serde_json::Value = scheduler_jobs
+        .iter()
+        .map(|j| {
+            (
+                j.job_name.clone(),
+                serde_json::json!({
+                    "last_run": j.last_run_at,
+                    "next_run": j.next_run_at,
+                    "last_result": j.last_result,
+                    "enabled": j.enabled,
+                }),
+            )
+        })
+        .collect::<serde_json::Map<String, serde_json::Value>>()
+        .into();
+
+    // Backward-compatible scheduled field (from scheduler_state DB)
+    let next_pipeline = scheduler_jobs
+        .iter()
+        .find(|j| j.job_name == "pipeline")
+        .and_then(|j| j.next_run_at);
+    let next_monitor = scheduler_jobs
+        .iter()
+        .find(|j| j.job_name == "monitor")
+        .and_then(|j| j.next_run_at);
 
     Ok(HttpResponse::Ok().json(serde_json::json!({
         "library": summary,
@@ -75,7 +100,12 @@ async fn get_status(
             "items": needs_tagging.iter().take(20).collect::<Vec<_>>(),
         },
         "tasks": tasks,
-        "scheduled": scheduled_tasks,
+        "scheduled": {
+            "next_pipeline": next_pipeline,
+            "next_monitor": next_monitor,
+        },
+        "work_queue": work_queue,
+        "scheduler": scheduler_json,
     })))
 }
 
@@ -84,27 +114,16 @@ async fn trigger_index(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_auth(&session)?;
-    let task_id = state.tasks.register("index");
+    let payload = serde_json::json!({"scan_all": true});
-    let state = state.clone();
+    let item = queries::work_queue::enqueue(
-    let tid = task_id.clone();
+        state.db.conn(),
-
+        WorkTaskType::Index,
-    tokio::spawn(async move {
+        &payload.to_string(),
-        let cfg = state.config.read().await.clone();
+        None,
-        state
+    )
-            .tasks
+    .await?;
-            .update_progress(&tid, 0, 0, "Scanning library...");
+    state.workers.notify(WorkTaskType::Index);
-        let scan_config = shanty_index::ScanConfig {
+    Ok(HttpResponse::Accepted().json(serde_json::json!({ "work_item_id": item.id })))
-            root: cfg.library_path.clone(),
-            dry_run: false,
-            concurrency: cfg.indexing.concurrency,
-        };
-        match shanty_index::run_scan(state.db.conn(), &scan_config).await {
-            Ok(stats) => state.tasks.complete(&tid, format!("{stats}")),
-            Err(e) => state.tasks.fail(&tid, e.to_string()),
-        }
-    });
-
-    Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id })))
 }
 
 async fn trigger_tag(
@@ -112,35 +131,18 @@ async fn trigger_tag(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_auth(&session)?;
-    let task_id = state.tasks.register("tag");
+    let conn = state.db.conn();
-    let state = state.clone();
+    let untagged = queries::tracks::get_needing_metadata(conn).await?;
-    let tid = task_id.clone();
+    let mut count = 0;
-
+    for track in &untagged {
-    tokio::spawn(async move {
+        let payload = serde_json::json!({"track_id": track.id});
-        let cfg = state.config.read().await.clone();
+        queries::work_queue::enqueue(conn, WorkTaskType::Tag, &payload.to_string(), None).await?;
-        state
+        count += 1;
-            .tasks
+    }
-            .update_progress(&tid, 0, 0, "Preparing tagger...");
+    if count > 0 {
-        let mb = match shanty_tag::MusicBrainzClient::new() {
+        state.workers.notify(WorkTaskType::Tag);
-            Ok(c) => c,
+    }
-            Err(e) => {
+    Ok(HttpResponse::Accepted().json(serde_json::json!({ "enqueued": count })))
-                state.tasks.fail(&tid, e.to_string());
-                return;
-            }
-        };
-        let tag_config = shanty_tag::TagConfig {
-            dry_run: false,
-            write_tags: cfg.tagging.write_tags,
-            confidence: cfg.tagging.confidence,
-        };
-        state.tasks.update_progress(&tid, 0, 0, "Tagging tracks...");
-        match shanty_tag::run_tagging(state.db.conn(), &mb, &tag_config, None).await {
-            Ok(stats) => state.tasks.complete(&tid, format!("{stats}")),
-            Err(e) => state.tasks.fail(&tid, e.to_string()),
-        }
-    });
-
-    Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id })))
 }
 
 async fn trigger_organize(
@@ -148,39 +150,26 @@ async fn trigger_organize(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_auth(&session)?;
-    let task_id = state.tasks.register("organize");
+    let conn = state.db.conn();
-    let state = state.clone();
+    let mut count = 0u64;
-    let tid = task_id.clone();
+    let mut offset = 0u64;
-
+    loop {
-    tokio::spawn(async move {
+        let tracks = queries::tracks::list(conn, 500, offset).await?;
-        let cfg = state.config.read().await.clone();
+        if tracks.is_empty() {
-        state
+            break;
-            .tasks
-            .update_progress(&tid, 0, 0, "Organizing files...");
-        let org_config = shanty_org::OrgConfig {
-            target_dir: cfg.library_path.clone(),
-            format: cfg.organization_format.clone(),
-            dry_run: false,
-            copy: false,
-        };
-        match shanty_org::organize_from_db(state.db.conn(), &org_config).await {
-            Ok(stats) => {
-                let promoted = queries::wanted::promote_downloaded_to_owned(state.db.conn())
-                    .await
-                    .unwrap_or(0);
-                let msg = if promoted > 0 {
-                    format!("{stats} — {promoted} items marked as owned")
-                } else {
-                    format!("{stats}")
-                };
-                state.tasks.complete(&tid, msg);
-                let _ = enrich_all_watched_artists(&state).await;
-            }
-            Err(e) => state.tasks.fail(&tid, e.to_string()),
         }
-    });
+        for track in &tracks {
-
+            let payload = serde_json::json!({"track_id": track.id});
-    Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id })))
+            queries::work_queue::enqueue(conn, WorkTaskType::Organize, &payload.to_string(), None)
+                .await?;
+            count += 1;
+        }
+        offset += 500;
+    }
+    if count > 0 {
+        state.workers.notify(WorkTaskType::Organize);
+    }
+    Ok(HttpResponse::Accepted().json(serde_json::json!({ "enqueued": count })))
 }
 
 async fn trigger_pipeline(
@@ -188,8 +177,8 @@ async fn trigger_pipeline(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_auth(&session)?;
-    let task_ids = crate::pipeline::spawn_pipeline(&state);
+    let pipeline_id = crate::pipeline::trigger_pipeline(&state).await?;
-    Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_ids": task_ids })))
+    Ok(HttpResponse::Accepted().json(serde_json::json!({ "pipeline_id": pipeline_id })))
 }
 
 async fn get_task(
@@ -313,10 +302,13 @@ async fn skip_pipeline(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_admin(&session)?;
-    let mut sched = state.scheduler.lock().await;
+    // Push next_run_at forward by one interval
-    sched.skip_pipeline = true;
+    let cfg = state.config.read().await;
-    sched.next_pipeline = None;
+    let hours = cfg.scheduling.pipeline_interval_hours.max(1);
-    Ok(HttpResponse::Ok().json(serde_json::json!({"status": "skipped"})))
+    drop(cfg);
+    let next = chrono::Utc::now().naive_utc() + chrono::Duration::hours(i64::from(hours));
+    queries::scheduler_state::update_next_run(state.db.conn(), "pipeline", Some(next)).await?;
+    Ok(HttpResponse::Ok().json(serde_json::json!({"status": "skipped", "next_run": next})))
 }
 
 async fn skip_monitor(
@@ -324,10 +316,12 @@ async fn skip_monitor(
     session: Session,
 ) -> Result<HttpResponse, ApiError> {
     auth::require_admin(&session)?;
-    let mut sched = state.scheduler.lock().await;
+    let cfg = state.config.read().await;
-    sched.skip_monitor = true;
+    let hours = cfg.scheduling.monitor_interval_hours.max(1);
-    sched.next_monitor = None;
+    drop(cfg);
-    Ok(HttpResponse::Ok().json(serde_json::json!({"status": "skipped"})))
+    let next = chrono::Utc::now().naive_utc() + chrono::Duration::hours(i64::from(hours));
+    queries::scheduler_state::update_next_run(state.db.conn(), "monitor", Some(next)).await?;
+    Ok(HttpResponse::Ok().json(serde_json::json!({"status": "skipped", "next_run": next})))
 }
 
 async fn get_mb_status(
@@ -389,11 +383,15 @@ async fn trigger_mb_import(
             state.tasks.update_progress(
                 &tid,
                 i as u64,
-                4 + 4, // 4 downloads + 4 imports
+                4 + 4,
                 &format!("Downloading {filename}..."),
             );
             if let Err(e) =
-                shanty_data::mb_import::download_dump(filename, &timestamp, &data_dir, |_| {}).await
+                shanty_data::mb_import::download_dump(filename, &timestamp, &data_dir, |msg| {
+                    tracing::info!("{msg}");
+                    state.tasks.update_progress(&tid, i as u64, 8, msg);
+                })
+                .await
             {
                 state
                     .tasks
@@ -409,9 +407,9 @@ async fn trigger_mb_import(
 
         let tid_clone = tid.clone();
         let state_clone = state.clone();
-        // Run import in blocking task since rusqlite is sync
         let result = tokio::task::spawn_blocking(move || {
             shanty_data::mb_import::run_import_at_path(&db_path, &data_dir, |msg| {
+                tracing::info!("{msg}");
                 state_clone.tasks.update_progress(&tid_clone, 4, 8, msg);
             })
         })
@@ -419,6 +417,7 @@ async fn trigger_mb_import(
 
         match result {
             Ok(Ok(stats)) => {
+                tracing::info!(%stats, "MusicBrainz import complete");
                 state.tasks.complete(&tid, format!("{stats}"));
             }
             Ok(Err(e)) => {

src/routes/tracks.rs

@@ -21,9 +21,21 @@ pub struct SearchParams {
     offset: u64,
 }
 
+#[derive(Deserialize)]
+pub struct WatchTrackRequest {
+    artist: Option<String>,
+    title: Option<String>,
+    mbid: Option<String>,
+}
+
 pub fn configure(cfg: &mut web::ServiceConfig) {
-    cfg.service(web::resource("/tracks").route(web::get().to(list_tracks)))
+    cfg.service(
-        .service(web::resource("/tracks/{id}").route(web::get().to(get_track)));
+        web::resource("/tracks/watch")
+            .route(web::post().to(watch_track))
+            .route(web::delete().to(unwatch_track)),
+    )
+    .service(web::resource("/tracks").route(web::get().to(list_tracks)))
+    .service(web::resource("/tracks/{id}").route(web::get().to(get_track)));
 }
 
 async fn list_tracks(
@@ -50,3 +62,46 @@ async fn get_track(
     let track = queries::tracks::get_by_id(state.db.conn(), id).await?;
     Ok(HttpResponse::Ok().json(track))
 }
+
+async fn watch_track(
+    state: web::Data<AppState>,
+    session: Session,
+    body: web::Json<WatchTrackRequest>,
+) -> Result<HttpResponse, ApiError> {
+    let (user_id, _, _) = auth::require_auth(&session)?;
+    if body.title.is_none() && body.mbid.is_none() {
+        return Err(ApiError::BadRequest(
+            "provide title or recording mbid".into(),
+        ));
+    }
+    let entry = shanty_watch::add_track(
+        state.db.conn(),
+        body.artist.as_deref(),
+        body.title.as_deref(),
+        body.mbid.as_deref(),
+        &state.mb_client,
+        Some(user_id),
+    )
+    .await?;
+
+    Ok(HttpResponse::Ok().json(serde_json::json!({
+        "id": entry.id,
+        "status": entry.status,
+        "name": entry.name,
+        "artist_name": entry.artist_name,
+    })))
+}
+
+async fn unwatch_track(
+    state: web::Data<AppState>,
+    session: Session,
+    body: web::Json<WatchTrackRequest>,
+) -> Result<HttpResponse, ApiError> {
+    auth::require_auth(&session)?;
+    let mbid = body
+        .mbid
+        .as_deref()
+        .ok_or_else(|| ApiError::BadRequest("provide recording mbid".into()))?;
+    let removed = queries::wanted::remove_by_mbid(state.db.conn(), mbid).await?;
+    Ok(HttpResponse::Ok().json(serde_json::json!({"removed": removed})))
+}

src/scheduler.rs

@@ -0,0 +1,161 @@
+//! Unified scheduler that manages all recurring background jobs.
+//!
+//! Replaces the separate pipeline_scheduler, monitor, and cookie_refresh spawn loops
+//! with a single loop backed by persistent state in the `scheduler_state` DB table.
+
+use std::time::Duration;
+
+use actix_web::web;
+use chrono::Utc;
+
+use shanty_db::queries;
+
+use crate::state::AppState;
+
+/// Spawn the unified scheduler background loop.
+pub fn spawn(state: web::Data<AppState>) {
+    tokio::spawn(async move {
+        // Initialize scheduler state rows in DB with next_run_at pre-populated
+        for job_name in ["pipeline", "monitor", "cookie_refresh"] {
+            match queries::scheduler_state::get_or_create(state.db.conn(), job_name).await {
+                Ok(job) => {
+                    if job.next_run_at.is_none() {
+                        let (enabled, interval_secs) = read_job_config(&state, job_name).await;
+                        if enabled {
+                            let next =
+                                Utc::now().naive_utc() + chrono::Duration::seconds(interval_secs);
+                            let _ = queries::scheduler_state::update_next_run(
+                                state.db.conn(),
+                                job_name,
+                                Some(next),
+                            )
+                            .await;
+                        }
+                    }
+                }
+                Err(e) => {
+                    tracing::error!(job = job_name, error = %e, "failed to init scheduler state");
+                }
+            }
+        }
+
+        loop {
+            // Check each job
+            check_and_run_job(&state, "pipeline", run_pipeline_job).await;
+            check_and_run_job(&state, "monitor", run_monitor_job).await;
+            check_and_run_job(&state, "cookie_refresh", run_cookie_refresh_job).await;
+
+            // Poll every 30 seconds
+            tokio::time::sleep(Duration::from_secs(30)).await;
+        }
+    });
+}
+
+async fn check_and_run_job<F, Fut>(state: &web::Data<AppState>, job_name: &str, run_fn: F)
+where
+    F: FnOnce(web::Data<AppState>) -> Fut,
+    Fut: std::future::Future<Output = Result<String, String>>,
+{
+    let job = match queries::scheduler_state::get_or_create(state.db.conn(), job_name).await {
+        Ok(j) => j,
+        Err(e) => {
+            tracing::error!(job = job_name, error = %e, "failed to read scheduler state");
+            return;
+        }
+    };
+
+    // Read config to check if enabled and get interval
+    let (config_enabled, interval_secs) = read_job_config(state, job_name).await;
+
+    // If config says disabled, ensure DB state reflects it
+    if !config_enabled {
+        if job.enabled {
+            let _ = queries::scheduler_state::set_enabled(state.db.conn(), job_name, false).await;
+            let _ =
+                queries::scheduler_state::update_next_run(state.db.conn(), job_name, None).await;
+        }
+        return;
+    }
+
+    // If DB says disabled (e.g. user skipped), re-enable from config
+    if !job.enabled {
+        let _ = queries::scheduler_state::set_enabled(state.db.conn(), job_name, true).await;
+    }
+
+    let now = Utc::now().naive_utc();
+
+    // If no next_run_at is set, schedule one
+    if job.next_run_at.is_none() {
+        let next = now + chrono::Duration::seconds(interval_secs);
+        let _ =
+            queries::scheduler_state::update_next_run(state.db.conn(), job_name, Some(next)).await;
+        return;
+    }
+
+    // Check if it's time to run
+    let next_run = job.next_run_at.unwrap();
+    if now < next_run {
+        return;
+    }
+
+    // Time to run!
+    tracing::info!(job = job_name, "scheduled job starting");
+
+    let result = run_fn(state.clone()).await;
+
+    let result_str = match &result {
+        Ok(msg) => {
+            tracing::info!(job = job_name, result = %msg, "scheduled job complete");
+            format!("ok: {msg}")
+        }
+        Err(e) => {
+            tracing::error!(job = job_name, error = %e, "scheduled job failed");
+            format!("error: {e}")
+        }
+    };
+
+    // Update last run and schedule next
+    let _ = queries::scheduler_state::update_last_run(state.db.conn(), job_name, &result_str).await;
+    let next = Utc::now().naive_utc() + chrono::Duration::seconds(interval_secs);
+    let _ = queries::scheduler_state::update_next_run(state.db.conn(), job_name, Some(next)).await;
+}
+
+async fn read_job_config(state: &web::Data<AppState>, job_name: &str) -> (bool, i64) {
+    let cfg = state.config.read().await;
+    match job_name {
+        "pipeline" => (
+            cfg.scheduling.pipeline_enabled,
+            i64::from(cfg.scheduling.pipeline_interval_hours.max(1)) * 3600,
+        ),
+        "monitor" => (
+            cfg.scheduling.monitor_enabled,
+            i64::from(cfg.scheduling.monitor_interval_hours.max(1)) * 3600,
+        ),
+        "cookie_refresh" => (
+            cfg.download.cookie_refresh_enabled,
+            i64::from(cfg.download.cookie_refresh_hours.max(1)) * 3600,
+        ),
+        _ => (false, 3600),
+    }
+}
+
+async fn run_pipeline_job(state: web::Data<AppState>) -> Result<String, String> {
+    let pipeline_id = crate::pipeline::trigger_pipeline(&state)
+        .await
+        .map_err(|e| e.to_string())?;
+    Ok(format!("pipeline triggered: {pipeline_id}"))
+}
+
+async fn run_monitor_job(state: web::Data<AppState>) -> Result<String, String> {
+    let stats = crate::monitor::check_monitored_artists(&state)
+        .await
+        .map_err(|e| e.to_string())?;
+    Ok(format!(
+        "{} artists checked, {} new releases, {} tracks added",
+        stats.artists_checked, stats.new_releases_found, stats.tracks_added
+    ))
+}
+
+async fn run_cookie_refresh_job(_state: web::Data<AppState>) -> Result<String, String> {
+    crate::cookie_refresh::run_refresh().await
+}

src/state.rs

@@ -8,24 +8,13 @@ use shanty_search::MusicBrainzSearch;
 
 use crate::config::AppConfig;
 use crate::tasks::TaskManager;
+use crate::workers::WorkerManager;
 
 /// Tracks an active Firefox login session for YouTube auth.
 pub struct FirefoxLoginSession {
     pub vnc_url: String,
 }
 
-/// Tracks next-run times for scheduled background tasks.
-pub struct SchedulerInfo {
-    /// When the next pipeline run is scheduled (None if disabled).
-    pub next_pipeline: Option<chrono::NaiveDateTime>,
-    /// When the next monitor check is scheduled (None if disabled).
-    pub next_monitor: Option<chrono::NaiveDateTime>,
-    /// Skip the next pipeline run (one-shot, resets after skip).
-    pub skip_pipeline: bool,
-    /// Skip the next monitor run (one-shot, resets after skip).
-    pub skip_monitor: bool,
-}
-
 pub struct AppState {
     pub db: Database,
     pub mb_client: HybridMusicBrainzFetcher,
@@ -34,6 +23,6 @@ pub struct AppState {
     pub config: Arc<RwLock<AppConfig>>,
     pub config_path: Option<String>,
     pub tasks: TaskManager,
+    pub workers: WorkerManager,
     pub firefox_login: Mutex<Option<FirefoxLoginSession>>,
-    pub scheduler: Mutex<SchedulerInfo>,
 }

src/workers.rs

@@ -0,0 +1,535 @@
+//! Work queue workers that process pipeline items concurrently.
+//!
+//! Each task type (Download, Index, Tag, Organize) has a dedicated worker loop
+//! that polls the work_queue table and processes items with bounded concurrency.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use actix_web::web;
+use sea_orm::ActiveValue::Set;
+use tokio::sync::{Notify, Semaphore};
+
+use shanty_db::entities::download_queue::DownloadStatus;
+use shanty_db::entities::wanted_item::WantedStatus;
+use shanty_db::entities::work_queue::WorkTaskType;
+use shanty_db::queries;
+use shanty_dl::DownloadBackend;
+
+use crate::state::AppState;
+
+/// Manages worker notification channels and spawns worker loops.
+pub struct WorkerManager {
+    notifiers: HashMap<WorkTaskType, Arc<Notify>>,
+}
+
+impl Default for WorkerManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl WorkerManager {
+    pub fn new() -> Self {
+        let mut notifiers = HashMap::new();
+        notifiers.insert(WorkTaskType::Download, Arc::new(Notify::new()));
+        notifiers.insert(WorkTaskType::Index, Arc::new(Notify::new()));
+        notifiers.insert(WorkTaskType::Tag, Arc::new(Notify::new()));
+        notifiers.insert(WorkTaskType::Organize, Arc::new(Notify::new()));
+        notifiers.insert(WorkTaskType::Enrich, Arc::new(Notify::new()));
+        Self { notifiers }
+    }
+
+    /// Wake the worker for a specific task type.
+    pub fn notify(&self, task_type: WorkTaskType) {
+        if let Some(n) = self.notifiers.get(&task_type) {
+            n.notify_one();
+        }
+    }
+
+    /// Spawn all worker loops and run startup recovery.
+    pub fn spawn_all(state: web::Data<AppState>) {
+        let state_clone = state.clone();
+        tokio::spawn(async move {
+            // Reset any items stuck in Running from a previous crash
+            match queries::work_queue::reset_stale_running(state_clone.db.conn()).await {
+                Ok(count) if count > 0 => {
+                    tracing::info!(count, "reset stale running work queue items");
+                }
+                Err(e) => tracing::error!(error = %e, "failed to reset stale work queue items"),
+                _ => {}
+            }
+
+            // Periodic cleanup of old completed items (every 6 hours)
+            let cleanup_state = state_clone.clone();
+            tokio::spawn(async move {
+                loop {
+                    tokio::time::sleep(std::time::Duration::from_secs(6 * 3600)).await;
+                    let _ =
+                        queries::work_queue::cleanup_completed(cleanup_state.db.conn(), 7).await;
+                }
+            });
+
+            // Read config for concurrency settings and spawn workers
+            let cfg = state_clone.config.read().await.clone();
+            spawn_worker(state_clone.clone(), WorkTaskType::Download, 1);
+            spawn_worker(
+                state_clone.clone(),
+                WorkTaskType::Index,
+                cfg.indexing.concurrency,
+            );
+            spawn_worker(
+                state_clone.clone(),
+                WorkTaskType::Tag,
+                cfg.tagging.concurrency,
+            );
+            spawn_worker(state_clone.clone(), WorkTaskType::Organize, 4);
+            spawn_worker(state_clone.clone(), WorkTaskType::Enrich, 2);
+        });
+    }
+}
+
+fn spawn_worker(state: web::Data<AppState>, task_type: WorkTaskType, concurrency: usize) {
+    let notify = state
+        .workers
+        .notifiers
+        .get(&task_type)
+        .cloned()
+        .unwrap_or_else(|| Arc::new(Notify::new()));
+
+    tokio::spawn(async move {
+        loop {
+            // Claim pending items
+            let items = match queries::work_queue::claim_next(
+                state.db.conn(),
+                task_type,
+                concurrency as u64,
+            )
+            .await
+            {
+                Ok(items) => items,
+                Err(e) => {
+                    tracing::error!(task_type = %task_type, error = %e, "failed to claim work items");
+                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+                    continue;
+                }
+            };
+
+            if items.is_empty() {
+                // Nothing to do — wait for notification or poll timeout
+                tokio::select! {
+                    _ = notify.notified() => {}
+                    _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {}
+                }
+                continue;
+            }
+
+            let semaphore = Arc::new(Semaphore::new(concurrency));
+            let mut handles = Vec::new();
+
+            for item in items {
+                let permit = semaphore.clone().acquire_owned().await.unwrap();
+                let state = state.clone();
+                let item_id = item.id;
+                let item_task_type = item.task_type;
+                let pipeline_id = item.pipeline_id.clone();
+
+                handles.push(tokio::spawn(async move {
+                    let _permit = permit;
+
+                    let result = match item_task_type {
+                        WorkTaskType::Download => process_download(&state, &item).await,
+                        WorkTaskType::Index => process_index(&state, &item).await,
+                        WorkTaskType::Tag => process_tag(&state, &item).await,
+                        WorkTaskType::Organize => process_organize(&state, &item).await,
+                        WorkTaskType::Enrich => process_enrich(&state, &item).await,
+                    };
+
+                    match result {
+                        Ok(downstream) => {
+                            if let Err(e) =
+                                queries::work_queue::complete(state.db.conn(), item_id).await
+                            {
+                                tracing::error!(id = item_id, error = %e, "failed to mark work item complete");
+                            }
+                            // Enqueue downstream items
+                            for (task_type, payload) in downstream {
+                                if let Err(e) = queries::work_queue::enqueue(
+                                    state.db.conn(),
+                                    task_type,
+                                    &payload,
+                                    pipeline_id.as_deref(),
+                                )
+                                .await
+                                {
+                                    tracing::error!(
+                                        error = %e,
+                                        "failed to enqueue downstream work item"
+                                    );
+                                }
+                                state.workers.notify(task_type);
+                            }
+                        }
+                        Err(error) => {
+                            tracing::warn!(
+                                id = item_id,
+                                task_type = %item_task_type,
+                                error = %error,
+                                "work item failed"
+                            );
+                            if let Err(e) =
+                                queries::work_queue::fail(state.db.conn(), item_id, &error).await
+                            {
+                                tracing::error!(id = item_id, error = %e, "failed to mark work item failed");
+                            }
+                        }
+                    }
+                }));
+            }
+
+            for handle in handles {
+                let _ = handle.await;
+            }
+        }
+    });
+}
+
+// --- Worker implementations ---
+
+type WorkResult = Result<Vec<(WorkTaskType, String)>, String>;
+
+async fn process_download(
+    state: &web::Data<AppState>,
+    item: &shanty_db::entities::work_queue::Model,
+) -> WorkResult {
+    let payload: serde_json::Value =
+        serde_json::from_str(&item.payload_json).map_err(|e| e.to_string())?;
+    let dl_queue_id = payload
+        .get("download_queue_id")
+        .and_then(|v| v.as_i64())
+        .map(|v| v as i32);
+
+    let conn = state.db.conn();
+
+    // Get the download queue item — either specific ID or next pending
+    let dl_item = if let Some(id) = dl_queue_id {
+        queries::downloads::list(conn, None)
+            .await
+            .map_err(|e| e.to_string())?
+            .into_iter()
+            .find(|i| i.id == id)
+    } else {
+        queries::downloads::get_next_pending(conn)
+            .await
+            .map_err(|e| e.to_string())?
+    };
+
+    let dl_item = match dl_item {
+        Some(item) => item,
+        None => return Ok(vec![]), // Nothing to download
+    };
+
+    // Mark as downloading
+    queries::downloads::update_status(conn, dl_item.id, DownloadStatus::Downloading, None)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Build download backend from config
+    let cfg = state.config.read().await.clone();
+    let cookies = cfg.download.cookies_path.clone();
+    let format: shanty_dl::AudioFormat = cfg
+        .download
+        .format
+        .parse()
+        .unwrap_or(shanty_dl::AudioFormat::Opus);
+    let source: shanty_dl::SearchSource = cfg
+        .download
+        .search_source
+        .parse()
+        .unwrap_or(shanty_dl::SearchSource::YouTubeMusic);
+    let rate = if cookies.is_some() {
+        cfg.download.rate_limit_auth
+    } else {
+        cfg.download.rate_limit
+    };
+
+    let backend = shanty_dl::YtDlpBackend::new(rate, source, cookies.clone());
+    let backend_config = shanty_dl::BackendConfig {
+        output_dir: cfg.download_path.clone(),
+        format,
+        cookies_path: cookies,
+    };
+
+    // Determine download target
+    let target = if let Some(ref url) = dl_item.source_url {
+        shanty_dl::DownloadTarget::Url(url.clone())
+    } else {
+        shanty_dl::DownloadTarget::Query(dl_item.query.clone())
+    };
+
+    // Execute download
+    let result = backend
+        .download(&target, &backend_config)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Mark download_queue entry as completed
+    queries::downloads::update_status(conn, dl_item.id, DownloadStatus::Completed, None)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Create track record if linked to a wanted item (same logic as shanty-dl queue.rs)
+    let mut downstream = Vec::new();
+    if let Some(wanted_item_id) = dl_item.wanted_item_id {
+        let wanted = queries::wanted::get_by_id(conn, wanted_item_id)
+            .await
+            .map_err(|e| e.to_string())?;
+
+        let file_size = std::fs::metadata(&result.file_path)
+            .map(|m| m.len() as i64)
+            .unwrap_or(0);
+
+        let now = chrono::Utc::now().naive_utc();
+        let track_active = shanty_db::entities::track::ActiveModel {
+            file_path: Set(result.file_path.to_string_lossy().to_string()),
+            title: Set(Some(result.title.clone())),
+            artist: Set(result.artist.clone()),
+            file_size: Set(file_size),
+            musicbrainz_id: Set(wanted.musicbrainz_id.clone()),
+            artist_id: Set(wanted.artist_id),
+            added_at: Set(now),
+            updated_at: Set(now),
+            ..Default::default()
+        };
+        let track = queries::tracks::upsert(conn, track_active)
+            .await
+            .map_err(|e| e.to_string())?;
+
+        queries::wanted::update_status(conn, wanted_item_id, WantedStatus::Downloaded)
+            .await
+            .map_err(|e| e.to_string())?;
+
+        // Create Tag work item for this track
+        let tag_payload = serde_json::json!({"track_id": track.id});
+        downstream.push((WorkTaskType::Tag, tag_payload.to_string()));
+    }
+
+    let _ = queries::cache::purge_prefix(conn, "artist_totals:").await;
+    Ok(downstream)
+}
+
+async fn process_index(
+    state: &web::Data<AppState>,
+    item: &shanty_db::entities::work_queue::Model,
+) -> WorkResult {
+    let payload: serde_json::Value =
+        serde_json::from_str(&item.payload_json).map_err(|e| e.to_string())?;
+    let conn = state.db.conn();
+    let cfg = state.config.read().await.clone();
+    let mut downstream = Vec::new();
+
+    if payload
+        .get("scan_all")
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false)
+    {
+        // Full library scan
+        let scan_config = shanty_index::ScanConfig {
+            root: cfg.library_path.clone(),
+            dry_run: false,
+            concurrency: cfg.indexing.concurrency,
+        };
+        shanty_index::run_scan(conn, &scan_config)
+            .await
+            .map_err(|e| e.to_string())?;
+
+        // Create Tag work items for tracks not yet tagged by shanty
+        let untagged = queries::tracks::get_untagged(conn)
+            .await
+            .map_err(|e| e.to_string())?;
+        for track in &untagged {
+            let tag_payload = serde_json::json!({"track_id": track.id});
+            downstream.push((WorkTaskType::Tag, tag_payload.to_string()));
+        }
+    } else if let Some(file_path) = payload.get("file_path").and_then(|v| v.as_str()) {
+        // Single file index
+        let path = std::path::PathBuf::from(file_path);
+        let track_id = shanty_index::index_file(conn, &path, false)
+            .await
+            .map_err(|e| e.to_string())?;
+
+        if let Some(id) = track_id {
+            let tag_payload = serde_json::json!({"track_id": id});
+            downstream.push((WorkTaskType::Tag, tag_payload.to_string()));
+        }
+    }
+
+    Ok(downstream)
+}
+
+async fn process_tag(
+    state: &web::Data<AppState>,
+    item: &shanty_db::entities::work_queue::Model,
+) -> WorkResult {
+    let payload: serde_json::Value =
+        serde_json::from_str(&item.payload_json).map_err(|e| e.to_string())?;
+    let track_id = payload
+        .get("track_id")
+        .and_then(|v| v.as_i64())
+        .ok_or("missing track_id in payload")? as i32;
+
+    let conn = state.db.conn();
+    let cfg = state.config.read().await.clone();
+
+    let track = queries::tracks::get_by_id(conn, track_id)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    let tag_config = shanty_tag::TagConfig {
+        dry_run: false,
+        write_tags: cfg.tagging.write_tags,
+        confidence: cfg.tagging.confidence,
+    };
+
+    shanty_tag::tag_track(conn, &state.mb_client, &track, &tag_config)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Re-read the track to get the MBID set by tagging
+    let track = queries::tracks::get_by_id(conn, track_id)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Ensure a wanted_item exists for this track (marks imported files as Owned)
+    if let Some(ref mbid) = track.musicbrainz_id
+        && queries::wanted::find_by_mbid(conn, mbid)
+            .await
+            .map_err(|e| e.to_string())?
+            .is_none()
+    {
+        let item = queries::wanted::add(
+            conn,
+            queries::wanted::AddWantedItem {
+                item_type: shanty_db::entities::wanted_item::ItemType::Track,
+                name: track.title.as_deref().unwrap_or("Unknown"),
+                musicbrainz_id: Some(mbid),
+                artist_id: track.artist_id,
+                album_id: track.album_id,
+                track_id: Some(track.id),
+                user_id: None,
+            },
+        )
+        .await
+        .map_err(|e| e.to_string())?;
+
+        // Mark as Owned immediately since the file already exists
+        let _ = queries::wanted::update_status(
+            conn,
+            item.id,
+            shanty_db::entities::wanted_item::WantedStatus::Owned,
+        )
+        .await;
+    }
+
+    // Create Organize work item
+    let org_payload = serde_json::json!({"track_id": track_id});
+    Ok(vec![(WorkTaskType::Organize, org_payload.to_string())])
+}
+
+async fn process_organize(
+    state: &web::Data<AppState>,
+    item: &shanty_db::entities::work_queue::Model,
+) -> WorkResult {
+    let payload: serde_json::Value =
+        serde_json::from_str(&item.payload_json).map_err(|e| e.to_string())?;
+    let track_id = payload
+        .get("track_id")
+        .and_then(|v| v.as_i64())
+        .ok_or("missing track_id in payload")? as i32;
+
+    let conn = state.db.conn();
+    let cfg = state.config.read().await.clone();
+
+    let org_config = shanty_org::OrgConfig {
+        target_dir: cfg.library_path.clone(),
+        format: cfg.organization_format.clone(),
+        dry_run: false,
+        copy: false,
+    };
+
+    shanty_org::organize_track(conn, track_id, &org_config)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // Promote this track's wanted item from Downloaded to Owned
+    let _ = queries::wanted::promote_downloaded_to_owned(conn).await;
+
+    // Check if pipeline is complete — run cleanup then create enrich work items
+    // Exclude this item from the check since it's still "running" in the DB
+    let mut downstream = Vec::new();
+    if let Some(ref pipeline_id) = item.pipeline_id
+        && let Ok(true) =
+            queries::work_queue::pipeline_is_complete(conn, pipeline_id, Some(item.id)).await
+    {
+        tracing::info!(pipeline_id = %pipeline_id, "pipeline complete, running cleanup");
+
+        // Cleanup: remove orphaned tracks, empty albums, unused artists
+        match queries::tracks::delete_orphaned(conn).await {
+            Ok(n) if n > 0 => tracing::info!(count = n, "cleaned up orphaned tracks"),
+            Err(e) => tracing::warn!(error = %e, "failed to clean orphaned tracks"),
+            _ => {}
+        }
+        match queries::albums::delete_empty(conn).await {
+            Ok(n) if n > 0 => tracing::info!(count = n, "cleaned up empty albums"),
+            Err(e) => tracing::warn!(error = %e, "failed to clean empty albums"),
+            _ => {}
+        }
+        match queries::artists::delete_unused(conn).await {
+            Ok(n) if n > 0 => tracing::info!(count = n, "cleaned up unused artists"),
+            Err(e) => tracing::warn!(error = %e, "failed to clean unused artists"),
+            _ => {}
+        }
+
+        // Create Enrich work items for each artist that has wanted items
+        let all_wanted = queries::wanted::list(conn, None, None)
+            .await
+            .unwrap_or_default();
+        let mut artist_ids: Vec<i32> = all_wanted.iter().filter_map(|w| w.artist_id).collect();
+        artist_ids.sort();
+        artist_ids.dedup();
+
+        for artist_id in &artist_ids {
+            let payload = serde_json::json!({"artist_id": artist_id});
+            downstream.push((WorkTaskType::Enrich, payload.to_string()));
+        }
+    }
+
+    Ok(downstream)
+}
+
+async fn process_enrich(
+    state: &web::Data<AppState>,
+    item: &shanty_db::entities::work_queue::Model,
+) -> WorkResult {
+    let payload: serde_json::Value =
+        serde_json::from_str(&item.payload_json).map_err(|e| e.to_string())?;
+    let artist_id = payload
+        .get("artist_id")
+        .and_then(|v| v.as_i64())
+        .ok_or("missing artist_id in payload")? as i32;
+
+    crate::routes::artists::enrich_artist(state, &artist_id.to_string(), false)
+        .await
+        .map_err(|e| e.to_string())?;
+
+    // If this pipeline is fully done, clear its completed items
+    if let Some(ref pipeline_id) = item.pipeline_id
+        && let Ok(true) =
+            queries::work_queue::pipeline_is_complete(state.db.conn(), pipeline_id, Some(item.id))
+                .await
+    {
+        let _ = queries::work_queue::clear_pipeline(state.db.conn(), pipeline_id).await;
+    }
+
+    Ok(vec![])
+}