dtourolle/jellytau
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 39 Wiki Activity

Background-audio handoff for video + repository/player refactor

Browse Source 
Hand video playback off to a native audio-only stream when the app is
backgrounded or locked, with no on-device video decode (UR-040). Adds
player_enter/exit_background_audio commands, an audio-only stream URL
for video items across the repository layer, and the frontend handoff
state machine wired into VideoPlayer. Includes accompanying
repository/offline/player refactoring and regenerates the traceability
matrix.
This commit is contained in:
dtourolle
2026-07-22 21:52:07 +02:00
parent 4e6ab017d4
commit 3fbf6afdbc
72 changed files with 6728 additions and 2338 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/download/mod.rs
+246 -67
View File
@@ -2,14 +2,14 @@
#[cfg(test)]
use crate::utils::lock::MutexSafe;
use log::{debug, error, info, warn};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use tauri::{Manager, State};
use log::{debug, error, info, warn};
use super::{DatabaseWrapper, SmartCacheWrapper};
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).
@@ -111,7 +111,13 @@ pub async fn download_item_and_start(
request: DownloadItemAndStartRequest,
) -> Result<i64, String> {
let DownloadItemAndStartRequest {
item_id, user_id, stream_url, target_dir, item_name, artist_name, album_name,
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));
@@ -132,7 +138,8 @@ pub async fn download_item_and_start(
album_name,
expected_size: None,
},
).await?;
)
.await?;
// Start the download immediately
start_download(
@@ -142,7 +149,8 @@ pub async fn download_item_and_start(
download_id,
stream_url,
target_dir,
).await?;
)
.await?;
Ok(download_id)
}
@@ -156,7 +164,15 @@ pub async fn download_item(
request: DownloadItemRequest,
) -> Result<i64, String> {
let DownloadItemRequest {
item_id, user_id, file_path, mime_type, priority, item_name, artist_name, album_name, expected_size,
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())?;
@@ -172,18 +188,24 @@ pub async fn download_item(
};
// Check if we have space
let can_download = cache_arc.can_download_async(&db_service, &user_id, size as u64).await;
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 {
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);
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
@@ -220,7 +242,10 @@ pub async fn download_item(
],
);
db_service.execute(insert_query).await.map_err(|e| e.to_string())?;
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
@@ -291,12 +316,18 @@ pub async fn download_album(
],
);
db_service.execute(insert_query).await.map_err(|e| e.to_string())?;
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())],
vec![
QueryParam::String(track_id),
QueryParam::String(user_id.clone()),
],
);
let download_id: i64 = db_service
@@ -317,8 +348,17 @@ pub async fn download_video(
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,
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())?;
@@ -358,7 +398,10 @@ pub async fn download_video(
],
);
db_service.execute(insert_query).await.map_err(|e| e.to_string())?;
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(
@@ -403,7 +446,13 @@ pub async fn download_series(
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)?))
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
))
})
.await
.map_err(|e| e.to_string())?;
@@ -413,7 +462,9 @@ pub async fn download_series(
// 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() {
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
@@ -425,7 +476,12 @@ pub async fn download_series(
episode_num,
sanitize_filename(&episode_name)
);
let file_path = format!("{}/{}/{}", base_path, sanitize_filename(&series_name), file_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,
@@ -450,17 +506,29 @@ pub async fn download_series(
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),
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())?;
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())],
vec![
QueryParam::String(episode_id),
QueryParam::String(user_id.clone()),
],
);
let download_id: i64 = db_service
@@ -471,7 +539,10 @@ pub async fn download_series(
download_ids.push(download_id);
}
info!("[download_series] Queued {} episodes for series '{}'", total_episodes, series_name);
info!(
"[download_series] Queued {} episodes for series '{}'",
total_episodes, series_name
);
Ok(download_ids)
}
@@ -524,7 +595,12 @@ pub async fn download_season(
episode_num,
sanitize_filename(&episode_name)
);
let file_path = format!("{}/{}/{}", base_path, sanitize_filename(&series_name), file_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,
@@ -550,11 +626,17 @@ pub async fn download_season(
],
);
db_service.execute(insert_query).await.map_err(|e| e.to_string())?;
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())],
vec![
QueryParam::String(episode_id),
QueryParam::String(user_id.clone()),
],
);
let download_id: i64 = db_service
@@ -565,11 +647,15 @@ pub async fn download_season(
download_ids.push(download_id);
}
info!("[download_season] Queued {} episodes for {} - {}", download_ids.len(), series_name, season_name);
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 {
@@ -674,7 +760,10 @@ pub async fn get_downloads(
/// Pause a download
#[tauri::command]
#[specta::specta]
pub async fn pause_download(db: State<'_, DatabaseWrapper>, download_id: i64) -> Result<(), String> {
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())
@@ -692,7 +781,10 @@ pub async fn pause_download(db: State<'_, DatabaseWrapper>, download_id: i64) ->
/// Resume a paused download
#[tauri::command]
#[specta::specta]
pub async fn resume_download(db: State<'_, DatabaseWrapper>, download_id: i64) -> Result<(), String> {
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())
@@ -738,13 +830,20 @@ pub async fn cancel_download(
vec![QueryParam::Int64(download_id)],
);
db_service.execute(delete_query).await.map_err(|e| e.to_string())?;
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());
info!(
"Cancelled download {}. Active downloads: {}",
download_id,
manager.active_count()
);
}
// Delete partial file if exists
@@ -800,7 +899,10 @@ pub async fn mark_download_failed(
let query = Query::with_params(
"UPDATE downloads SET status = 'failed', error_message = ? WHERE id = ?",
vec![QueryParam::String(error_message), QueryParam::Int64(download_id)],
vec![
QueryParam::String(error_message),
QueryParam::Int64(download_id),
],
);
db_service.execute(query).await.map_err(|e| e.to_string())?;
@@ -834,7 +936,10 @@ pub async fn start_download(
})?;
if !manager.can_start_download() {
warn!("Cannot start download: maximum concurrent downloads ({}) reached", manager.max_concurrent());
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.",
@@ -845,12 +950,19 @@ pub async fn start_download(
// 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);
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());
info!(
"Download {} registered. Active downloads: {}/{}",
download_id,
manager.active_count(),
manager.max_concurrent()
);
}
// Get download info from DB
@@ -868,21 +980,23 @@ pub async fn start_download(
);
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)?)))
.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);
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 head_response = reqwest::Client::new().head(&stream_url).send().await;
let file_size_from_server = match head_response {
Ok(response) => {
@@ -893,7 +1007,11 @@ pub async fn start_download(
.and_then(|v| v.parse::<i64>().ok());
if let Some(size) = size {
debug!(" Got file size from server: {} bytes ({} MB)", size, size / 1024 / 1024);
debug!(
" Got file size from server: {} bytes ({} MB)",
size,
size / 1024 / 1024
);
} else {
warn!(" Server didn't provide Content-Length header");
}
@@ -929,7 +1047,10 @@ pub async fn start_download(
)
};
db_service.execute(update_query).await.map_err(|e| e.to_string())?;
db_service
.execute(update_query)
.await
.map_err(|e| e.to_string())?;
// Emit started event
let started_event = DownloadEvent::Started {
@@ -937,7 +1058,10 @@ pub async fn start_download(
item_id: item_id.clone(),
};
debug!("Emitting download-event: {:?}", started_event);
debug!(" Serialized: {}", serde_json::to_string(&started_event).unwrap_or_default());
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),
@@ -998,7 +1122,10 @@ pub async fn enqueue_download(
QueryParam::Int64(download_id),
],
);
db_service.execute(update_query).await.map_err(|e| e.to_string())?;
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 = {
@@ -1056,7 +1183,9 @@ pub async fn enqueue_video_downloads(
};
// 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 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 = ?",
@@ -1067,7 +1196,10 @@ pub async fn enqueue_video_downloads(
],
);
if let Err(e) = db_service.execute(update_query).await {
warn!("[enqueue_video] Failed to persist URL for download {}: {}", download_id, e);
warn!(
"[enqueue_video] Failed to persist URL for download {}: {}",
download_id, e
);
}
}
@@ -1137,7 +1269,13 @@ pub(crate) async fn pump_download_queue(
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)?))
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
))
})
.await
{
@@ -1189,7 +1327,10 @@ pub(crate) async fn pump_download_queue(
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);
error!(
"[pump] Failed to mark download {} downloading: {}",
download_id, e
);
if let Ok(mut a) = active_downloads.lock() {
a.remove(&download_id);
}
@@ -1228,8 +1369,8 @@ fn spawn_download_worker(
target_path: std::path::PathBuf,
active_downloads: Arc<Mutex<std::collections::HashSet<i64>>>,
) {
use crate::download::{DownloadTask, DownloadWorker};
use crate::download::events::DownloadEvent;
use crate::download::{DownloadTask, DownloadWorker};
use tauri::Emitter;
let task = DownloadTask {
@@ -1265,7 +1406,11 @@ fn spawn_download_worker(
// 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());
debug!(
" Unregistered download {}. Active downloads: {}",
download_id,
active.len()
);
}
// The pump runs downloads in the background, so the terminal status MUST
@@ -1281,7 +1426,10 @@ fn spawn_download_worker(
let database = match db.0.lock() {
Ok(d) => d,
Err(e) => {
error!("[pump] Failed to lock database after download {}: {}", download_id, e);
error!(
"[pump] Failed to lock database after download {}: {}",
download_id, e
);
return;
}
};
@@ -1290,7 +1438,10 @@ fn spawn_download_worker(
match result {
Ok(res) => {
info!("Download completed successfully: {} bytes", res.bytes_downloaded);
info!(
"Download completed successfully: {} bytes",
res.bytes_downloaded
);
let file_path = target_path.to_string_lossy().to_string();
let update = Query::with_params(
@@ -1305,7 +1456,10 @@ fn spawn_download_worker(
],
);
if let Err(e) = db_service.execute(update).await {
error!("[pump] Failed to persist completed status for download {}: {}", download_id, e);
error!(
"[pump] Failed to persist completed status for download {}: {}",
download_id, e
);
}
let completed_event = DownloadEvent::Completed {
@@ -1329,7 +1483,10 @@ fn spawn_download_worker(
],
);
if let Err(db_err) = db_service.execute(update).await {
error!("[pump] Failed to persist failed status for download {}: {}", download_id, db_err);
error!(
"[pump] Failed to persist failed status for download {}: {}",
download_id, db_err
);
}
let failed_event = DownloadEvent::Failed {
@@ -1352,7 +1509,10 @@ fn spawn_download_worker(
/// Delete a completed download
#[tauri::command]
#[specta::specta]
pub async fn delete_download(db: State<'_, DatabaseWrapper>, download_id: i64) -> Result<(), String> {
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())
@@ -1376,7 +1536,10 @@ pub async fn delete_download(db: State<'_, DatabaseWrapper>, download_id: i64) -
vec![QueryParam::Int64(download_id)],
);
db_service.execute(delete_query).await.map_err(|e| e.to_string())?;
db_service
.execute(delete_query)
.await
.map_err(|e| e.to_string())?;
// Delete actual file if exists
if let Some(path) = file_path {
@@ -1413,8 +1576,12 @@ fn map_download_row(row: &rusqlite::Row) -> rusqlite::Result<DownloadInfo> {
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()),
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()),
})
}
@@ -1500,7 +1667,10 @@ pub async fn get_download_storage_stats(
/// 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> {
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())
@@ -1598,7 +1768,10 @@ pub async fn delete_album_downloads(
"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())],
vec![
QueryParam::String(user_id.clone()),
QueryParam::String(album_id.clone()),
],
);
let file_paths: Vec<String> = db_service
@@ -1665,7 +1838,6 @@ pub async fn set_max_concurrent_downloads(
Ok(())
}
// TRACES: UR-011, UR-018 | DR-015, DR-018 | UT-042, UT-043
#[cfg(test)]
mod tests {
@@ -1834,7 +2006,10 @@ mod tests {
)
.unwrap();
assert_eq!(status, "pending", "Status should be reset to pending after UPSERT");
assert_eq!(
status, "pending",
"Status should be reset to pending after UPSERT"
);
}
#[test]
@@ -2069,7 +2244,11 @@ mod tests {
.unwrap();
let status: String = conn
.query_row("SELECT status FROM downloads WHERE id = ?1", params![id], |row| row.get(0))
.query_row(
"SELECT status FROM downloads WHERE id = ?1",
params![id],
|row| row.get(0),
)
.unwrap();
assert_eq!(status, "downloading");
src-tauri/src/commands/download/pinning.rs
+6 -1
View File
@@ -1,4 +1,6 @@
//! Pinning commands - protect an item's cached metadata from cache clearing.
//!
//! TRACES: UR-044 | DR-056
use std::sync::Arc;
use tauri::State;
@@ -45,7 +47,10 @@ pub async fn unpin_item(db: State<'_, DatabaseWrapper>, item_id: String) -> Resu
/// Check if an item is pinned
#[tauri::command]
#[specta::specta]
pub async fn is_item_pinned(db: State<'_, DatabaseWrapper>, item_id: String) -> Result<bool, String> {
pub async fn is_item_pinned(
db: State<'_, DatabaseWrapper>,
item_id: String,
) -> Result<bool, String> {
let db_service = {
let database = db.0.lock().map_err(|e| e.to_string())?;
Arc::new(database.service())
src-tauri/src/commands/download/smart_cache.rs
+3 -1
View File
@@ -1,7 +1,9 @@
//! Smart-cache statistics/config and album recommendation commands.
//!
//! TRACES: UR-045 | DR-057
use std::sync::Arc;
use log::info;
use std::sync::Arc;
use tauri::State;
use crate::commands::{DatabaseWrapper, SmartCacheWrapper};