feat(library): focused music/TV/movie landing screens + self-draining download queue

Library screens:
- Add dedicated music, TV, and movie landing pages (hero banner +
  horizontal carousels) backed by new music/tv/movies stores.
- Route tvshows libraries to /library/tv; surface rediscover ("haven't
  listened to in a while") albums via a new repository method across
  online/offline/hybrid repos plus the repository_get_rediscover_albums
  command.
- Add an A-Z jump bar for long alphabetically-sorted lists, with grid
  index anchors in LibraryGrid/LibraryListView/TrackList.
- Filter the "Podcasts" folder out of music library queries.

Downloads:
- Add a backend queue pump: enqueue_download / enqueue_video_downloads
  persist the resolved stream URL + target dir on each row (migration
  017), and the pump starts up to max_concurrent and drains the rest
  automatically as slots free, instead of the frontend silently dropping
  items past the concurrency limit. Album/series/season buttons now
  enqueue rather than calling start_download directly.

Other fixes:
- Hybrid search now returns instant cache results and pushes the merged
  cache+server union via a request-id-tagged search-event, so superseded
  queries can't clobber fresher results.
- URL-encode SearchTerm / genres / item types in online repo requests.
- Android: pause on audio-becoming-noisy (headphone/BT disconnect).

src-tauri/src/repository/hybrid.rs

@@ -57,6 +57,81 @@ impl HybridRepository {
         self.online.get_video_stream_url(item_id, media_source_id, start_time_seconds, audio_stream_index).await
     }
 
+    /// Search only the local SQLite cache (downloaded content).
+    ///
+    /// Fast (100ms timeout) — used to render instant results before the server
+    /// responds. Returns an empty result rather than erroring on timeout so the
+    /// caller can still fall through to the server.
+    pub async fn search_cache_only(
+        &self,
+        query: &str,
+        options: Option<SearchOptions>,
+    ) -> Result<SearchResult, RepoError> {
+        let offline = Arc::clone(&self.offline);
+        let query = query.to_string();
+        self.cache_with_timeout(async move { offline.search(&query, options).await })
+            .await
+    }
+
+    /// Search only the live Jellyfin server (full library).
+    pub async fn search_server_only(
+        &self,
+        query: &str,
+        options: Option<SearchOptions>,
+    ) -> Result<SearchResult, RepoError> {
+        self.online.search(query, options).await
+    }
+
+    /// Merge cache and server search results into a single de-duplicated list.
+    ///
+    /// Ordering: local (cached/downloaded) items first, then server-only items
+    /// appended. On a duplicate `id`, the server's item wins (fresher, more
+    /// complete metadata) but keeps the local item's earlier position.
+    pub fn merge_search_results(cache: SearchResult, server: SearchResult) -> SearchResult {
+        use std::collections::HashMap;
+
+        // Index server items by id so we can (a) override duplicates with the
+        // server's metadata and (b) know which server items are brand new.
+        let mut server_by_id: HashMap<String, MediaItem> = HashMap::new();
+        let mut server_order: Vec<String> = Vec::with_capacity(server.items.len());
+        for item in server.items {
+            if !server_by_id.contains_key(&item.id) {
+                server_order.push(item.id.clone());
+            }
+            server_by_id.insert(item.id.clone(), item);
+        }
+
+        let mut items: Vec<MediaItem> = Vec::new();
+        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
+
+        // Local items first, in their original order. If the server also
+        // returned this item, take the server's copy (newer metadata).
+        for local in cache.items {
+            if !seen.insert(local.id.clone()) {
+                continue;
+            }
+            match server_by_id.remove(&local.id) {
+                Some(server_item) => items.push(server_item),
+                None => items.push(local),
+            }
+        }
+
+        // Then append server-only items, preserving the server's order.
+        for id in server_order {
+            if let Some(server_item) = server_by_id.remove(&id) {
+                if seen.insert(id) {
+                    items.push(server_item);
+                }
+            }
+        }
+
+        let total_record_count = items.len();
+        SearchResult {
+            items,
+            total_record_count,
+        }
+    }
+
     /// Cache-first query: try cache, fall back to server on miss.
     ///
     /// 1. Check cache (100ms timeout applied by caller via cache_with_timeout)
@@ -274,6 +349,27 @@ impl MediaRepository for HybridRepository {
         self.parallel_race(cache_future, server_future).await
     }
 
+    async fn get_rediscover_albums(
+        &self,
+        parent_id: Option<&str>,
+        limit: Option<usize>,
+    ) -> Result<Vec<MediaItem>, RepoError> {
+        let offline = Arc::clone(&self.offline);
+        let online = Arc::clone(&self.online);
+        let parent_id_owned = parent_id.map(|s| s.to_string());
+        let parent_id_clone = parent_id_owned.clone();
+
+        let cache_future = self.cache_with_timeout(async move {
+            offline.get_rediscover_albums(parent_id_owned.as_deref(), limit).await
+        });
+
+        let server_future = async move {
+            online.get_rediscover_albums(parent_id_clone.as_deref(), limit).await
+        };
+
+        self.parallel_race(cache_future, server_future).await
+    }
+
     async fn get_genres(&self, parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
         let offline = Arc::clone(&self.offline);
         let online = Arc::clone(&self.online);
@@ -602,6 +698,14 @@ mod tests {
             unimplemented!()
         }
 
+        async fn get_rediscover_albums(
+            &self,
+            _parent_id: Option<&str>,
+            _limit: Option<usize>,
+        ) -> Result<Vec<MediaItem>, RepoError> {
+            unimplemented!()
+        }
+
         async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
             unimplemented!()
         }
@@ -759,6 +863,14 @@ mod tests {
             unimplemented!()
         }
 
+        async fn get_rediscover_albums(
+            &self,
+            _parent_id: Option<&str>,
+            _limit: Option<usize>,
+        ) -> Result<Vec<MediaItem>, RepoError> {
+            unimplemented!()
+        }
+
         async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
             unimplemented!()
         }
