dtourolle/jellytau
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 40 Wiki Activity
Files
2ff07bfa49addbe5d99753e1f12fa0c64dd6b1f2
jellytau/src-tauri/src/commands/download/mod.rs
T
dtourolle 048c99ebcc fix(downloads): allow the deliberate join_absolute_paths lint in a test 
The assertion documents that PathBuf::join discards its base when handed an
absolute path — which is why confinement has to happen after the join, not
instead of it. clippy::join_absolute_paths flags that shape, correctly for
production code, so the lint is allowed here rather than the test weakened.

Worth recording: this lint would not have caught the original defect. The real
join sites pass a variable, and it only fires on a literal.
2026-08-20 20:09:19 +02:00

3459 lines
123 KiB
Rust
Raw Blame History

//! Tauri commands for download operations
#[cfg(test)]
use crate::utils::lock::MutexSafe;
use log::{debug, error, info, warn};
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
use tauri::{Manager, State};
use super::{DatabaseWrapper, SmartCacheWrapper};
use crate::download::network::{NetworkState, NetworkStateHandle, NetworkType};
use crate::download::{DownloadInfo, DownloadManager};
use crate::storage::db_service::{DatabaseService, Query, QueryParam};
// Cohesive command clusters in their own submodules, re-exported so the command
// names remain at `commands::download::*` (invoke_handler unchanged).
mod pinning;
mod smart_cache;
pub use pinning::*;
pub use smart_cache::*;
/// One row of the series episode listing used when queueing a whole series:
/// `(id, name, season_name, index_number, parent_index_number)`.
type EpisodeRow = (String, String, Option<String>, Option<i32>, Option<i32>);
/// Wrapper for DownloadManager to be used as Tauri state
pub struct DownloadManagerWrapper(pub Mutex<DownloadManager>);
/// Wrapper for the current network transport, used by the WiFi-only gate.
///
/// TRACES: UR-053 | DR-074
pub struct NetworkStateWrapper(pub NetworkStateHandle);
/// Report the device's current network transport (Android → Rust).
///
/// The frontend calls this on startup and whenever the native network callback
/// fires. Updating to an acceptable network re-pumps the download queue, so a
/// queue parked on "waiting for WiFi" drains itself without user action.
///
/// TRACES: UR-053 | DR-074
#[tauri::command]
#[specta::specta]
pub async fn set_network_state(
app: tauri::AppHandle,
network: NetworkStateWrapperArg,
db: State<'_, DatabaseWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
) -> Result<(), String> {
let new_state = NetworkState {
network_type: network.network_type,
unmetered: network.unmetered,
};
let handle = app.state::<NetworkStateWrapper>().0.clone();
let previous = handle.get().await;
handle.set(new_state).await;
if previous != new_state {
info!(
"[network] Transport changed: {:?} (unmetered={}) -> {:?} (unmetered={})",
previous.network_type, previous.unmetered, new_state.network_type, new_state.unmetered
);
}
// If the new network unblocks the gate, drain whatever was waiting.
if downloads_allowed_on_current_network(&app).await {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let active = {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.get_active_downloads()
};
pump_download_queue(app.clone(), db_service, active).await;
}
Ok(())
}
/// Argument struct for [`set_network_state`].
///
/// TRACES: UR-053 | DR-074
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NetworkStateWrapperArg {
pub network_type: NetworkType,
pub unmetered: bool,
}
/// Whether downloads are currently permitted by the WiFi-only gate.
///
/// The downloads UI uses this to render "Waiting for WiFi" on pending rows
/// rather than leaving them looking silently stuck.
///
/// TRACES: UR-053 | DR-074
#[tauri::command]
#[specta::specta]
pub async fn get_downloads_allowed(app: tauri::AppHandle) -> Result<bool, String> {
Ok(downloads_allowed_on_current_network(&app).await)
}
/// Download statistics computed server-side
#[allow(dead_code)]
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DownloadStats {
pub total: usize,
pub active_count: usize,
pub queued_count: usize,
pub completed_count: usize,
pub failed_count: usize,
pub paused_count: usize,
}
/// Enhanced response with pre-computed stats
#[allow(dead_code)]
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DownloadsResponse {
pub downloads: Vec<DownloadInfo>,
pub stats: DownloadStats,
}
/// Sanitize filename by removing invalid characters
fn sanitize_filename(name: &str) -> String {
name.chars()
.map(|c| match c {
'/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
_ => c,
})
.collect()
}
/// The directory every download has to stay inside: the storage root
/// `storage_get_path` hands the frontend, which is the database's parent.
///
/// TRACES: DR-211 | UT-205
fn download_root(db: &DatabaseWrapper) -> Result<PathBuf, String> {
let database = db.0.lock().map_err(|e| e.to_string())?;
database
.path()
.parent()
.map(|p| p.to_path_buf())
.ok_or_else(|| "Database path has no parent directory".to_string())
}
/// Fold `..` out of `candidate` and require what is left to sit inside `root`.
///
/// Lexical rather than `canonicalize`, the same way `media_server::resolve_path`
/// does it: the file usually does not exist yet, so canonicalising would fail on
/// the ordinary case. The check has to come *after* the caller's join, because
/// `Path::join` drops the base when the joined half is absolute — such a path is
/// not folded, it is obeyed, and only the `starts_with` below catches it.
///
/// TRACES: DR-211 | UT-205
fn confine_to_root(root: &Path, candidate: &Path) -> Result<PathBuf, String> {
let mut resolved = PathBuf::new();
for component in candidate.components() {
match component {
Component::ParentDir => {
resolved.pop();
}
Component::CurDir => {}
other => resolved.push(other),
}
}
if resolved.starts_with(root) {
Ok(resolved)
} else {
Err(format!(
"Refusing a download path outside the download directory: {}",
candidate.display()
))
}
}
/// Sanitize a queued download's path and confine it to the download directory.
///
/// Every path the app builds for itself comes back unchanged — files on disk and
/// `downloads` rows point at these exact spellings — and [`sanitize_filename`]
/// is idempotent, so the already-safe name `download_item_and_start` passes in
/// is not sanitized into a second, different one.
///
/// TRACES: DR-211 | UT-205
fn confine_queued_path(root: &Path, file_path: &str) -> Result<String, String> {
let mut sanitized = PathBuf::new();
for component in Path::new(file_path).components() {
match component {
Component::Normal(part) => sanitized.push(sanitize_filename(&part.to_string_lossy())),
// Kept as they are, so `confine_to_root` is the single thing
// deciding whether what they add up to is still inside the root.
other => sanitized.push(other),
}
}
confine_to_root(root, &root.join(&sanitized))?;
Ok(sanitized.to_string_lossy().to_string())
}
/// Request payload for download_item_and_start (bundled to stay within specta's
/// 10-argument command limit).
#[derive(Debug, specta::Type, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DownloadItemAndStartRequest {
pub item_id: String,
pub user_id: String,
pub stream_url: String,
pub target_dir: String,
pub item_name: Option<String>,
pub artist_name: Option<String>,
pub album_name: Option<String>,
}
/// Request payload for download_item.
#[derive(Debug, specta::Type, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DownloadItemRequest {
pub item_id: String,
pub user_id: String,
pub file_path: String,
pub mime_type: Option<String>,
pub priority: Option<i32>,
pub item_name: Option<String>,
pub artist_name: Option<String>,
pub album_name: Option<String>,
pub expected_size: Option<i64>,
}
/// Request payload for download_video.
#[derive(Debug, specta::Type, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DownloadVideoRequest {
pub item_id: String,
pub user_id: String,
pub file_path: String,
pub mime_type: Option<String>,
pub priority: Option<i32>,
pub item_name: Option<String>,
pub quality_preset: Option<String>,
pub series_name: Option<String>,
pub season_name: Option<String>,
pub episode_number: Option<i32>,
pub season_number: Option<i32>,
}
/// Queue and start a download in a single atomic operation
/// This simplifies the frontend flow by combining multiple steps
#[tauri::command]
#[specta::specta]
pub async fn download_item_and_start(
db: State<'_, DatabaseWrapper>,
smart_cache: State<'_, SmartCacheWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
app: tauri::AppHandle,
request: DownloadItemAndStartRequest,
) -> Result<i64, String> {
let DownloadItemAndStartRequest {
item_id,
user_id,
stream_url,
target_dir,
item_name,
artist_name,
album_name,
} = request;
// Sanitize filename
let safe_name = sanitize_filename(item_name.as_deref().unwrap_or(&item_id));
let file_path = format!("downloads/{}.mp3", safe_name);
// Queue the download
let download_id = download_item(
db.clone(),
smart_cache.clone(),
DownloadItemRequest {
item_id,
user_id,
file_path,
mime_type: None,
priority: None,
item_name,
artist_name,
album_name,
expected_size: None,
},
)
.await?;
// Start the download immediately
start_download(
db,
download_manager,
app,
download_id,
stream_url,
target_dir,
)
.await?;
Ok(download_id)
}
/// Queue a media item for download
#[tauri::command]
#[specta::specta]
pub async fn download_item(
db: State<'_, DatabaseWrapper>,
smart_cache: State<'_, SmartCacheWrapper>,
request: DownloadItemRequest,
) -> Result<i64, String> {
let DownloadItemRequest {
item_id,
user_id,
file_path,
mime_type,
priority,
item_name,
artist_name,
album_name,
expected_size,
} = request;
// `start_download` joins this onto the target directory, and `Path::join`
// drops the base when the second half is absolute, so the row itself has to
// be confined — not only the place it is used. `download_item_and_start`
// sanitizes the name it builds, but `download_item` is a command in its own
// right, so that guard was simply routed around by calling this directly.
// TRACES: DR-211 | UT-205
let file_path = {
let root = download_root(&db)?;
confine_queued_path(&root, &file_path)?
};
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Check storage limit if size is known
if let Some(size) = expected_size {
// Clone Arc to avoid holding lock during async operations
let cache_arc = {
let guard = smart_cache.0.lock().map_err(|e| e.to_string())?;
guard.clone()
};
// Check if we have space
let can_download = cache_arc
.can_download_async(&db_service, &user_id, size as u64)
.await;
if !can_download {
warn!("Storage limit reached. Attempting to free space...");
// Reclaim expired temporary entries first: they are dead weight, so
// freeing them may avoid evicting cache that is still within its
// life. Best-effort — a failure here just means eviction does more.
// TRACES: UR-071 | DR-127
match cache_arc
.reclaim_expired_async(&db_service, &user_id, &chrono::Utc::now().to_rfc3339())
.await
{
Ok(n) if n > 0 => info!("Reclaimed {} expired cache entries", n),
Ok(_) => {}
Err(e) => warn!("Expired-entry reclaim failed: {}", e),
}
// Try to evict LRU items to make space
match cache_arc
.evict_lru_async(&db_service, &user_id, size as u64)
.await
{
Ok(freed) if freed > 0 => {
info!("Freed {} bytes, proceeding with download", freed);
}
Ok(_) => {
let storage_limit =
cache_arc.get_config().map(|c| c.storage_limit).unwrap_or(0);
return Err(format!(
"Storage limit reached ({} bytes). Unable to free enough space.",
storage_limit
));
}
Err(e) => {
error!("Failed to evict items: {}", e);
return Err(format!("Storage limit reached. Eviction failed: {}", e));
}
}
}
}
// Insert or update download record with metadata
let insert_query = Query::with_params(
"INSERT INTO downloads (item_id, user_id, file_path, mime_type, status, priority, queued_at, item_name, artist_name, album_name)
VALUES (?, ?, ?, ?, 'pending', ?, CURRENT_TIMESTAMP, ?, ?, ?)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = excluded.priority,
status = 'pending',
queued_at = CURRENT_TIMESTAMP,
item_name = COALESCE(excluded.item_name, downloads.item_name),
artist_name = COALESCE(excluded.artist_name, downloads.artist_name),
album_name = COALESCE(excluded.album_name, downloads.album_name)",
vec![
QueryParam::String(item_id.clone()),
QueryParam::String(user_id.clone()),
QueryParam::String(file_path),
mime_type.map(QueryParam::String).unwrap_or(QueryParam::Null),
QueryParam::Int(priority.unwrap_or(0)),
item_name.map(QueryParam::String).unwrap_or(QueryParam::Null),
artist_name.map(QueryParam::String).unwrap_or(QueryParam::Null),
album_name.map(QueryParam::String).unwrap_or(QueryParam::Null),
],
);
db_service
.execute(insert_query)
.await
.map_err(|e| e.to_string())?;
// Query for the download ID by unique constraint columns
// NOTE: last_insert_rowid() doesn't work reliably with UPSERT - it only updates on INSERT, not UPDATE
let id_query = Query::with_params(
"SELECT id FROM downloads WHERE item_id = ? AND user_id = ?",
vec![QueryParam::String(item_id), QueryParam::String(user_id)],
);
let download_id: i64 = db_service
.query_one(id_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
Ok(download_id)
}
/// One track of an album, as the album-download path queues it.
///
/// `artist_name` carries whatever the catalog holds for the track's artists (a
/// JSON array, as stored on `items.artists`); it is display metadata for the
/// downloads list, not a lookup key.
///
/// TRACES: UR-018, UR-055 | DR-173
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct AlbumTrack {
pub id: String,
pub name: String,
pub artist_name: Option<String>,
pub album_name: Option<String>,
pub index_number: Option<i32>,
}
impl From<&crate::repository::types::MediaItem> for AlbumTrack {
fn from(item: &crate::repository::types::MediaItem) -> Self {
Self {
id: item.id.clone(),
name: item.name.clone(),
artist_name: item
.artists
.as_ref()
.and_then(|a| serde_json::to_string(a).ok()),
album_name: item.album_name.clone(),
index_number: item.index_number,
}
}
}
/// The album's tracks as the local catalog cache knows them.
///
/// Only a fallback for [`download_album`]: the cache links a track to its album
/// through `items.album_id`, which Jellyfin does not populate on every listing
/// endpoint, so this can legitimately return fewer tracks than the album has.
///
/// TRACES: UR-018, UR-055 | DR-173
pub(crate) async fn cached_album_tracks(
db_service: &Arc<crate::storage::db_service::RusqliteService>,
album_id: &str,
) -> Result<Vec<AlbumTrack>, String> {
let tracks_query = Query::with_params(
"SELECT id, name, artists, album_name, index_number FROM items
WHERE (album_id = ? OR parent_id = ?) AND item_type = 'Audio'
ORDER BY index_number",
vec![
QueryParam::String(album_id.to_string()),
QueryParam::String(album_id.to_string()),
],
);
db_service
.query_many(tracks_query, |row| {
Ok(AlbumTrack {
id: row.get(0)?,
name: row.get(1)?,
artist_name: row.get(2)?,
album_name: row.get(3)?,
index_number: row.get(4)?,
})
})
.await
.map_err(|e| e.to_string())
}
/// Queue one download row per track and link every track to its album.
///
/// The linkage is the half that is easy to miss: offline browsing joins a track
/// to its album on `items.album_id` (see `OfflineRepository::get_items`), so a
/// track whose cached row lacks it stays invisible under the album even after
/// its file is on disk. Queuing a track *is* the statement that it belongs to
/// this album, so the link is written here rather than hoped for from whichever
/// listing endpoint happened to cache the row.
///
/// Idempotent: re-queuing an album fills in what is missing and returns the same
/// row ids, in the order the tracks were given.
///
/// A file name per track, unique within the album.
///
/// A title is not a unique name inside its own album: a deluxe edition carries
/// the album version and a demo of the same song, and a two-disc set repeats
/// titles across discs. Naming files after the title alone gave those tracks one
/// path, and each download overwrote the previous one — an album that quietly
/// ends up short by however many titles it repeats. The track number
/// disambiguates the ordinary case; anything still colliding falls back to the
/// item id, which is unique by construction.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-172
pub(crate) fn album_file_names(tracks: &[AlbumTrack]) -> Vec<String> {
let mut counts: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
for track in tracks {
*counts.entry(track.name.to_lowercase()).or_default() += 1;
}
tracks
.iter()
.map(|track| {
let title = sanitize_filename(&track.name);
if counts.get(&track.name.to_lowercase()).copied().unwrap_or(0) <= 1 {
return format!("{}.mp3", title);
}
match track.index_number {
Some(n) => format!("{:02} - {} [{}].mp3", n, title, track.id),
None => format!("{} [{}].mp3", title, track.id),
}
})
.collect()
}
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
pub(crate) async fn queue_album_tracks(
db_service: &Arc<crate::storage::db_service::RusqliteService>,
album_id: &str,
tracks: &[AlbumTrack],
user_id: &str,
base_path: &str,
) -> Result<Vec<i64>, String> {
let mut download_ids = Vec::with_capacity(tracks.len());
let file_names = album_file_names(tracks);
for (track, file_name) in tracks.iter().zip(file_names) {
// Cache a row for a track the catalog has never seen, borrowing the
// album's server. Nothing is inserted when the album itself is unknown,
// which also keeps the parent_id foreign key satisfiable.
let cache_query = Query::with_params(
"INSERT OR IGNORE INTO items
(id, server_id, parent_id, name, item_type, album_id, album_name, artists, index_number)
SELECT ?, a.server_id, a.id, ?, 'Audio', a.id, ?, ?, ?
FROM items a WHERE a.id = ?",
vec![
QueryParam::String(track.id.clone()),
QueryParam::String(track.name.clone()),
track
.album_name
.clone()
.map(QueryParam::String)
.unwrap_or(QueryParam::Null),
track
.artist_name
.clone()
.map(QueryParam::String)
.unwrap_or(QueryParam::Null),
track
.index_number
.map(QueryParam::Int)
.unwrap_or(QueryParam::Null),
QueryParam::String(album_id.to_string()),
],
);
db_service
.execute(cache_query)
.await
.map_err(|e| e.to_string())?;
// Link an already-cached track to the album. The parent_id subquery
// resolves to NULL when the album is not cached, so the foreign key
// holds either way.
let link_query = Query::with_params(
"UPDATE items
SET album_id = ?,
parent_id = COALESCE(parent_id, (SELECT id FROM items WHERE id = ?))
WHERE id = ?",
vec![
QueryParam::String(album_id.to_string()),
QueryParam::String(album_id.to_string()),
QueryParam::String(track.id.clone()),
],
);
db_service
.execute(link_query)
.await
.map_err(|e| e.to_string())?;
let file_path = format!("{}/{}", base_path, file_name);
// Queue at album priority (100). A track already downloaded stays
// completed — re-queuing an album must fill the gaps, not re-fetch it.
let insert_query = Query::with_params(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at, item_name, artist_name, album_name, media_type)
VALUES (?, ?, ?, 'pending', 100, CURRENT_TIMESTAMP, ?, ?, ?, 'audio')
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = 100,
status = CASE WHEN downloads.status = 'completed' THEN 'completed' ELSE 'pending' END,
media_type = 'audio',
item_name = COALESCE(excluded.item_name, downloads.item_name),
artist_name = COALESCE(excluded.artist_name, downloads.artist_name),
album_name = COALESCE(excluded.album_name, downloads.album_name)",
vec![
QueryParam::String(track.id.clone()),
QueryParam::String(user_id.to_string()),
QueryParam::String(file_path),
QueryParam::String(track.name.clone()),
track
.artist_name
.clone()
.map(QueryParam::String)
.unwrap_or(QueryParam::Null),
track
.album_name
.clone()
.map(QueryParam::String)
.unwrap_or(QueryParam::Null),
],
);
db_service
.execute(insert_query)
.await
.map_err(|e| e.to_string())?;
// Query for the actual download ID (last_insert_rowid doesn't work with UPSERT)
let id_query = Query::with_params(
"SELECT id FROM downloads WHERE item_id = ? AND user_id = ?",
vec![
QueryParam::String(track.id.clone()),
QueryParam::String(user_id.to_string()),
],
);
let download_id: i64 = db_service
.query_one(id_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
download_ids.push(download_id);
}
Ok(download_ids)
}
/// Queue an entire album for download.
///
/// Owns the whole operation: the album's track list comes from the server (the
/// only place that knows all of it), every track is queued and linked to its
/// album, each row's stream URL is resolved here, and the queue is pumped.
///
/// The frontend used to do the second half — resolve one URL per track and pair
/// it with the returned ids **by position**. That pairing had no basis: the ids
/// came back in the backend's own order over a different set of rows, so
/// whenever the two lists disagreed a row was handed another track's URL, and
/// any track past the end of the shorter list was never started at all. Nothing
/// crosses the boundary now except the album id.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
#[tauri::command]
#[specta::specta]
// Three of the eight arguments are Tauri `State<'_, _>` injections plus the
// `AppHandle`, not caller input. Folding the rest into a struct would change the
// IPC contract and the generated TypeScript for no readability gain.
#[allow(clippy::too_many_arguments)]
pub async fn download_album(
db: State<'_, DatabaseWrapper>,
repository: State<'_, crate::commands::repository::RepositoryManagerWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
app: tauri::AppHandle,
handle: String,
album_id: String,
user_id: String,
base_path: String,
) -> Result<Vec<i64>, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let repo = repository.0.get(&handle);
// Ask the server what the album contains; the cache is only a fallback for
// when it cannot answer.
let tracks: Vec<AlbumTrack> = match &repo {
Some(repo) => match repo.get_album_tracks(&album_id).await {
Ok(items) if !items.is_empty() => items.iter().map(AlbumTrack::from).collect(),
Ok(_) => cached_album_tracks(&db_service, &album_id).await?,
Err(e) => {
warn!(
"[download_album] Could not list album {} from the repository ({:?}); \
falling back to the cached track list",
album_id, e
);
cached_album_tracks(&db_service, &album_id).await?
}
},
None => cached_album_tracks(&db_service, &album_id).await?,
};
if tracks.is_empty() {
warn!("[download_album] No tracks found for album {}", album_id);
return Ok(Vec::new());
}
let download_ids =
queue_album_tracks(&db_service, &album_id, &tracks, &user_id, &base_path).await?;
info!(
"[download_album] Queued {} track(s) for album {}",
download_ids.len(),
album_id
);
// Resolve each queued row's stream URL here, then pump. Without a
// repository (or while offline) the rows stay pending with no URL and
// `resume_queued_downloads` picks them up on reconnect.
let Some(repo) = repo else {
return Ok(download_ids);
};
let target_dir = {
let database = db.0.lock().map_err(|e| e.to_string())?;
database
.path()
.parent()
.ok_or_else(|| "Database path has no parent directory".to_string())?
.to_string_lossy()
.to_string()
};
let repo_for_resolve = Arc::clone(&repo);
let outcome = crate::commands::catalog::resolve_pending_download_urls(
&db_service,
&target_dir,
Some(&download_ids),
move |item_id: String, _media_type: String, _quality: String| {
let repo = Arc::clone(&repo_for_resolve);
async move {
use crate::repository::MediaRepository;
match repo.get_audio_stream_url(&item_id).await {
Ok(url) => Some(url),
Err(e) => {
warn!(
"[download_album] Failed to resolve stream URL for {}: {:?}",
item_id, e
);
None
}
}
}
},
)
.await?;
if outcome.failed > 0 {
warn!(
"[download_album] {} track(s) could not be resolved and stay queued for the next \
reconnect",
outcome.failed
);
}
let active_downloads = {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.get_active_downloads()
};
pump_download_queue(app, db_service, active_downloads).await;
Ok(download_ids)
}
/// Queue a video item (movie or episode) for download with quality preset
#[tauri::command]
#[specta::specta]
pub async fn download_video(
db: State<'_, DatabaseWrapper>,
request: DownloadVideoRequest,
) -> Result<i64, String> {
let DownloadVideoRequest {
item_id,
user_id,
file_path,
mime_type,
priority,
item_name,
quality_preset,
series_name,
season_name,
episode_number,
season_number,
} = request;
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let quality = quality_preset.unwrap_or_else(|| "original".to_string());
// Insert or update download record with video metadata
let insert_query = Query::with_params(
"INSERT INTO downloads (item_id, user_id, file_path, mime_type, status, priority, queued_at,
item_name, quality_preset, media_type, series_name, season_name,
episode_number, season_number)
VALUES (?, ?, ?, ?, 'pending', ?, CURRENT_TIMESTAMP, ?, ?, 'video', ?, ?, ?, ?)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = excluded.priority,
status = 'pending',
queued_at = CURRENT_TIMESTAMP,
quality_preset = excluded.quality_preset,
item_name = COALESCE(excluded.item_name, downloads.item_name),
series_name = COALESCE(excluded.series_name, downloads.series_name),
season_name = COALESCE(excluded.season_name, downloads.season_name),
episode_number = COALESCE(excluded.episode_number, downloads.episode_number),
season_number = COALESCE(excluded.season_number, downloads.season_number)",
vec![
QueryParam::String(item_id.clone()),
QueryParam::String(user_id.clone()),
QueryParam::String(file_path),
mime_type.map(QueryParam::String).unwrap_or(QueryParam::Null),
QueryParam::Int(priority.unwrap_or(0)),
item_name.map(QueryParam::String).unwrap_or(QueryParam::Null),
QueryParam::String(quality),
series_name.map(QueryParam::String).unwrap_or(QueryParam::Null),
season_name.map(QueryParam::String).unwrap_or(QueryParam::Null),
episode_number.map(QueryParam::Int).unwrap_or(QueryParam::Null),
season_number.map(QueryParam::Int).unwrap_or(QueryParam::Null),
],
);
db_service
.execute(insert_query)
.await
.map_err(|e| e.to_string())?;
// Query for the download ID by unique constraint columns
let id_query = Query::with_params(
"SELECT id FROM downloads WHERE item_id = ? AND user_id = ?",
vec![QueryParam::String(item_id), QueryParam::String(user_id)],
);
let download_id: i64 = db_service
.query_one(id_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
Ok(download_id)
}
/// Queue all episodes of a series for download
#[tauri::command]
#[specta::specta]
pub async fn download_series(
db: State<'_, DatabaseWrapper>,
series_id: String,
series_name: String,
user_id: String,
base_path: String,
quality_preset: Option<String>,
) -> Result<Vec<i64>, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let quality = quality_preset.unwrap_or_else(|| "original".to_string());
// Get all episodes for this series, ordered by season and episode number
let episodes_query = Query::with_params(
"SELECT id, name, season_name, index_number, parent_index_number
FROM items
WHERE series_id = ? AND item_type = 'Episode'
ORDER BY parent_index_number, index_number",
vec![QueryParam::String(series_id)],
);
let episodes: Vec<EpisodeRow> = db_service
.query_many(episodes_query, |row| {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
))
})
.await
.map_err(|e| e.to_string())?;
let mut download_ids = Vec::new();
// Queue each episode with descending priority (first episodes download first)
// Priority starts high and decreases so earlier episodes finish first
let total_episodes = episodes.len() as i32;
for (idx, (episode_id, episode_name, season_name, episode_number, season_number)) in
episodes.into_iter().enumerate()
{
let priority = 1000 - idx as i32; // High priority for first episodes
// Create path like: videos/SeriesName/S01E01_Title.mp4
let season_num = season_number.unwrap_or(1);
let episode_num = episode_number.unwrap_or(1);
let file_name = format!(
"S{:02}E{:02}_{}.mp4",
season_num,
episode_num,
sanitize_filename(&episode_name)
);
let file_path = format!(
"{}/{}/{}",
base_path,
sanitize_filename(&series_name),
file_name
);
let insert_query = Query::with_params(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at,
item_name, quality_preset, media_type, series_name, season_name,
episode_number, season_number)
VALUES (?, ?, ?, 'pending', ?, CURRENT_TIMESTAMP, ?, ?, 'video', ?, ?, ?, ?)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = excluded.priority,
status = 'pending',
queued_at = CURRENT_TIMESTAMP,
quality_preset = excluded.quality_preset,
item_name = COALESCE(excluded.item_name, downloads.item_name),
series_name = COALESCE(excluded.series_name, downloads.series_name),
season_name = COALESCE(excluded.season_name, downloads.season_name),
episode_number = COALESCE(excluded.episode_number, downloads.episode_number),
season_number = COALESCE(excluded.season_number, downloads.season_number)",
vec![
QueryParam::String(episode_id.clone()),
QueryParam::String(user_id.clone()),
QueryParam::String(file_path),
QueryParam::Int(priority),
QueryParam::String(episode_name),
QueryParam::String(quality.clone()),
QueryParam::String(series_name.clone()),
season_name
.map(QueryParam::String)
.unwrap_or(QueryParam::Null),
episode_number
.map(QueryParam::Int)
.unwrap_or(QueryParam::Null),
season_number
.map(QueryParam::Int)
.unwrap_or(QueryParam::Null),
],
);
db_service
.execute(insert_query)
.await
.map_err(|e| e.to_string())?;
let id_query = Query::with_params(
"SELECT id FROM downloads WHERE item_id = ? AND user_id = ?",
vec![
QueryParam::String(episode_id),
QueryParam::String(user_id.clone()),
],
);
let download_id: i64 = db_service
.query_one(id_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
download_ids.push(download_id);
}
info!(
"[download_series] Queued {} episodes for series '{}'",
total_episodes, series_name
);
Ok(download_ids)
}
/// Queue all episodes of a specific season for download
#[tauri::command]
#[specta::specta]
// One of the eight arguments is a Tauri `State<'_, _>` injection; the rest are
// the season's identifying fields. Folding them into a struct would change the
// IPC contract and the generated TypeScript for no readability gain.
#[allow(clippy::too_many_arguments)]
pub async fn download_season(
db: State<'_, DatabaseWrapper>,
season_id: String,
series_name: String,
season_name: String,
season_number: i32,
user_id: String,
base_path: String,
quality_preset: Option<String>,
) -> Result<Vec<i64>, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let quality = quality_preset.unwrap_or_else(|| "original".to_string());
// Get all episodes for this season, ordered by episode number
let episodes_query = Query::with_params(
"SELECT id, name, index_number
FROM items
WHERE parent_id = ? AND item_type = 'Episode'
ORDER BY index_number",
vec![QueryParam::String(season_id)],
);
let episodes: Vec<(String, String, Option<i32>)> = db_service
.query_many(episodes_query, |row| {
Ok((row.get(0)?, row.get(1)?, row.get(2)?))
})
.await
.map_err(|e| e.to_string())?;
let mut download_ids = Vec::new();
// Queue each episode with priority based on episode number
for (idx, (episode_id, episode_name, episode_number)) in episodes.into_iter().enumerate() {
let priority = 1000 - idx as i32;
let episode_num = episode_number.unwrap_or(1);
let file_name = format!(
"S{:02}E{:02}_{}.mp4",
season_number,
episode_num,
sanitize_filename(&episode_name)
);
let file_path = format!(
"{}/{}/{}",
base_path,
sanitize_filename(&series_name),
file_name
);
let insert_query = Query::with_params(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at,
item_name, quality_preset, media_type, series_name, season_name,
episode_number, season_number)
VALUES (?, ?, ?, 'pending', ?, CURRENT_TIMESTAMP, ?, ?, 'video', ?, ?, ?, ?)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = excluded.priority,
status = 'pending',
queued_at = CURRENT_TIMESTAMP,
quality_preset = excluded.quality_preset",
vec![
QueryParam::String(episode_id.clone()),
QueryParam::String(user_id.clone()),
QueryParam::String(file_path),
QueryParam::Int(priority),
QueryParam::String(episode_name),
QueryParam::String(quality.clone()),
QueryParam::String(series_name.clone()),
QueryParam::String(season_name.clone()),
QueryParam::Int(episode_num),
QueryParam::Int(season_number),
],
);
db_service
.execute(insert_query)
.await
.map_err(|e| e.to_string())?;
let id_query = Query::with_params(
"SELECT id FROM downloads WHERE item_id = ? AND user_id = ?",
vec![
QueryParam::String(episode_id),
QueryParam::String(user_id.clone()),
],
);
let download_id: i64 = db_service
.query_one(id_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
download_ids.push(download_id);
}
info!(
"[download_season] Queued {} episodes for {} - {}",
download_ids.len(),
series_name,
season_name
);
Ok(download_ids)
}
/// Helper to compute download statistics from a list of downloads
#[allow(dead_code)]
fn compute_download_stats(downloads: &[DownloadInfo]) -> DownloadStats {
let mut stats = DownloadStats {
total: downloads.len(),
active_count: 0,
queued_count: 0,
completed_count: 0,
failed_count: 0,
paused_count: 0,
};
for download in downloads {
match download.status.as_str() {
"downloading" => stats.active_count += 1,
"pending" => stats.queued_count += 1,
"completed" => stats.completed_count += 1,
"failed" => stats.failed_count += 1,
"paused" => stats.paused_count += 1,
_ => {}
}
}
stats
}
/// Get all downloads for a user, optionally filtered by status
#[tauri::command]
#[specta::specta]
pub async fn get_downloads(
db: State<'_, DatabaseWrapper>,
user_id: String,
status_filter: Option<Vec<String>>,
) -> Result<DownloadsResponse, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Use COALESCE to prefer stored metadata over LEFT JOIN results
// This ensures correct display even when items aren't synced locally
let (query_str, params) = if let Some(ref statuses) = status_filter {
let placeholders = statuses.iter().map(|_| "?").collect::<Vec<_>>().join(",");
let sql = format!(
"SELECT d.id, d.item_id, d.user_id, d.file_path, d.file_size, d.mime_type, d.status, d.progress,
d.bytes_downloaded, d.queued_at, d.started_at, d.completed_at, d.error_message,
d.retry_count, d.priority,
COALESCE(d.item_name, i.name) as item_name,
COALESCE(d.artist_name, i.artists) as artist_name,
COALESCE(d.album_name, i.album_name) as album_name,
COALESCE(d.series_name, i.series_name) as series_name,
COALESCE(d.season_name, i.season_name) as season_name,
COALESCE(d.episode_number, i.index_number) as episode_number,
COALESCE(d.season_number, i.parent_index_number) as season_number,
d.quality_preset,
COALESCE(d.media_type, 'audio') as media_type,
COALESCE(d.download_source, 'user') as download_source
FROM downloads d
LEFT JOIN items i ON d.item_id = i.id
WHERE d.user_id = ? AND d.status IN ({})
ORDER BY d.priority DESC, d.queued_at ASC",
placeholders
);
let mut p = vec![QueryParam::String(user_id.clone())];
p.extend(statuses.iter().map(|s| QueryParam::String(s.clone())));
(sql, p)
} else {
let sql = "SELECT d.id, d.item_id, d.user_id, d.file_path, d.file_size, d.mime_type, d.status, d.progress,
d.bytes_downloaded, d.queued_at, d.started_at, d.completed_at, d.error_message,
d.retry_count, d.priority,
COALESCE(d.item_name, i.name) as item_name,
COALESCE(d.artist_name, i.artists) as artist_name,
COALESCE(d.album_name, i.album_name) as album_name,
COALESCE(d.series_name, i.series_name) as series_name,
COALESCE(d.season_name, i.season_name) as season_name,
COALESCE(d.episode_number, i.index_number) as episode_number,
COALESCE(d.season_number, i.parent_index_number) as season_number,
d.quality_preset,
COALESCE(d.media_type, 'audio') as media_type,
COALESCE(d.download_source, 'user') as download_source
FROM downloads d
LEFT JOIN items i ON d.item_id = i.id
WHERE d.user_id = ?
ORDER BY d.priority DESC, d.queued_at ASC"
.to_string();
(sql, vec![QueryParam::String(user_id)])
};
let query = Query::with_params(query_str, params);
let downloads = db_service
.query_many(query, map_download_row)
.await
.map_err(|e| e.to_string())?;
// Compute stats in single pass
let stats = compute_download_stats(&downloads);
Ok(DownloadsResponse { downloads, stats })
}
/// Pause a download.
///
/// Writing `status = 'paused'` is only half of it, and used to be all of it: the
/// streaming task knew nothing about the row and kept running, then overwrote it
/// with `completed`/`failed` when it finished. The row flicked to "paused" and
/// undid itself — the reported "pause does not work". Signalling the worker is
/// what actually stops the bytes; it leaves the `.part` file in place so
/// [`resume_download`] can continue from it.
///
/// A queued (not yet started) download has no worker to signal, and the status
/// write alone is enough — the pump skips anything that is not `pending`.
///
/// TRACES: UR-055 | DR-168
#[tauri::command]
#[specta::specta]
pub async fn pause_download(
db: State<'_, DatabaseWrapper>,
download_id: i64,
) -> Result<(), String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let query = Query::with_params(
"UPDATE downloads SET status = 'paused' WHERE id = ? AND status IN ('downloading', 'pending')",
vec![QueryParam::Int64(download_id)],
);
db_service.execute(query).await.map_err(|e| e.to_string())?;
let was_running = crate::download::stop::signal(download_id);
info!(
"[pause] Download {} paused (in flight: {})",
download_id, was_running
);
Ok(())
}
/// Resume a paused download.
///
/// Flipping the row back to `pending` is likewise not enough on its own: the
/// pump is not a poller, it runs when something calls it, so a resumed download
/// sat untouched until some unrelated event happened to pump the queue. That is
/// the other half of "resume does not work".
///
/// TRACES: UR-055 | DR-168
#[tauri::command]
#[specta::specta]
pub async fn resume_download(
app: tauri::AppHandle,
db: State<'_, DatabaseWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
download_id: i64,
) -> Result<(), String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let query = Query::with_params(
"UPDATE downloads SET status = 'pending', error_message = NULL WHERE id = ? AND status IN ('paused', 'failed')",
vec![QueryParam::Int64(download_id)],
);
db_service.execute(query).await.map_err(|e| e.to_string())?;
// Drop any stale stop flag before the pump can start this id again, or the
// resumed run would read the pause that stopped it and halt immediately.
crate::download::stop::clear(download_id);
let active_downloads = {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.get_active_downloads()
};
pump_download_queue(app, db_service, active_downloads).await;
Ok(())
}
/// Cancel a download
#[tauri::command]
#[specta::specta]
pub async fn cancel_download(
db: State<'_, DatabaseWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
download_id: i64,
) -> Result<(), String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Get file path before deleting
let file_query = Query::with_params(
"SELECT file_path FROM downloads WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
let file_path: Option<String> = db_service
.query_optional(file_query, |row| row.get(0))
.await
.ok()
.flatten();
// Delete from database
let delete_query = Query::with_params(
"DELETE FROM downloads WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
// Stop the worker if this download is actually running. Without this the
// task keeps streaming into a `.part` file whose `downloads` row has just
// been deleted — bytes with nothing pointing at them, and the file below is
// removed while still being written to. (DR-168)
crate::download::stop::signal(download_id);
crate::download::stop::clear(download_id);
// Unregister from download manager (in case it was active)
{
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.unregister_download(download_id);
info!(
"Cancelled download {}. Active downloads: {}",
download_id,
manager.active_count()
);
}
// Delete the partial file, and any completed file, if present. Both go
// through `partial_path` so this cannot drift from what the worker writes —
// it did, and every cancelled download leaked its partial. (DR-169)
if let Some(path) = file_path {
let target = std::path::PathBuf::from(&path);
let _ = std::fs::remove_file(crate::download::worker::partial_path(&target));
}
Ok(())
}
/// Mark a download as completed
#[tauri::command]
#[specta::specta]
pub async fn mark_download_completed(
db: State<'_, DatabaseWrapper>,
download_id: i64,
bytes_downloaded: i64,
file_path: String,
) -> Result<(), String> {
// Deleting a download reads this straight back into `std::fs::remove_file`,
// so a row must never come to name a file outside the download directory.
// The worker reports the absolute path it wrote, and joining an absolute
// path onto the root yields it unchanged, so that case is stored verbatim;
// the frontend's fallback to the row's own (relative) path resolves under
// the root, where the worker put it.
// TRACES: DR-211 | UT-205
let file_path = {
let root = download_root(&db)?;
confine_to_root(&root, &root.join(&file_path))?
.to_string_lossy()
.to_string()
};
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let query = Query::with_params(
"UPDATE downloads SET status = 'completed', progress = 1.0, bytes_downloaded = ?,
file_size = ?, file_path = ?, completed_at = CURRENT_TIMESTAMP WHERE id = ?",
vec![
QueryParam::Int64(bytes_downloaded),
QueryParam::Int64(bytes_downloaded),
QueryParam::String(file_path),
QueryParam::Int64(download_id),
],
);
db_service.execute(query).await.map_err(|e| e.to_string())?;
Ok(())
}
/// Mark a download as failed
#[tauri::command]
#[specta::specta]
pub async fn mark_download_failed(
db: State<'_, DatabaseWrapper>,
download_id: i64,
error_message: String,
) -> Result<(), String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
let query = Query::with_params(
"UPDATE downloads SET status = 'failed', error_message = ? WHERE id = ?",
vec![
QueryParam::String(error_message),
QueryParam::Int64(download_id),
],
);
db_service.execute(query).await.map_err(|e| e.to_string())?;
Ok(())
}
/// Start downloading a file immediately
/// This command actually downloads the file using the worker
#[tauri::command]
#[specta::specta]
pub async fn start_download(
db: State<'_, DatabaseWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
app: tauri::AppHandle,
download_id: i64,
stream_url: String,
target_dir: String,
) -> Result<(), String> {
use crate::download::events::DownloadEvent;
use tauri::Emitter;
debug!("start_download called for download_id: {}", download_id);
debug!(" stream_url: {}", stream_url);
debug!(" target_dir: {}", target_dir);
// Check concurrent download limit and register this download
{
let manager = download_manager.0.lock().map_err(|e| {
error!("Failed to lock download manager: {}", e);
format!("Failed to lock download manager: {}", e)
})?;
if !manager.can_start_download() {
warn!(
"Cannot start download: maximum concurrent downloads ({}) reached",
manager.max_concurrent()
);
debug!(" Active downloads: {}", manager.active_count());
return Err(format!(
"Maximum concurrent downloads ({}) reached. Please wait for existing downloads to complete.",
manager.max_concurrent()
));
}
// Register this download as active
let registered = manager.register_download(download_id);
if !registered {
warn!(
"Failed to register download {}: already registered or limit reached",
download_id
);
return Err("Download already in progress or limit reached".to_string());
}
info!(
"Download {} registered. Active downloads: {}/{}",
download_id,
manager.active_count(),
manager.max_concurrent()
);
}
// Get download info from DB
let db_service = {
let database = db.0.lock().map_err(|e| {
error!("Failed to lock database: {}", e);
e.to_string()
})?;
Arc::new(database.service())
};
let info_query = Query::with_params(
"SELECT item_id, file_path, file_size FROM downloads WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
let (item_id, file_path, file_size): (String, String, Option<i64>) = db_service
.query_one(info_query, |row| {
Ok((row.get(0)?, row.get(1)?, row.get(2)?))
})
.await
.map_err(|e| {
error!("Failed to query download info: {}", e);
e.to_string()
})?;
debug!(
" Retrieved: item_id={}, file_path={}, file_size={:?}",
item_id, file_path, file_size
);
// Both halves of this join reached us from the frontend, so resolve them
// against the download directory before a single byte is written.
// TRACES: DR-211 | UT-205
let target_path = {
let root = download_root(&db)?;
confine_to_root(&root, &PathBuf::from(&target_dir).join(&file_path))?
};
// Make a HEAD request to get the file size from Content-Length header
debug!("Making HEAD request to get file size...");
let head_response = reqwest::Client::new().head(&stream_url).send().await;
let file_size_from_server = match head_response {
Ok(response) => {
let size = response
.headers()
.get(reqwest::header::CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse::<i64>().ok());
if let Some(size) = size {
debug!(
" Got file size from server: {} bytes ({} MB)",
size,
size / 1024 / 1024
);
} else {
warn!(" Server didn't provide Content-Length header");
}
size
}
Err(e) => {
warn!(" HEAD request failed: {}, continuing anyway...", e);
None
}
};
// Update status to downloading and save file_size if we got it.
// Also persist the resolved stream URL + target dir so the queue pump can
// restart/resume this download by itself if needed.
let update_query = if let Some(size) = file_size_from_server {
Query::with_params(
"UPDATE downloads SET status = 'downloading', started_at = CURRENT_TIMESTAMP, file_size = ?, stream_url = ?, target_dir = ? WHERE id = ?",
vec![
QueryParam::Int64(size),
QueryParam::String(stream_url.clone()),
QueryParam::String(target_dir.clone()),
QueryParam::Int64(download_id),
],
)
} else {
Query::with_params(
"UPDATE downloads SET status = 'downloading', started_at = CURRENT_TIMESTAMP, stream_url = ?, target_dir = ? WHERE id = ?",
vec![
QueryParam::String(stream_url.clone()),
QueryParam::String(target_dir.clone()),
QueryParam::Int64(download_id),
],
)
};
db_service
.execute(update_query)
.await
.map_err(|e| e.to_string())?;
// Emit started event
let started_event = DownloadEvent::Started {
download_id,
item_id: item_id.clone(),
};
debug!("Emitting download-event: {:?}", started_event);
debug!(
" Serialized: {}",
serde_json::to_string(&started_event).unwrap_or_default()
);
match app.emit("download-event", started_event) {
Ok(_) => debug!(" Event emitted successfully"),
Err(e) => error!(" Event emit failed: {:?}", e),
}
// Get a clone of the active downloads Arc for unregistering later
let active_downloads = {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.get_active_downloads()
};
// Run the worker in the background; on completion/failure it frees the slot
// and pumps the next pending download.
spawn_download_worker(
app.clone(),
download_id,
item_id,
stream_url,
target_path,
active_downloads,
);
Ok(())
}
/// Enqueue a download with its resolved stream URL, then let the queue pump
/// start it (or a higher-priority pending item) when a slot is free.
///
/// Unlike [`start_download`], this never errors when the concurrency limit is
/// reached: the URL is persisted on the row and the pump will pick it up once a
/// slot frees. This is the path bulk operations (album/series/season) use so
/// every queued item eventually downloads without the frontend re-issuing it.
#[tauri::command]
#[specta::specta]
pub async fn enqueue_download(
db: State<'_, DatabaseWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
app: tauri::AppHandle,
download_id: i64,
stream_url: String,
target_dir: String,
) -> Result<(), String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Persist the resolved URL/dir and mark the row pending so the pump can
// start it. We don't flip to 'downloading' here — the pump owns that.
let update_query = Query::with_params(
"UPDATE downloads SET status = 'pending', stream_url = ?, target_dir = ? WHERE id = ?",
vec![
QueryParam::String(stream_url),
QueryParam::String(target_dir),
QueryParam::Int64(download_id),
],
);
db_service
.execute(update_query)
.await
.map_err(|e| e.to_string())?;
// Kick the pump: it will start as many pending downloads as there are slots.
let active_downloads = {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.get_active_downloads()
};
pump_download_queue(app, db_service, active_downloads).await;
Ok(())
}
/// Enqueue a batch of already-queued video downloads, resolving each one's
/// transcode URL from the repository using the `quality_preset` stored on the
/// row. Then let the pump start them subject to the concurrency limit.
///
/// This is the bulk video path (series/season): `download_series`/
/// `download_season` insert the rows, then this resolves URLs and enqueues them
/// so they actually start. Resolving server-side avoids round-tripping every
/// episode URL through the frontend.
#[tauri::command]
#[specta::specta]
pub async fn enqueue_video_downloads(
db: State<'_, DatabaseWrapper>,
download_manager: State<'_, DownloadManagerWrapper>,
repository: State<'_, crate::commands::repository::RepositoryManagerWrapper>,
app: tauri::AppHandle,
handle: String,
download_ids: Vec<i64>,
target_dir: String,
) -> Result<(), String> {
let repo = repository.0.get(&handle).ok_or("Repository not found")?;
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
for download_id in download_ids {
// Read the item + quality preset for this queued download.
let info_query = Query::with_params(
"SELECT item_id, COALESCE(quality_preset, 'original') FROM downloads WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
let (item_id, quality): (String, String) = match db_service
.query_one(info_query, |row| Ok((row.get(0)?, row.get(1)?)))
.await
{
Ok(row) => row,
Err(e) => {
warn!("[enqueue_video] Skipping download {}: {}", download_id, e);
continue;
}
};
// Build the download URL, resolving the source's audio codec first so a
// track this device cannot decode is re-encoded on the way down rather
// than saved as a silent file (DR-167).
let stream_url =
crate::repository::resolve_video_download_url(repo.as_ref(), &item_id, &quality, None)
.await;
let update_query = Query::with_params(
"UPDATE downloads SET status = 'pending', stream_url = ?, target_dir = ? WHERE id = ?",
vec![
QueryParam::String(stream_url),
QueryParam::String(target_dir.clone()),
QueryParam::Int64(download_id),
],
);
if let Err(e) = db_service.execute(update_query).await {
warn!(
"[enqueue_video] Failed to persist URL for download {}: {}",
download_id, e
);
}
}
// Pump once: starts up to max_concurrent, the rest drain as slots free.
let active_downloads = {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.get_active_downloads()
};
pump_download_queue(app, db_service, active_downloads).await;
Ok(())
}
/// Whether the current network permits downloads, given the user's WiFi-only
/// preference.
///
/// Reads `wifi_only` from the SmartCache config (the single home of the
/// setting) and checks it against the transport reported by the platform. On
/// desktop the transport defaults to unmetered ethernet, so this is always
/// true there.
///
/// TRACES: UR-053 | DR-074
pub(crate) async fn downloads_allowed_on_current_network(app: &tauri::AppHandle) -> bool {
let wifi_only = {
let smart_cache = app.state::<SmartCacheWrapper>();
let cache = match smart_cache.0.lock() {
Ok(c) => c,
Err(e) => {
error!("[pump] Failed to lock smart cache: {}", e);
// Fail open: a lock problem must not silently wedge downloads.
return true;
}
};
cache.get_config().map(|c| c.wifi_only).unwrap_or(false)
};
if !wifi_only {
return true;
}
let network = app.state::<NetworkStateWrapper>();
network.0.allows_download(true).await
}
/// Start as many pending downloads as there are free concurrency slots.
///
/// Picks the highest-priority `pending` rows that have a persisted `stream_url`
/// (FIFO within a priority), registers each, flips it to `downloading`, and
/// spawns a worker. Each spawned worker calls this again on completion/failure,
/// so the queue drains itself without any frontend involvement.
pub(crate) async fn pump_download_queue(
app: tauri::AppHandle,
db_service: Arc<crate::storage::db_service::RusqliteService>,
active_downloads: Arc<Mutex<std::collections::HashSet<i64>>>,
) {
use crate::download::events::DownloadEvent;
use tauri::Emitter;
// WiFi-only gate (UR-053): when the user has restricted downloads to
// unmetered networks and we're on cellular (or can't tell), leave every
// pending row exactly as it is. They stay 'pending' and the Android
// network callback re-pumps us as soon as an acceptable network appears.
if !downloads_allowed_on_current_network(&app).await {
info!("[pump] Downloads paused: waiting for an unmetered network (WiFi-only enabled)");
let _ = app.emit("download-event", DownloadEvent::WaitingForNetwork);
return;
}
let max_concurrent = {
let manager = app.state::<DownloadManagerWrapper>();
let manager = match manager.0.lock() {
Ok(m) => m,
Err(e) => {
error!("[pump] Failed to lock download manager: {}", e);
return;
}
};
manager.max_concurrent()
};
loop {
// How many slots are free right now?
let free_slots = {
let active = match active_downloads.lock() {
Ok(a) => a,
Err(e) => {
error!("[pump] Failed to lock active downloads: {}", e);
return;
}
};
max_concurrent.saturating_sub(active.len())
};
if free_slots == 0 {
return;
}
// Find the next pending, startable download (has a stream URL). Exclude
// anything already registered as active to avoid double-starting.
let next_query = Query::with_params(
"SELECT id, item_id, file_path, stream_url, target_dir
FROM downloads
WHERE status = 'pending'
AND stream_url IS NOT NULL
AND target_dir IS NOT NULL
ORDER BY priority DESC, queued_at ASC",
vec![],
);
let candidates: Vec<(i64, String, String, String, String)> = match db_service
.query_many(next_query, |row| {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
))
})
.await
{
Ok(rows) => rows,
Err(e) => {
error!("[pump] Failed to query pending downloads: {}", e);
return;
}
};
// Pick the first candidate not already active.
let next = candidates.into_iter().find(|(id, _, _, _, _)| {
active_downloads
.lock()
.map(|active| !active.contains(id))
.unwrap_or(false)
});
let (download_id, item_id, file_path, stream_url, target_dir) = match next {
Some(n) => n,
None => return, // Nothing pending to start
};
// Confine the row's path before it takes a slot. A row whose target
// escapes the download directory can never start, so it is failed here
// rather than picked again on the next pass — this loop re-queries, so
// merely skipping it would not terminate.
// TRACES: DR-211 | UT-205
let confined = {
let db_state = app.state::<DatabaseWrapper>();
download_root(&db_state).and_then(|root| {
confine_to_root(&root, &PathBuf::from(&target_dir).join(&file_path))
})
};
let target_path = match confined {
Ok(path) => path,
Err(e) => {
error!("[pump] Refusing download {}: {}", download_id, e);
let fail_query = Query::with_params(
"UPDATE downloads SET status = 'failed', error_message = ? WHERE id = ?",
vec![QueryParam::String(e), QueryParam::Int64(download_id)],
);
if let Err(db_err) = db_service.execute(fail_query).await {
error!(
"[pump] Failed to mark download {} failed: {}",
download_id, db_err
);
return;
}
continue;
}
};
// Register the slot. If registration fails (race: another pump filled
// the last slot), stop — we'll be re-pumped when a slot frees.
{
let manager = app.state::<DownloadManagerWrapper>();
let manager = match manager.0.lock() {
Ok(m) => m,
Err(e) => {
error!("[pump] Failed to lock download manager: {}", e);
return;
}
};
if !manager.register_download(download_id) {
return;
}
info!(
"[pump] Download {} started. Active downloads: {}/{}",
download_id,
manager.active_count(),
manager.max_concurrent()
);
}
// Mark as downloading and stamp started_at.
let update_query = Query::with_params(
"UPDATE downloads SET status = 'downloading', started_at = CURRENT_TIMESTAMP WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
if let Err(e) = db_service.execute(update_query).await {
error!(
"[pump] Failed to mark download {} downloading: {}",
download_id, e
);
if let Ok(mut a) = active_downloads.lock() {
a.remove(&download_id);
}
continue;
}
// Emit started event so the UI flips the row.
let _ = app.emit(
"download-event",
DownloadEvent::Started {
download_id,
item_id: item_id.clone(),
},
);
spawn_download_worker(
app.clone(),
download_id,
item_id,
stream_url,
target_path,
active_downloads.clone(),
);
}
}
/// Spawn the background worker for one download. On completion or failure it
/// unregisters the slot, emits the terminal event, and pumps the queue so the
/// next pending download starts automatically.
fn spawn_download_worker(
app: tauri::AppHandle,
download_id: i64,
item_id: String,
stream_url: String,
target_path: std::path::PathBuf,
active_downloads: Arc<Mutex<std::collections::HashSet<i64>>>,
) {
use crate::download::events::DownloadEvent;
use crate::download::{DownloadTask, DownloadWorker};
use tauri::Emitter;
let task = DownloadTask {
url: stream_url,
target_path: target_path.clone(),
};
tauri::async_runtime::spawn(async move {
debug!("Download task started for download_id: {}", download_id);
let worker = DownloadWorker::new();
// Progress callback that emits events to the frontend
let progress_app = app.clone();
let progress_item_id = item_id.clone();
let on_progress = move |bytes_downloaded: u64, total_bytes: Option<u64>| {
let progress = total_bytes
.filter(|&t| t > 0)
.map(|t| bytes_downloaded as f64 / t as f64)
.unwrap_or(0.0);
let event = DownloadEvent::Progress {
download_id,
item_id: progress_item_id.clone(),
bytes_downloaded: bytes_downloaded as i64,
total_bytes: total_bytes.map(|t| t as i64),
progress,
};
let _ = progress_app.emit("download-event", event);
};
// Registering returns a fresh flag, so a download resumed after a pause
// does not inherit the stop that ended its previous run. (DR-168)
let stop_flag = crate::download::stop::register(download_id);
let result = worker.download(&task, &stop_flag, on_progress).await;
crate::download::stop::clear(download_id);
// Free the slot before pumping so the next download can take it.
if let Ok(mut active) = active_downloads.lock() {
active.remove(&download_id);
debug!(
" Unregistered download {}. Active downloads: {}",
download_id,
active.len()
);
}
// The pump runs downloads in the background, so the terminal status MUST
// be persisted to the DB here — the frontend event handler only writes it
// when that download happens to be loaded in its store, which is not the
// case for auto-pumped rows (or any completion while the downloads page is
// closed). `check_for_local_download` filters on status = 'completed', so a
// missed write leaves finished files unrecognized: albums never show as
// downloaded and playback never switches from the (expiring) stream to the
// local file, cutting tracks off mid-play.
let db_service = {
let db = app.state::<DatabaseWrapper>();
let database = match db.0.lock() {
Ok(d) => d,
Err(e) => {
error!(
"[pump] Failed to lock database after download {}: {}",
download_id, e
);
return;
}
};
Arc::new(database.service())
};
match result {
Ok(res) => {
info!(
"Download completed successfully: {} bytes",
res.bytes_downloaded
);
let file_path = target_path.to_string_lossy().to_string();
let update = Query::with_params(
"UPDATE downloads SET status = 'completed', progress = 1.0, \
bytes_downloaded = ?, file_size = ?, file_path = ?, \
completed_at = CURRENT_TIMESTAMP WHERE id = ?",
vec![
QueryParam::Int64(res.bytes_downloaded as i64),
QueryParam::Int64(res.bytes_downloaded as i64),
QueryParam::String(file_path.clone()),
QueryParam::Int64(download_id),
],
);
if let Err(e) = db_service.execute(update).await {
error!(
"[pump] Failed to persist completed status for download {}: {}",
download_id, e
);
}
let completed_event = DownloadEvent::Completed {
download_id,
item_id,
file_path,
};
match app.emit("download-event", completed_event) {
Ok(_) => debug!(" Completed event emitted successfully"),
Err(e) => error!(" Completed event emit failed: {:?}", e),
}
}
// A pause or cancel is not a failure. The row already says `paused`
// (or the row is gone, for a cancel), and overwriting that with
// `failed` is what made a pause look like an error and stranded the
// download outside the resumable set. The `.part` file is deliberately
// left alone — it is what the resume continues from. (DR-168)
Err(e) if e.is_stopped() => {
info!(
"[pump] Download {} stopped by request; partial file kept for resume",
download_id
);
}
Err(e) => {
error!("Download failed: {:?}", e);
let update = Query::with_params(
"UPDATE downloads SET status = 'failed', error_message = ? WHERE id = ?",
vec![
QueryParam::String(e.to_string()),
QueryParam::Int64(download_id),
],
);
if let Err(db_err) = db_service.execute(update).await {
error!(
"[pump] Failed to persist failed status for download {}: {}",
download_id, db_err
);
}
let failed_event = DownloadEvent::Failed {
download_id,
item_id,
error: e.to_string(),
};
match app.emit("download-event", failed_event) {
Ok(_) => debug!(" Failed event emitted successfully"),
Err(e) => error!(" Failed event emit failed: {:?}", e),
}
}
}
// A slot just freed — start the next pending download (if any).
pump_download_queue(app.clone(), db_service, active_downloads).await;
});
}
/// Delete a completed download
#[tauri::command]
#[specta::specta]
pub async fn delete_download(
db: State<'_, DatabaseWrapper>,
download_id: i64,
) -> Result<(), String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Get file path
let file_query = Query::with_params(
"SELECT file_path FROM downloads WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
let file_path: Option<String> = db_service
.query_optional(file_query, |row| row.get(0))
.await
.ok()
.flatten();
// Delete from database
let delete_query = Query::with_params(
"DELETE FROM downloads WHERE id = ?",
vec![QueryParam::Int64(download_id)],
);
db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
// Delete actual file if exists
if let Some(path) = file_path {
let _ = std::fs::remove_file(&path); // Ignore errors
}
Ok(())
}
/// Helper function to map database row to DownloadInfo
fn map_download_row(row: &rusqlite::Row) -> rusqlite::Result<DownloadInfo> {
Ok(DownloadInfo {
id: row.get(0)?,
item_id: row.get(1)?,
user_id: row.get(2)?,
file_path: row.get(3)?,
file_size: row.get(4)?,
mime_type: row.get(5)?,
status: row.get(6)?,
progress: row.get(7)?,
bytes_downloaded: row.get(8)?,
queued_at: row.get(9)?,
started_at: row.get(10)?,
completed_at: row.get(11)?,
error_message: row.get(12)?,
retry_count: row.get(13)?,
priority: row.get(14)?,
item_name: row.get(15)?,
artist_name: row.get(16)?,
album_name: row.get(17)?,
// Video-specific metadata
series_name: row.get(18)?,
season_name: row.get(19)?,
episode_number: row.get(20)?,
season_number: row.get(21)?,
quality_preset: row.get(22)?,
media_type: row
.get::<_, Option<String>>(23)?
.unwrap_or_else(|| "audio".to_string()),
download_source: row
.get::<_, Option<String>>(24)?
.unwrap_or_else(|| "user".to_string()),
})
}
/// Storage statistics for downloads
#[derive(specta::Type, Debug, Clone, serde::Serialize)]
pub struct StorageStats {
pub total_bytes: i64,
pub total_items: i64,
pub albums: Vec<AlbumStorageInfo>,
}
/// Storage info for a single album
#[derive(specta::Type, Debug, Clone, serde::Serialize)]
pub struct AlbumStorageInfo {
pub album_id: String,
pub album_name: String,
pub artist_name: Option<String>,
pub bytes_used: i64,
pub track_count: i64,
}
/// Get storage statistics for downloads
#[tauri::command]
#[specta::specta]
pub async fn get_download_storage_stats(
db: State<'_, DatabaseWrapper>,
user_id: String,
) -> Result<StorageStats, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Get total storage and item count
let total_query = Query::with_params(
"SELECT COALESCE(SUM(file_size), 0), COUNT(*)
FROM downloads
WHERE user_id = ? AND status = 'completed'",
vec![QueryParam::String(user_id.clone())],
);
let (total_bytes, total_items): (i64, i64) = db_service
.query_one(total_query, |row| Ok((row.get(0)?, row.get(1)?)))
.await
.map_err(|e| e.to_string())?;
// Get storage breakdown by album
let albums_query = Query::with_params(
"SELECT
COALESCE(i.album_id, 'unknown') as album_id,
COALESCE(i.album_name, 'Unknown Album') as album_name,
i.artists as artist_name,
COALESCE(SUM(d.file_size), 0) as bytes_used,
COUNT(*) as track_count
FROM downloads d
LEFT JOIN items i ON d.item_id = i.id
WHERE d.user_id = ? AND d.status = 'completed'
GROUP BY COALESCE(i.album_id, 'unknown')
ORDER BY bytes_used DESC",
vec![QueryParam::String(user_id)],
);
let albums: Vec<AlbumStorageInfo> = db_service
.query_many(albums_query, |row| {
Ok(AlbumStorageInfo {
album_id: row.get(0)?,
album_name: row.get(1)?,
artist_name: row.get(2)?,
bytes_used: row.get(3)?,
track_count: row.get(4)?,
})
})
.await
.map_err(|e| e.to_string())?;
Ok(StorageStats {
total_bytes,
total_items,
albums,
})
}
/// Delete all downloads for a user
#[tauri::command]
#[specta::specta]
pub async fn delete_all_downloads(
db: State<'_, DatabaseWrapper>,
user_id: String,
) -> Result<i64, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Get all file paths for completed downloads
let file_query = Query::with_params(
"SELECT file_path FROM downloads WHERE user_id = ? AND status = 'completed'",
vec![QueryParam::String(user_id.clone())],
);
let file_paths: Vec<String> = db_service
.query_many(file_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
// Delete all downloads for user
let delete_query = Query::with_params(
"DELETE FROM downloads WHERE user_id = ?",
vec![QueryParam::String(user_id)],
);
let deleted_count = db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
// Delete actual files
for path in file_paths {
let _ = std::fs::remove_file(&path); // Ignore errors
let _ = std::fs::remove_file(format!("{}.part", path)); // Also clean up partials
}
Ok(deleted_count as i64)
}
/// Clear all stale pending/failed/paused downloads
#[tauri::command]
#[specta::specta]
pub async fn clear_stale_downloads(
db: State<'_, DatabaseWrapper>,
user_id: String,
) -> Result<i64, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Ids as well as paths: a stale row may still have a worker attached (a
// 'downloading' row that was paused mid-flight is 'paused' here), and
// deleting the row without stopping the task leaves it writing to a file we
// are about to remove. (DR-168)
let file_query = Query::with_params(
"SELECT id, file_path FROM downloads WHERE user_id = ? AND status IN ('pending', 'paused', 'failed')",
vec![QueryParam::String(user_id.clone())],
);
let stale: Vec<(i64, String)> = db_service
.query_many(file_query, |row| Ok((row.get(0)?, row.get(1)?)))
.await
.map_err(|e| e.to_string())?;
for (id, _) in &stale {
crate::download::stop::signal(*id);
crate::download::stop::clear(*id);
}
let file_paths: Vec<String> = stale.into_iter().map(|(_, path)| path).collect();
// Delete all pending, paused, and failed downloads (but keep completed ones)
let delete_query = Query::with_params(
"DELETE FROM downloads WHERE user_id = ? AND status IN ('pending', 'paused', 'failed')",
vec![QueryParam::String(user_id)],
);
let deleted_count = db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
// Delete any partial files, via the shared helper so this cannot drift from
// what the worker actually writes. (DR-169)
for path in file_paths {
let target = std::path::PathBuf::from(&path);
let _ = std::fs::remove_file(&target);
let _ = std::fs::remove_file(crate::download::worker::partial_path(&target));
}
Ok(deleted_count as i64)
}
/// Delete all downloads for a specific album
#[tauri::command]
#[specta::specta]
pub async fn delete_album_downloads(
db: State<'_, DatabaseWrapper>,
album_id: String,
user_id: String,
) -> Result<i64, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// Get file paths for this album's downloads
let file_query = Query::with_params(
"SELECT d.file_path FROM downloads d
JOIN items i ON d.item_id = i.id
WHERE d.user_id = ? AND i.album_id = ? AND d.status = 'completed'",
vec![
QueryParam::String(user_id.clone()),
QueryParam::String(album_id.clone()),
],
);
let file_paths: Vec<String> = db_service
.query_many(file_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
// Delete downloads for this album
let delete_query = Query::with_params(
"DELETE FROM downloads WHERE user_id = ? AND item_id IN (SELECT id FROM items WHERE album_id = ?)",
vec![QueryParam::String(user_id), QueryParam::String(album_id)],
);
let deleted_count = db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
// Delete actual files
for path in file_paths {
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(format!("{}.part", path));
}
Ok(deleted_count as i64)
}
/// Remove every completed download at or under a container item.
///
/// Works at any level of the Downloaded browse: a leaf (removes just that
/// download), an album/season/series (removes all downloaded descendants linked
/// via album_id/season_id/series_id/parent_id). Deletes the DB rows and the
/// on-disk files. Returns the number of downloads removed. Idempotent.
///
/// TRACES: UR-055 | DR-083
#[tauri::command]
#[specta::specta]
pub async fn delete_downloads_under(
db: State<'_, DatabaseWrapper>,
item_id: String,
user_id: String,
) -> Result<i64, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
};
// The item itself, or any child linked to it by container id.
const SCOPE: &str = "d.user_id = ? AND d.status = 'completed'
AND (
d.item_id = ?
OR d.item_id IN (
SELECT c.id FROM items c
WHERE c.album_id = ? OR c.season_id = ? OR c.series_id = ? OR c.parent_id = ?
)
)";
let file_query = Query::with_params(
format!("SELECT d.file_path FROM downloads d WHERE {SCOPE}"),
vec![
QueryParam::String(user_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
],
);
let file_paths: Vec<String> = db_service
.query_many(file_query, |row| row.get(0))
.await
.map_err(|e| e.to_string())?;
let delete_query = Query::with_params(
format!("DELETE FROM downloads WHERE id IN (SELECT d.id FROM downloads d WHERE {SCOPE})"),
vec![
QueryParam::String(user_id),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::String(item_id),
],
);
let deleted_count = db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
for path in file_paths {
let _ = std::fs::remove_file(&path);
let _ = std::fs::remove_file(format!("{}.part", path));
}
Ok(deleted_count as i64)
}
/// Download manager statistics
#[derive(specta::Type, Debug, Clone, serde::Serialize)]
pub struct DownloadManagerStats {
pub max_concurrent: usize,
pub active_count: usize,
pub available_slots: usize,
}
/// Get download manager statistics
#[tauri::command]
#[specta::specta]
pub async fn get_download_manager_stats(
download_manager: State<'_, DownloadManagerWrapper>,
) -> Result<DownloadManagerStats, String> {
let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
let active_count = manager.active_count();
let max_concurrent = manager.max_concurrent();
Ok(DownloadManagerStats {
max_concurrent,
active_count,
available_slots: max_concurrent.saturating_sub(active_count),
})
}
/// Set the maximum concurrent downloads
#[tauri::command]
#[specta::specta]
pub async fn set_max_concurrent_downloads(
download_manager: State<'_, DownloadManagerWrapper>,
max: usize,
) -> Result<(), String> {
let mut manager = download_manager.0.lock().map_err(|e| e.to_string())?;
manager.set_max_concurrent(max);
info!("Set max concurrent downloads to: {}", max);
Ok(())
}
// TRACES: UR-011, UR-018 | DR-015, DR-018 | UT-042, UT-043
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::Database;
use rusqlite::params;
#[test]
fn test_sanitize_filename() {
assert_eq!(sanitize_filename("normal.mp3"), "normal.mp3");
assert_eq!(
sanitize_filename("track/with\\invalid:chars"),
"track_with_invalid_chars"
);
assert_eq!(sanitize_filename("song?.mp3"), "song_.mp3");
assert_eq!(sanitize_filename("file<>|?.txt"), "file____.txt");
}
#[test]
fn test_sanitize_filename_preserves_extension() {
assert_eq!(sanitize_filename("my:song.mp3"), "my_song.mp3");
assert_eq!(sanitize_filename("track/1.flac"), "track_1.flac");
}
/// The download directory as it looks on a device, for the path tests.
const TEST_ROOT: &str = "/data/data/com.dtourolle.jellytau/files";
/// A queued `file_path` cannot walk out of the download directory.
///
/// `download_item` is a command in its own right, so sanitizing in
/// `download_item_and_start` was routed around by invoking it directly, and
/// `start_download` then joined the raw string onto the target directory.
///
/// TRACES: DR-211 | UT-205
#[test]
fn test_queued_download_paths_cannot_escape_the_download_directory() {
let root = Path::new(TEST_ROOT);
assert!(confine_queued_path(root, "downloads/../../../../etc/cron.d/pwn").is_err());
assert!(confine_queued_path(root, "../.bashrc").is_err());
assert!(confine_queued_path(root, "/etc/cron.d/pwn").is_err());
// Why the absolute case needs its own guard rather than folding: the
// join the download path performs discards the base entirely.
//
// clippy::join_absolute_paths flags exactly this shape, and is right to
// in production code — here the discarded base *is* the assertion, so
// the lint is allowed rather than the code changed. Note the lint would
// not have caught the original defect: the real join sites take a
// variable, and the lint only fires on a literal starting with `/`.
#[allow(clippy::join_absolute_paths)]
{
assert_eq!(
PathBuf::from(root).join("/etc/cron.d/pwn"),
PathBuf::from("/etc/cron.d/pwn")
);
}
}
/// The paths the app builds for itself have to survive unchanged: files are
/// already on disk and `downloads` rows point at these exact spellings.
///
/// TRACES: DR-211 | UT-205
#[test]
fn test_queued_download_paths_are_otherwise_unchanged() {
let root = Path::new(TEST_ROOT);
for path in [
"downloads/9f8e7d6c", // MediaCard's queue-for-reconnect
"videos/movies/Arrival.mp4", // VideoDownloadButton
"albums/abc123/01 - Opening.mp3", // queue_album_tracks
// download_series/download_season build an absolute path, because
// their base_path is `${targetDir}/videos`.
"/data/data/com.dtourolle.jellytau/files/videos/Show/S01E02_Pilot.mp4",
] {
assert_eq!(confine_queued_path(root, path).unwrap(), path);
}
// `download_item_and_start` sanitizes the name before calling
// `download_item`; sanitizing it again must not yield a second, different
// name, which would orphan the row and the file it names.
let already = format!("downloads/{}.mp3", sanitize_filename("AC/DC: Live?"));
assert_eq!(confine_queued_path(root, &already).unwrap(), already);
}
/// A completed row's `file_path` is read straight back into
/// `std::fs::remove_file` when the download is deleted, so `mark_download_completed`
/// must not be able to register a file outside the download directory.
///
/// TRACES: DR-211 | UT-205
#[test]
fn test_a_completed_download_cannot_register_a_file_outside_the_root() {
let root = Path::new(TEST_ROOT);
// What the worker actually reports — the absolute path it wrote. Stored
// exactly as it arrives.
let written = "/data/data/com.dtourolle.jellytau/files/downloads/9f8e7d6c";
assert_eq!(
confine_to_root(root, Path::new(written)).unwrap(),
PathBuf::from(written)
);
// The row's own path, if the frontend falls back to it: relative, and it
// resolves to where the worker wrote the file.
assert_eq!(
confine_to_root(root, &root.join("downloads/9f8e7d6c")).unwrap(),
PathBuf::from(written)
);
assert!(confine_to_root(root, Path::new("/home/u/.ssh/id_ed25519")).is_err());
assert!(confine_to_root(
root,
Path::new("/data/data/com.dtourolle.jellytau/files/../../../../etc/passwd")
)
.is_err());
}
/// Helper to set up test database with required foreign key data
fn setup_test_db() -> Database {
let db = Database::open_in_memory().unwrap();
let conn = db.connection();
let conn = conn.lock_safe();
// Create server
conn.execute(
"INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
params!["server1", "Test Server", "http://localhost:8096"],
)
.unwrap();
// Create user
conn.execute(
"INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
params!["user1", "server1", "testuser"],
)
.unwrap();
// Create some items for download testing
conn.execute(
"INSERT INTO items (id, server_id, name, item_type, album_id, index_number) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params!["item1", "server1", "Test Song 1", "Audio", "album1", 1],
)
.unwrap();
conn.execute(
"INSERT INTO items (id, server_id, name, item_type, album_id, index_number) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params!["item2", "server1", "Test Song 2", "Audio", "album1", 2],
)
.unwrap();
conn.execute(
"INSERT INTO items (id, server_id, name, item_type, album_id, index_number) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params!["item3", "server1", "Test Song 3", "Audio", "album1", 3],
)
.unwrap();
drop(conn);
db
}
#[test]
fn test_download_item_returns_correct_id_on_insert() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// Insert a new download
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, mime_type, status, priority, queued_at)
VALUES (?1, ?2, ?3, ?4, 'pending', ?5, CURRENT_TIMESTAMP)",
params!["item1", "user1", "/path/to/file.mp3", "audio/mpeg", 0],
)
.unwrap();
// Query for the download ID (like our fixed code does)
let download_id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params!["item1", "user1"],
|row| row.get(0),
)
.unwrap();
assert!(download_id > 0, "Download ID should be positive");
// Verify the download record exists with correct data
let (status, file_path): (String, String) = conn
.query_row(
"SELECT status, file_path FROM downloads WHERE id = ?1",
params![download_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert_eq!(status, "pending");
assert_eq!(file_path, "/path/to/file.mp3");
}
#[test]
fn test_download_item_upsert_returns_correct_id() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// First insert
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 0, CURRENT_TIMESTAMP)",
params!["item1", "user1", "/path/to/file.mp3"],
)
.unwrap();
let first_id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params!["item1", "user1"],
|row| row.get(0),
)
.unwrap();
// Simulate failure - mark as failed
conn.execute(
"UPDATE downloads SET status = 'failed', error_message = 'Network error' WHERE id = ?1",
params![first_id],
)
.unwrap();
// Now re-download (UPSERT) - this is the scenario that was broken
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 0, CURRENT_TIMESTAMP)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = excluded.priority,
status = 'pending',
queued_at = CURRENT_TIMESTAMP",
params!["item1", "user1", "/path/to/file.mp3"],
)
.unwrap();
// The old buggy code used last_insert_rowid() which would return 0 or wrong value
// Our fixed code queries by unique constraint
let second_id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params!["item1", "user1"],
|row| row.get(0),
)
.unwrap();
// The ID should be the same as the first insert (UPSERT updated the existing row)
assert_eq!(first_id, second_id, "UPSERT should return the same row ID");
// Verify the status was reset to pending
let status: String = conn
.query_row(
"SELECT status FROM downloads WHERE id = ?1",
params![second_id],
|row| row.get(0),
)
.unwrap();
assert_eq!(
status, "pending",
"Status should be reset to pending after UPSERT"
);
}
#[test]
fn test_query_by_unique_constraint_is_reliable() {
// This test demonstrates that querying by unique constraint is always reliable,
// unlike last_insert_rowid() which has undefined behavior with UPSERT.
//
// SQLite documentation states that last_insert_rowid() behavior is undefined
// when ON CONFLICT triggers an UPDATE instead of INSERT. Some versions return 0,
// others return the existing row ID - it's not consistent.
//
// Our fix: always query by the unique constraint columns (item_id, user_id)
// to get the correct download ID, regardless of whether INSERT or UPDATE occurred.
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// First insert
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 0, CURRENT_TIMESTAMP)",
params!["item1", "user1", "/path/to/file.mp3"],
)
.unwrap();
let first_id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params!["item1", "user1"],
|row| row.get(0),
)
.unwrap();
// UPSERT (triggers UPDATE)
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 0, CURRENT_TIMESTAMP)
ON CONFLICT(item_id, user_id) DO UPDATE SET
status = 'pending'",
params!["item1", "user1", "/path/to/file.mp3"],
)
.unwrap();
// Our approach: query by unique constraint - always works!
let id_after_upsert: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params!["item1", "user1"],
|row| row.get(0),
)
.unwrap();
// This should ALWAYS be the same ID - our approach is reliable
assert_eq!(
first_id, id_after_upsert,
"Query by unique constraint should always return correct ID"
);
}
#[test]
fn test_download_album_returns_correct_ids() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// Insert downloads for multiple items (simulating album download)
let track_ids = vec!["item1", "item2", "item3"];
let mut download_ids = Vec::new();
for track_id in &track_ids {
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 100, CURRENT_TIMESTAMP)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = 100,
status = 'pending'",
params![track_id, "user1", format!("/path/{}.mp3", track_id)],
)
.unwrap();
// Use our fixed approach - query by unique constraint
let download_id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params![track_id, "user1"],
|row| row.get(0),
)
.unwrap();
download_ids.push(download_id);
}
// All IDs should be unique and positive
assert_eq!(download_ids.len(), 3);
for id in &download_ids {
assert!(*id > 0, "Download ID should be positive");
}
// IDs should be unique
let mut sorted_ids = download_ids.clone();
sorted_ids.sort();
sorted_ids.dedup();
assert_eq!(sorted_ids.len(), 3, "All download IDs should be unique");
}
#[test]
fn test_download_album_upsert_returns_correct_ids() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// First download attempt - insert all
let track_ids = vec!["item1", "item2", "item3"];
let mut first_ids = Vec::new();
for track_id in &track_ids {
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 100, CURRENT_TIMESTAMP)",
params![track_id, "user1", format!("/path/{}.mp3", track_id)],
)
.unwrap();
let id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params![track_id, "user1"],
|row| row.get(0),
)
.unwrap();
first_ids.push(id);
}
// Mark all as failed
conn.execute("UPDATE downloads SET status = 'failed'", [])
.unwrap();
// Re-download (UPSERT all)
let mut second_ids = Vec::new();
for track_id in &track_ids {
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at)
VALUES (?1, ?2, ?3, 'pending', 100, CURRENT_TIMESTAMP)
ON CONFLICT(item_id, user_id) DO UPDATE SET
priority = 100,
status = 'pending'",
params![track_id, "user1", format!("/path/{}.mp3", track_id)],
)
.unwrap();
let id: i64 = conn
.query_row(
"SELECT id FROM downloads WHERE item_id = ?1 AND user_id = ?2",
params![track_id, "user1"],
|row| row.get(0),
)
.unwrap();
second_ids.push(id);
}
// IDs should be the same (UPSERT updates existing rows)
assert_eq!(first_ids, second_ids, "UPSERT should preserve original IDs");
}
#[test]
fn test_get_downloads_returns_correct_metadata() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// Insert a download
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, progress, priority)
VALUES (?1, ?2, ?3, 'downloading', 0.5, 10)",
params!["item1", "user1", "/path/to/song.mp3"],
)
.unwrap();
// Query downloads with item metadata
let download: DownloadInfo = conn
.query_row(
"SELECT d.id, d.item_id, d.user_id, d.file_path, d.file_size, d.mime_type, d.status, d.progress,
d.bytes_downloaded, d.queued_at, d.started_at, d.completed_at, d.error_message,
d.retry_count, d.priority,
COALESCE(d.item_name, i.name) as item_name,
COALESCE(d.artist_name, i.artists) as artist_name,
COALESCE(d.album_name, i.album_name) as album_name,
COALESCE(d.series_name, i.series_name) as series_name,
COALESCE(d.season_name, i.season_name) as season_name,
COALESCE(d.episode_number, i.index_number) as episode_number,
COALESCE(d.season_number, i.parent_index_number) as season_number,
d.quality_preset,
COALESCE(d.media_type, 'audio') as media_type,
COALESCE(d.download_source, 'user') as download_source
FROM downloads d
LEFT JOIN items i ON d.item_id = i.id
WHERE d.user_id = ?1",
params!["user1"],
map_download_row,
)
.unwrap();
assert_eq!(download.item_id, "item1");
assert_eq!(download.status, "downloading");
assert!((download.progress - 0.5).abs() < 0.001);
assert_eq!(download.priority, 10);
assert_eq!(download.item_name, Some("Test Song 1".to_string()));
}
#[test]
fn test_download_status_transitions() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
// Insert pending download
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status)
VALUES (?1, ?2, ?3, 'pending')",
params!["item1", "user1", "/path/to/song.mp3"],
)
.unwrap();
let id: i64 = conn.last_insert_rowid();
// Transition: pending -> downloading
conn.execute(
"UPDATE downloads SET status = 'downloading', started_at = CURRENT_TIMESTAMP WHERE id = ?1",
params![id],
)
.unwrap();
let status: String = conn
.query_row(
"SELECT status FROM downloads WHERE id = ?1",
params![id],
|row| row.get(0),
)
.unwrap();
assert_eq!(status, "downloading");
// Transition: downloading -> completed
conn.execute(
"UPDATE downloads SET status = 'completed', progress = 1.0, completed_at = CURRENT_TIMESTAMP WHERE id = ?1",
params![id],
)
.unwrap();
let (status, progress): (String, f64) = conn
.query_row(
"SELECT status, progress FROM downloads WHERE id = ?1",
params![id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert_eq!(status, "completed");
assert!((progress - 1.0).abs() < 0.001);
}
#[test]
fn test_download_progress_updates() {
let db = setup_test_db();
let conn = db.connection();
let conn = conn.lock_safe();
conn.execute(
"INSERT INTO downloads (item_id, user_id, file_path, status, progress, bytes_downloaded, file_size)
VALUES (?1, ?2, ?3, 'downloading', 0.0, 0, 10000000)",
params!["item1", "user1", "/path/to/song.mp3"],
)
.unwrap();
let id: i64 = conn.last_insert_rowid();
// Simulate progress updates
for i in 1..=10 {
let progress = i as f64 / 10.0;
let bytes = i * 1000000;
conn.execute(
"UPDATE downloads SET progress = ?1, bytes_downloaded = ?2 WHERE id = ?3",
params![progress, bytes, id],
)
.unwrap();
let (actual_progress, actual_bytes): (f64, i64) = conn
.query_row(
"SELECT progress, bytes_downloaded FROM downloads WHERE id = ?1",
params![id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.unwrap();
assert!((actual_progress - progress).abs() < 0.001);
assert_eq!(actual_bytes, bytes);
}
}
#[test]
fn test_compute_download_stats_empty() {
let downloads = vec![];
let stats = compute_download_stats(&downloads);
assert_eq!(stats.total, 0);
assert_eq!(stats.active_count, 0);
assert_eq!(stats.queued_count, 0);
assert_eq!(stats.completed_count, 0);
assert_eq!(stats.failed_count, 0);
assert_eq!(stats.paused_count, 0);
}
#[test]
fn test_compute_download_stats_mixed() {
let downloads = vec![
create_test_download(1, "downloading"),
create_test_download(2, "pending"),
create_test_download(3, "downloading"),
create_test_download(4, "completed"),
create_test_download(5, "failed"),
create_test_download(6, "paused"),
];
let stats = compute_download_stats(&downloads);
assert_eq!(stats.total, 6);
assert_eq!(stats.active_count, 2);
assert_eq!(stats.queued_count, 1);
assert_eq!(stats.completed_count, 1);
assert_eq!(stats.failed_count, 1);
assert_eq!(stats.paused_count, 1);
}
#[test]
fn test_compute_download_stats_all_same_status() {
let downloads = vec![
create_test_download(1, "completed"),
create_test_download(2, "completed"),
create_test_download(3, "completed"),
];
let stats = compute_download_stats(&downloads);
assert_eq!(stats.total, 3);
assert_eq!(stats.active_count, 0);
assert_eq!(stats.queued_count, 0);
assert_eq!(stats.completed_count, 3);
assert_eq!(stats.failed_count, 0);
assert_eq!(stats.paused_count, 0);
}
/// Helper to create a test DownloadInfo for stats testing
fn create_test_download(id: i64, status: &str) -> DownloadInfo {
DownloadInfo {
id,
item_id: format!("item{}", id),
user_id: "test_user".to_string(),
file_path: format!("/tmp/download{}", id),
file_size: Some(1000),
mime_type: Some("audio/flac".to_string()),
status: status.to_string(),
progress: 0.0,
bytes_downloaded: 0,
queued_at: "2024-01-01T00:00:00Z".to_string(),
started_at: None,
completed_at: None,
error_message: None,
retry_count: 0,
priority: 0,
item_name: Some(format!("Track {}", id)),
artist_name: None,
album_name: None,
series_name: None,
season_name: None,
episode_number: None,
season_number: None,
quality_preset: None,
media_type: "audio".to_string(),
download_source: "user".to_string(),
}
}
// ===== Album download: track sourcing and album linkage =====
/// A database with just the tables the album-download path touches.
fn album_test_db() -> Arc<crate::storage::db_service::RusqliteService> {
let conn = rusqlite::Connection::open_in_memory().unwrap();
conn.execute_batch(
r#"
CREATE TABLE items (
id TEXT PRIMARY KEY,
server_id TEXT NOT NULL,
parent_id TEXT,
name TEXT NOT NULL,
item_type TEXT NOT NULL,
album_id TEXT,
album_name TEXT,
album_artist TEXT,
artists TEXT,
index_number INTEGER
);
CREATE TABLE downloads (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT NOT NULL,
user_id TEXT NOT NULL,
file_path TEXT NOT NULL,
status TEXT DEFAULT 'pending',
priority INTEGER DEFAULT 0,
progress REAL DEFAULT 0,
queued_at TEXT,
item_name TEXT,
artist_name TEXT,
album_name TEXT,
media_type TEXT,
stream_url TEXT,
target_dir TEXT,
UNIQUE(item_id, user_id)
);
INSERT INTO items (id, server_id, name, item_type)
VALUES ('album1', 'server1', 'The Golden Age', 'MusicAlbum');
"#,
)
.unwrap();
Arc::new(crate::storage::db_service::RusqliteService::new(Arc::new(
Mutex::new(conn),
)))
}
fn album_track(id: &str, name: &str, index: i32) -> AlbumTrack {
AlbumTrack {
id: id.to_string(),
name: name.to_string(),
artist_name: Some("Woodkid".to_string()),
album_name: Some("The Golden Age".to_string()),
index_number: Some(index),
}
}
/// The album-download regression: every track the album actually has must be
/// queued, and each queued track must be linked to its album.
///
/// `download_album` used to take its track list from
/// `items WHERE album_id = ?`. Jellyfin does not return `AlbumId` on every
/// listing endpoint, so tracks cached from those endpoints sit in `items`
/// with a NULL `album_id` — invisible to that query. "Download album" then
/// silently queued only the subset that happened to carry the link, which is
/// the reported "only 4-5 songs downloaded". The same column is what offline
/// browsing joins tracks to their album on (`i.album_id = ?` in
/// `OfflineRepository::get_items`), so even a track that did download stayed
/// invisible under its album offline.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
#[tokio::test]
async fn test_queue_album_tracks_queues_every_track_and_links_it_to_the_album() {
let db = album_test_db();
// The cache holds all three tracks, but only one carries `album_id` —
// exactly the state the bug report's database is in.
for sql in [
"INSERT INTO items (id, server_id, name, item_type, album_id) \
VALUES ('t1', 'server1', 'Run Boy Run', 'Audio', 'album1')",
"INSERT INTO items (id, server_id, name, item_type, album_id) \
VALUES ('t2', 'server1', 'The Great Escape', 'Audio', NULL)",
"INSERT INTO items (id, server_id, name, item_type, album_id) \
VALUES ('t3', 'server1', 'Boat Song', 'Audio', NULL)",
] {
db.execute(Query::new(sql)).await.unwrap();
}
let tracks = vec![
album_track("t1", "Run Boy Run", 1),
album_track("t2", "The Great Escape", 2),
album_track("t3", "Boat Song", 3),
];
let ids = queue_album_tracks(&db, "album1", &tracks, "user1", "albums/album1")
.await
.unwrap();
assert_eq!(
ids.len(),
3,
"every track of the album must get a download row"
);
let queued: i64 = db
.query_one(
Query::new("SELECT COUNT(*) FROM downloads WHERE status = 'pending'"),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(queued, 3);
// Each track is now linked to its album, so the offline album page can
// find it once the download completes.
let linked: i64 = db
.query_one(
Query::new("SELECT COUNT(*) FROM items WHERE album_id = 'album1'"),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(
linked, 3,
"queued tracks must be linked to their album; offline browsing joins on album_id"
);
}
/// The returned ids must line up with the tracks that were passed in. The
/// frontend used to pair `downloadIds[i]` with its own `tracks[i]`, which is
/// only sound if both lists agree — they did not, because the backend
/// ordered by `index_number` over a different set of rows. Resolving URLs in
/// Rust removes the pairing entirely, but the order is still the contract
/// for anything that reads the ids back.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
#[tokio::test]
async fn test_queue_album_tracks_returns_ids_in_track_order() {
let db = album_test_db();
let tracks = vec![
album_track("t1", "Run Boy Run", 1),
album_track("t2", "The Great Escape", 2),
];
let ids = queue_album_tracks(&db, "album1", &tracks, "user1", "albums/album1")
.await
.unwrap();
for (id, track) in ids.iter().zip(tracks.iter()) {
let item_id: String = db
.query_one(
Query::with_params(
"SELECT item_id FROM downloads WHERE id = ?",
vec![QueryParam::Int64(*id)],
),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(&item_id, &track.id, "id {} must be {}'s row", id, track.id);
}
}
/// Re-queueing an album already partly downloaded must not duplicate rows or
/// reset a completed track — it fills in what is missing.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
#[tokio::test]
async fn test_queue_album_tracks_is_idempotent() {
let db = album_test_db();
let tracks = vec![
album_track("t1", "Run Boy Run", 1),
album_track("t2", "The Great Escape", 2),
];
let first = queue_album_tracks(&db, "album1", &tracks, "user1", "albums/album1")
.await
.unwrap();
let second = queue_album_tracks(&db, "album1", &tracks, "user1", "albums/album1")
.await
.unwrap();
assert_eq!(first, second, "the same tracks must map to the same rows");
let rows: i64 = db
.query_one(Query::new("SELECT COUNT(*) FROM downloads"), |row| {
row.get(0)
})
.await
.unwrap();
assert_eq!(rows, 2, "re-queueing must not duplicate download rows");
}
/// Two tracks of one album can share a title — a deluxe edition carrying the
/// album version and a demo of the same song, or the same song on two discs.
/// Naming the file after the title alone gave them one path, so the second
/// download overwrote the first and the album ended up short however many
/// duplicates it had.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-172
#[test]
fn test_album_file_names_are_unique_within_the_album() {
let tracks = vec![
album_track("t1", "Crucified Again", 5),
album_track("t2", "Crucified Again", 5),
album_track("t3", "Get Right", 7),
];
let names = album_file_names(&tracks);
assert_eq!(names.len(), 3);
let unique: std::collections::HashSet<_> = names.iter().collect();
assert_eq!(
unique.len(),
3,
"every track of an album needs its own file: {:?}",
names
);
assert!(names.iter().all(|n| n.ends_with(".mp3")), "{:?}", names);
assert!(
names[2].contains("Get Right"),
"an unambiguous title keeps its name: {}",
names[2]
);
}
/// Path separators in a track title must not escape the album directory.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-172
#[test]
fn test_album_file_names_sanitize_the_title() {
let names = album_file_names(&[album_track("t1", "AC/DC: Live?", 1)]);
assert!(!names[0].contains('/'), "{}", names[0]);
assert!(!names[0].contains(':'), "{}", names[0]);
}
/// The offline fallback reads the catalog directly, not through the
/// availability-gated offline listing: queueing an album while the server is
/// unreachable is a supported flow (the rows resolve on reconnect), and
/// gating it on what is already downloaded would queue only the tracks the
/// device already has.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
#[tokio::test]
async fn test_cached_album_tracks_finds_tracks_by_either_album_link() {
let db = album_test_db();
for sql in [
"INSERT INTO items (id, server_id, name, item_type, album_id, index_number) \
VALUES ('t1', 'server1', 'Run Boy Run', 'Audio', 'album1', 1)",
// Linked by parent_id only — how a track cached from a folder
// listing lands in the catalog.
"INSERT INTO items (id, server_id, name, item_type, parent_id, index_number) \
VALUES ('t2', 'server1', 'The Great Escape', 'Audio', 'album1', 2)",
// A different album's track must not be swept in.
"INSERT INTO items (id, server_id, name, item_type, album_id) \
VALUES ('other', 'server1', 'Iron', 'Audio', 'album2')",
] {
db.execute(Query::new(sql)).await.unwrap();
}
let tracks = cached_album_tracks(&db, "album1").await.unwrap();
let ids: Vec<_> = tracks.iter().map(|t| t.id.as_str()).collect();
assert_eq!(ids, vec!["t1", "t2"]);
}
/// Tracks the cache has never seen still get queued: the row is created and
/// an `items` row is written for it, so the download is both startable and
/// visible offline afterwards.
///
/// TRACES: UR-018, UR-055 | DR-173 | UT-170
#[tokio::test]
async fn test_queue_album_tracks_handles_tracks_absent_from_the_cache() {
let db = album_test_db();
let tracks = vec![album_track("never-cached", "Iron", 1)];
let ids = queue_album_tracks(&db, "album1", &tracks, "user1", "albums/album1")
.await
.unwrap();
assert_eq!(ids.len(), 1);
let (item_type, album_id): (String, Option<String>) = db
.query_one(
Query::new("SELECT item_type, album_id FROM items WHERE id = 'never-cached'"),
|row| Ok((row.get(0)?, row.get(1)?)),
)
.await
.unwrap();
assert_eq!(item_type, "Audio");
assert_eq!(album_id.as_deref(), Some("album1"));
}
}
Reference in New Issue View Git Blame Copy Permalink