Duncan Tourolle
acb7e5f221
All checks were successful
🏗️ Build and Test JellyTau / Run Tests (push) Successful in 4m32s
Traceability Validation / Check Requirement Traces (push) Successful in 22s
Build & Release / Run Tests (push) Successful in 5m21s
🏗️ Build and Test JellyTau / Android Compile Check (push) Successful in 4m25s
Build & Release / Build Linux (push) Successful in 17m31s
Build & Release / Build Android (push) Successful in 22m5s
Build & Release / Create Release (push) Successful in 16s
2211 lines
79 KiB
Rust
2211 lines
79 KiB
Rust
//! Tauri commands for download operations
|
|
|
|
#[cfg(test)]
|
|
use crate::utils::lock::MutexSafe;
|
|
use std::path::PathBuf;
|
|
use std::sync::{Arc, Mutex};
|
|
use tauri::{Manager, State};
|
|
use log::{debug, error, info, warn};
|
|
|
|
use crate::download::{DownloadInfo, DownloadManager};
|
|
use crate::storage::db_service::{DatabaseService, Query, QueryParam};
|
|
use super::{DatabaseWrapper, SmartCacheWrapper};
|
|
|
|
// 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::*;
|
|
|
|
/// Wrapper for DownloadManager to be used as Tauri state
|
|
pub struct DownloadManagerWrapper(pub Mutex<DownloadManager>);
|
|
|
|
/// 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()
|
|
}
|
|
|
|
/// 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;
|
|
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...");
|
|
|
|
// 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)
|
|
}
|
|
|
|
/// Queue an entire album for download
|
|
#[tauri::command]
|
|
#[specta::specta]
|
|
pub async fn download_album(
|
|
db: State<'_, DatabaseWrapper>,
|
|
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())
|
|
};
|
|
|
|
// Get all tracks in the album with metadata
|
|
let tracks_query = Query::with_params(
|
|
"SELECT id, name, artists, album_name FROM items
|
|
WHERE album_id = ? AND item_type = 'Audio'
|
|
ORDER BY index_number",
|
|
vec![QueryParam::String(album_id)],
|
|
);
|
|
|
|
let tracks: Vec<(String, String, Option<String>, Option<String>)> = db_service
|
|
.query_many(tracks_query, |row| {
|
|
Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
|
|
})
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
|
|
let mut download_ids = Vec::new();
|
|
|
|
// Queue each track with album priority (100) and metadata
|
|
for (track_id, track_name, artist_name, album_name) in tracks {
|
|
let file_path = format!("{}/{}.mp3", base_path, sanitize_filename(&track_name));
|
|
|
|
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)
|
|
VALUES (?, ?, ?, 'pending', 100, CURRENT_TIMESTAMP, ?, ?, ?)
|
|
ON CONFLICT(item_id, user_id) DO UPDATE SET
|
|
priority = 100,
|
|
status = 'pending',
|
|
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.clone()),
|
|
QueryParam::String(file_path),
|
|
QueryParam::String(track_name),
|
|
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 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), 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);
|
|
}
|
|
|
|
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<(String, String, Option<String>, Option<i32>, Option<i32>)> = 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]
|
|
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
|
|
#[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 = 'downloading'",
|
|
vec![QueryParam::Int64(download_id)],
|
|
);
|
|
|
|
db_service.execute(query).await.map_err(|e| e.to_string())?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Resume a paused download
|
|
#[tauri::command]
|
|
#[specta::specta]
|
|
pub async fn resume_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 = 'pending' WHERE id = ? AND status = 'paused'",
|
|
vec![QueryParam::Int64(download_id)],
|
|
);
|
|
|
|
db_service.execute(query).await.map_err(|e| e.to_string())?;
|
|
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())?;
|
|
|
|
// 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 partial file if exists
|
|
if let Some(path) = file_path {
|
|
let partial_path = format!("{}.part", path);
|
|
let _ = std::fs::remove_file(&partial_path); // Ignore errors
|
|
}
|
|
|
|
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> {
|
|
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);
|
|
|
|
// 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),
|
|
}
|
|
|
|
// Build target path
|
|
let target_path = PathBuf::from(&target_dir).join(&file_path);
|
|
|
|
// 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> {
|
|
use crate::repository::MediaRepository;
|
|
|
|
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 transcode URL (pure URL builder, no server round-trip).
|
|
let stream_url = repo.as_ref().get_video_download_url(&item_id, &quality, None);
|
|
|
|
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(())
|
|
}
|
|
|
|
/// 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;
|
|
|
|
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
|
|
};
|
|
|
|
// 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(),
|
|
},
|
|
);
|
|
|
|
let target_path = PathBuf::from(&target_dir).join(&file_path);
|
|
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::{DownloadTask, DownloadWorker};
|
|
use crate::download::events::DownloadEvent;
|
|
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);
|
|
};
|
|
|
|
let result = worker.download(&task, on_progress).await;
|
|
|
|
// 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),
|
|
}
|
|
}
|
|
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())
|
|
};
|
|
|
|
// Get file paths for stale downloads (pending/paused/failed)
|
|
let file_query = Query::with_params(
|
|
"SELECT file_path FROM downloads WHERE user_id = ? AND status IN ('pending', 'paused', 'failed')",
|
|
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 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
|
|
for path in file_paths {
|
|
let _ = std::fs::remove_file(&path);
|
|
let _ = std::fs::remove_file(format!("{}.part", path));
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
/// 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");
|
|
}
|
|
|
|
/// 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(),
|
|
}
|
|
}
|
|
}
|