@@ -1058,4 +1170,75 @@ mod tests {
         };
         assert!(result_with_items.has_content(), "Result with items should have content");
     }
+
+    #[test]
+    fn test_merge_search_local_first_then_server_appended() {
+        let cache = SearchResult {
+            items: vec![
+                create_test_item("a", "Cached A"),
+                create_test_item("b", "Cached B"),
+            ],
+            total_record_count: 2,
+        };
+        let server = SearchResult {
+            items: vec![
+                create_test_item("c", "Server C"),
+                create_test_item("d", "Server D"),
+            ],
+            total_record_count: 2,
+        };
+
+        let merged = HybridRepository::merge_search_results(cache, server);
+
+        // Local items first (in order), then server-only items appended.
+        let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect();
+        assert_eq!(ids, vec!["a", "b", "c", "d"]);
+        assert_eq!(merged.total_record_count, 4);
+    }
+
+    #[test]
+    fn test_merge_search_dedupes_with_server_winning() {
+        // "b" appears in both. Server metadata should win, but the item keeps
+        // its earlier (local) position and is not duplicated.
+        let cache = SearchResult {
+            items: vec![
+                create_test_item("a", "Cached A"),
+                create_test_item("b", "Cached B"),
+            ],
+            total_record_count: 2,
+        };
+        let server = SearchResult {
+            items: vec![
+                create_test_item("b", "Server B (fresher)"),
+                create_test_item("c", "Server C"),
+            ],
+            total_record_count: 2,
+        };
+
+        let merged = HybridRepository::merge_search_results(cache, server);
+
+        let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect();
+        assert_eq!(ids, vec!["a", "b", "c"], "no duplicate, local position kept");
+
+        let b = merged.items.iter().find(|i| i.id == "b").unwrap();
+        assert_eq!(b.name, "Server B (fresher)", "server metadata wins on conflict");
+        assert_eq!(merged.total_record_count, 3);
+    }
+
+    #[test]
+    fn test_merge_search_handles_empty_sides() {
+        let only_server = HybridRepository::merge_search_results(
+            SearchResult { items: vec![], total_record_count: 0 },
+            SearchResult { items: vec![create_test_item("x", "X")], total_record_count: 1 },
+        );
+        assert_eq!(only_server.items.len(), 1);
+        assert_eq!(only_server.items[0].id, "x");
+
+        let only_cache = HybridRepository::merge_search_results(
+            SearchResult { items: vec![create_test_item("y", "Y")], total_record_count: 1 },
+            SearchResult { items: vec![], total_record_count: 0 },
+        );
+        assert_eq!(only_cache.items.len(), 1);
+        assert_eq!(only_cache.items[0].id, "y");
+    }
 }