dtourolle
ac4fccd499
Work from a parallel session in the same working tree, committed here so the branch is not left half-written. Attribution note: authored in a concurrent Claude session, not by the author of the preceding commit. - DR-171: a downloaded video keeps audio the device can actually decode. `original` quality asked for a straight copy, so an E-AC-3/AC-3/DTS/TrueHD track came down untouched and the webview had nothing to play it with. - `get_video_download_url` gains the media source, so the URL is built against the source actually chosen rather than the item's default. - Device profile and repository plumbing updated to match. Verified green as a whole: 656 Rust tests, 945 frontend tests, svelte-check clean.
1053 lines
39 KiB
Rust
1053 lines
39 KiB
Rust
//! Tauri commands for the offline "browse & queue" feature.
|
|
//!
|
|
//! TRACES: UR-002, UR-007, UR-024 | JA-004, JA-016 | DR-012, DR-027
|
|
//!
|
|
//! Two backend pieces support browsing the full server catalog while offline
|
|
//! and queueing downloads that fire on reconnect:
|
|
//!
|
|
//! - [`sync_full_catalog`] walks every library while online and persists all
|
|
//! items to the offline cache so the whole catalog is browsable (greyed out)
|
|
//! offline. It reuses [`HybridRepository::cache_items_from_server`], which in
|
|
//! turn reuses `OfflineRepository::save_to_cache` (sets `synced_at`, which is
|
|
//! what `get_items` branch 3 serves offline).
|
|
//! - [`resume_queued_downloads`] resolves and pumps the `pending` download rows
|
|
//! that were queued offline (they have `stream_url IS NULL`), mirroring the
|
|
//! heal-and-pump pattern in `player_preload_upcoming`.
|
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use log::{info, warn};
|
|
use tauri::{Emitter, Manager, State};
|
|
|
|
use crate::commands::download::{pump_download_queue, DownloadManagerWrapper};
|
|
use crate::commands::repository::RepositoryManagerWrapper;
|
|
use crate::commands::storage::DatabaseWrapper;
|
|
use crate::repository::types::GetItemsOptions;
|
|
use crate::storage::db_service::{DatabaseService, Query, QueryParam};
|
|
|
|
/// app_settings key holding the RFC-3339 timestamp of the last successful
|
|
/// full-catalog sync.
|
|
const LAST_CATALOG_SYNC_KEY: &str = "last_catalog_sync";
|
|
|
|
/// How long an index stays fresh before a re-index is due.
|
|
///
|
|
/// This lives in Rust rather than being a frontend constant because it decides
|
|
/// *whether the local cache is authoritative* — the same class of decision as
|
|
/// `include_catalog_browse`, and squarely the "sync policy" the spec review
|
|
/// checklist keeps out of the presentation layer. If it later becomes
|
|
/// user-configurable it stays a Rust-owned setting edited through a command.
|
|
const CATALOG_INDEX_TTL: Duration = Duration::from_secs(6 * 60 * 60);
|
|
|
|
/// How often the scheduler wakes to *check* staleness. Far shorter than the TTL
|
|
/// because a tick is nearly free — one indexed `app_settings` lookup — and it is
|
|
/// what makes the indexer responsive to events it cannot subscribe to: signing
|
|
/// in, and coming back online. The TTL, not the tick, decides whether a crawl
|
|
/// actually happens.
|
|
const CATALOG_INDEX_TICK: Duration = Duration::from_secs(5 * 60);
|
|
|
|
/// Delay before the first staleness check, to let sign-in complete and the
|
|
/// repository be registered. Without it the first check runs against an empty
|
|
/// repository manager and a fresh install would sit unindexed until the next
|
|
/// tick.
|
|
const CATALOG_INDEX_FIRST_CHECK: Duration = Duration::from_secs(15);
|
|
|
|
/// Kebab-case, per the project's event convention.
|
|
pub const CATALOG_INDEX_EVENT: &str = "catalog-index-event";
|
|
|
|
/// Guards against two passes running at once. Replaces the frontend's
|
|
/// `syncInProgress` boolean in `offlineCatalog.ts`, which could not see a pass
|
|
/// started by the scheduler.
|
|
static INDEX_IN_PROGRESS: AtomicBool = AtomicBool::new(false);
|
|
|
|
/// Clears [`INDEX_IN_PROGRESS`] however the pass leaves — including on the `?`
|
|
/// early return when `get_libraries` fails, which a plain store at the end of
|
|
/// the function would leak.
|
|
struct IndexPassGuard;
|
|
|
|
impl Drop for IndexPassGuard {
|
|
fn drop(&mut self) {
|
|
INDEX_IN_PROGRESS.store(false, Ordering::SeqCst);
|
|
}
|
|
}
|
|
|
|
/// Progress of a background index pass, for the staleness hint in the UI.
|
|
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct CatalogIndexEvent {
|
|
/// `started` | `finished` | `failed`
|
|
pub state: String,
|
|
pub items_cached: usize,
|
|
pub items_pruned: usize,
|
|
pub libraries_failed: usize,
|
|
/// Present on `failed`.
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
/// Item types worth caching for offline browsing: containers the library
|
|
/// landing pages render plus the playable leaves users queue for download.
|
|
/// `MusicArtist` and `Playlist` are here because search groups results by them
|
|
/// (UR-060's Artists group). Without them in the crawl, the local index can
|
|
/// never answer an artist query and those groups can only ever be filled by the
|
|
/// server leg. Keep this in step with what `prune_stale_catalog` is allowed to
|
|
/// sweep — the crawl is only authoritative for the types it asks for.
|
|
///
|
|
/// TRACES: UR-065, UR-060 | DR-111
|
|
const CATALOG_ITEM_TYPES: &[&str] = &[
|
|
"MusicAlbum",
|
|
"MusicArtist",
|
|
"Movie",
|
|
"Series",
|
|
"Season",
|
|
"Episode",
|
|
"Audio",
|
|
"BoxSet",
|
|
"Playlist",
|
|
];
|
|
|
|
/// Jellyfin item types whose download is a *video* stream rather than an audio
|
|
/// one. The download queue stores an opaque `media_type` ('audio'/'video'); this
|
|
/// is where the taxonomy that produces it lives, so the frontend never has to
|
|
/// know which item types are video.
|
|
///
|
|
/// TRACES: UR-071 | DR-135
|
|
const VIDEO_ITEM_TYPES: &[&str] = &["Movie", "Episode", "Video", "MusicVideo"];
|
|
|
|
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct CatalogSyncResult {
|
|
/// Total items persisted to the offline cache across all libraries.
|
|
pub items_cached: usize,
|
|
/// Libraries that failed to sync (e.g. server hiccup); best-effort.
|
|
pub libraries_failed: usize,
|
|
/// Entries removed because the server no longer has them. Always 0 when any
|
|
/// library failed, since a partial crawl cannot prove an item is gone.
|
|
pub items_pruned: usize,
|
|
}
|
|
|
|
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct CatalogSyncStatus {
|
|
/// RFC-3339 timestamp of the last successful sync, if any.
|
|
pub last_synced_at: Option<String>,
|
|
}
|
|
|
|
/// Walk every library on the server and persist all items to the offline cache
|
|
/// so the full catalog is browsable offline (greyed out when not downloaded).
|
|
///
|
|
/// Best-effort: a library that fails to fetch is counted and skipped rather than
|
|
/// aborting the whole sync. Runs libraries sequentially to avoid hammering the
|
|
/// server. Uses `Recursive=true` so a single request per library returns the
|
|
/// containers and their playable children.
|
|
#[tauri::command]
|
|
#[specta::specta]
|
|
pub async fn sync_full_catalog(
|
|
repository: State<'_, RepositoryManagerWrapper>,
|
|
db: State<'_, DatabaseWrapper>,
|
|
handle: String,
|
|
) -> Result<CatalogSyncResult, String> {
|
|
let repo = repository.0.get(&handle).ok_or("Repository not found")?;
|
|
|
|
let db_service = {
|
|
let database = db.0.lock().map_err(|e| e.to_string())?;
|
|
Arc::new(database.service())
|
|
};
|
|
|
|
run_index_pass(repo, db_service).await
|
|
}
|
|
|
|
/// One full-catalog indexing pass, shared by the [`sync_full_catalog`] command
|
|
/// and the background scheduler (DR-109) so there is exactly one implementation
|
|
/// and one concurrency guard.
|
|
///
|
|
/// TRACES: UR-065 | DR-109, DR-110
|
|
pub(crate) async fn run_index_pass(
|
|
repo: Arc<crate::repository::HybridRepository>,
|
|
db_service: Arc<crate::storage::db_service::RusqliteService>,
|
|
) -> Result<CatalogSyncResult, String> {
|
|
use crate::repository::MediaRepository;
|
|
|
|
// One pass at a time. The command and the scheduler can both land here, and
|
|
// two concurrent crawls would double the server load and race on the sweep.
|
|
if INDEX_IN_PROGRESS.swap(true, Ordering::SeqCst) {
|
|
return Err("A catalog index pass is already running".to_string());
|
|
}
|
|
let _guard = IndexPassGuard;
|
|
|
|
let libraries = repo.get_libraries().await.map_err(|e| e.to_string())?;
|
|
info!(
|
|
"[Catalog] Full sync starting across {} libraries",
|
|
libraries.len()
|
|
);
|
|
|
|
let include_types: Vec<String> = CATALOG_ITEM_TYPES.iter().map(|s| s.to_string()).collect();
|
|
|
|
// Taken before the crawl: every row the crawl writes gets a `synced_at`
|
|
// newer than this, so anything still older afterwards is gone server-side.
|
|
let pass_started_at = chrono::Utc::now().to_rfc3339();
|
|
|
|
let mut items_cached = 0usize;
|
|
let mut libraries_failed = 0usize;
|
|
|
|
for library in &libraries {
|
|
let opts = GetItemsOptions {
|
|
recursive: Some(true),
|
|
include_item_types: Some(include_types.clone()),
|
|
limit: Some(100_000),
|
|
..Default::default()
|
|
};
|
|
|
|
match repo.cache_items_from_server(&library.id, Some(opts)).await {
|
|
Ok(items) => {
|
|
info!(
|
|
"[Catalog] Cached {} items from library '{}'",
|
|
items.len(),
|
|
library.name
|
|
);
|
|
items_cached += items.len();
|
|
}
|
|
Err(e) => {
|
|
warn!(
|
|
"[Catalog] Failed to sync library '{}': {:?}",
|
|
library.name, e
|
|
);
|
|
libraries_failed += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Propagate server-side deletions — but only after a *complete* crawl.
|
|
// `sync_full_catalog` is best-effort per library, and `items.parent_id` is
|
|
// ON DELETE CASCADE, so sweeping when a library failed to fetch could
|
|
// cascade a whole series away because one request timed out.
|
|
let mut items_pruned = 0usize;
|
|
if libraries_failed == 0 && !libraries.is_empty() {
|
|
match repo
|
|
.prune_stale_catalog(&pass_started_at, &include_types)
|
|
.await
|
|
{
|
|
Ok(removed) => {
|
|
items_pruned = removed;
|
|
if removed > 0 {
|
|
info!(
|
|
"[Catalog] Pruned {} entries no longer on the server",
|
|
removed
|
|
);
|
|
}
|
|
}
|
|
Err(e) => warn!("[Catalog] Prune of stale catalog entries failed: {:?}", e),
|
|
}
|
|
} else if libraries_failed > 0 {
|
|
info!(
|
|
"[Catalog] Skipping stale-entry prune: {} librar{} failed to sync, so the crawl is not authoritative",
|
|
libraries_failed,
|
|
if libraries_failed == 1 { "y" } else { "ies" }
|
|
);
|
|
}
|
|
|
|
// Record the sync time so callers can skip re-syncing too eagerly.
|
|
let now = chrono::Utc::now().to_rfc3339();
|
|
let upsert = Query::with_params(
|
|
"INSERT INTO app_settings (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP",
|
|
vec![
|
|
QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string()),
|
|
QueryParam::String(now),
|
|
],
|
|
);
|
|
if let Err(e) = db_service.execute(upsert).await {
|
|
warn!("[Catalog] Failed to persist last-sync timestamp: {}", e);
|
|
}
|
|
|
|
info!(
|
|
"[Catalog] Full sync complete: {} items cached, {} pruned, {} libraries failed",
|
|
items_cached, items_pruned, libraries_failed
|
|
);
|
|
|
|
Ok(CatalogSyncResult {
|
|
items_cached,
|
|
libraries_failed,
|
|
items_pruned,
|
|
})
|
|
}
|
|
|
|
/// Whether an index pass is due, given when one last completed.
|
|
///
|
|
/// Pure so the policy is unit-testable without a clock, a server, or a database.
|
|
/// `None` (never indexed) and an unparseable stored value both mean "due" — a
|
|
/// corrupt timestamp should trigger a re-index, not silently freeze the catalog.
|
|
///
|
|
/// TRACES: UR-065 | DR-109 | UT-115
|
|
pub(crate) fn index_is_due(
|
|
last_synced_at: Option<&str>,
|
|
now: chrono::DateTime<chrono::Utc>,
|
|
ttl: Duration,
|
|
) -> bool {
|
|
let Some(raw) = last_synced_at else {
|
|
return true;
|
|
};
|
|
let Ok(last) = chrono::DateTime::parse_from_rfc3339(raw) else {
|
|
return true;
|
|
};
|
|
now.signed_duration_since(last.with_timezone(&chrono::Utc))
|
|
.to_std()
|
|
.map(|elapsed| elapsed >= ttl)
|
|
// Negative elapsed => the stored stamp is in the future (clock skew).
|
|
// Not due; a future stamp will age into due-ness on its own.
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
/// Read the last-sync timestamp straight from `app_settings`.
|
|
async fn read_last_sync(
|
|
db_service: &Arc<crate::storage::db_service::RusqliteService>,
|
|
) -> Option<String> {
|
|
db_service
|
|
.query_optional(
|
|
Query::with_params(
|
|
"SELECT value FROM app_settings WHERE key = ?",
|
|
vec![QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string())],
|
|
),
|
|
|row| row.get(0),
|
|
)
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
}
|
|
|
|
/// Start the background catalog indexer.
|
|
///
|
|
/// Replaces the frontend's startup-only `syncCatalog()` call: index freshness is
|
|
/// sync policy and belongs in Rust (see the layer assignment in
|
|
/// docs/specs/catalog-index-search.md). Ticks every [`CATALOG_INDEX_TICK`] and
|
|
/// runs a pass when a repository exists, the server is reachable, and the index
|
|
/// is older than [`CATALOG_INDEX_TTL`].
|
|
///
|
|
/// TRACES: UR-065 | DR-109, IR-030
|
|
pub fn spawn_catalog_indexer(app: tauri::AppHandle) {
|
|
tauri::async_runtime::spawn(async move {
|
|
// Check shortly after launch, then on every tick — not tick-then-check,
|
|
// which would leave a fresh install unindexed for a full tick.
|
|
tokio::time::sleep(CATALOG_INDEX_FIRST_CHECK).await;
|
|
|
|
loop {
|
|
if let Err(e) = maybe_run_scheduled_pass(&app).await {
|
|
// Never fatal — a failed pass leaves the existing index in place
|
|
// and we retry on the next tick.
|
|
warn!("[Catalog] Scheduled index pass skipped: {}", e);
|
|
}
|
|
|
|
tokio::time::sleep(CATALOG_INDEX_TICK).await;
|
|
}
|
|
});
|
|
}
|
|
|
|
/// One scheduler tick: check the preconditions, then index if due.
|
|
async fn maybe_run_scheduled_pass(app: &tauri::AppHandle) -> Result<(), String> {
|
|
if INDEX_IN_PROGRESS.load(Ordering::SeqCst) {
|
|
return Ok(());
|
|
}
|
|
|
|
let db_service = {
|
|
let db = app.state::<DatabaseWrapper>();
|
|
let database = db.0.lock().map_err(|e| e.to_string())?;
|
|
Arc::new(database.service())
|
|
};
|
|
|
|
if !index_is_due(
|
|
read_last_sync(&db_service).await.as_deref(),
|
|
chrono::Utc::now(),
|
|
CATALOG_INDEX_TTL,
|
|
) {
|
|
return Ok(());
|
|
}
|
|
|
|
// Offline: leave the index alone. The crawl would fail every library and,
|
|
// more importantly, a partial crawl must never reach the deletion sweep.
|
|
{
|
|
let monitor = app.state::<crate::commands::connectivity::ConnectivityMonitorWrapper>();
|
|
let monitor = monitor.0.lock().await;
|
|
if !monitor.get_status().await.is_server_reachable {
|
|
return Ok(());
|
|
}
|
|
}
|
|
|
|
let repo = {
|
|
let manager = app.state::<RepositoryManagerWrapper>();
|
|
let handles = manager.0.handles();
|
|
let Some(handle) = handles.first() else {
|
|
// Not signed in yet.
|
|
return Ok(());
|
|
};
|
|
manager.0.get(handle).ok_or("Repository not found")?
|
|
};
|
|
|
|
info!("[Catalog] Index is stale; starting a scheduled pass");
|
|
let _ = app.emit(
|
|
CATALOG_INDEX_EVENT,
|
|
CatalogIndexEvent {
|
|
state: "started".to_string(),
|
|
items_cached: 0,
|
|
items_pruned: 0,
|
|
libraries_failed: 0,
|
|
error: None,
|
|
},
|
|
);
|
|
|
|
match run_index_pass(repo, db_service).await {
|
|
Ok(result) => {
|
|
let _ = app.emit(
|
|
CATALOG_INDEX_EVENT,
|
|
CatalogIndexEvent {
|
|
state: "finished".to_string(),
|
|
items_cached: result.items_cached,
|
|
items_pruned: result.items_pruned,
|
|
libraries_failed: result.libraries_failed,
|
|
error: None,
|
|
},
|
|
);
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
let _ = app.emit(
|
|
CATALOG_INDEX_EVENT,
|
|
CatalogIndexEvent {
|
|
state: "failed".to_string(),
|
|
items_cached: 0,
|
|
items_pruned: 0,
|
|
libraries_failed: 0,
|
|
error: Some(e.clone()),
|
|
},
|
|
);
|
|
Err(e)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Report the last-synced timestamp so the UI can show a hint / decide whether
|
|
/// to trigger a fresh sync.
|
|
#[tauri::command]
|
|
#[specta::specta]
|
|
pub async fn catalog_sync_status(
|
|
db: State<'_, DatabaseWrapper>,
|
|
) -> Result<CatalogSyncStatus, String> {
|
|
let db_service = {
|
|
let database = db.0.lock().map_err(|e| e.to_string())?;
|
|
Arc::new(database.service())
|
|
};
|
|
|
|
let query = Query::with_params(
|
|
"SELECT value FROM app_settings WHERE key = ?",
|
|
vec![QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string())],
|
|
);
|
|
let last_synced_at: Option<String> = db_service
|
|
.query_optional(query, |row| row.get(0))
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
|
|
Ok(CatalogSyncStatus { last_synced_at })
|
|
}
|
|
|
|
/// Control whether offline library queries reveal the full synced catalog
|
|
/// (greyed-out, non-downloaded media) or only downloaded/local media.
|
|
///
|
|
/// The frontend calls this from the "Show all server media" toggle: pass `true`
|
|
/// when online, or when offline with the toggle on; pass `false` when offline
|
|
/// with the toggle off so library pages show downloaded media only. Fixes the
|
|
/// bug where offline library pages showed every server item regardless of the
|
|
/// toggle.
|
|
#[tauri::command]
|
|
#[specta::specta]
|
|
pub fn set_show_server_catalog(show: bool) {
|
|
crate::repository::offline::set_include_catalog_browse(show);
|
|
}
|
|
|
|
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct ResumeQueuedResult {
|
|
/// Rows whose stream URL was resolved and are now pump-eligible.
|
|
pub resolved: usize,
|
|
/// Rows that couldn't be resolved (item metadata / URL lookup failed).
|
|
pub failed: usize,
|
|
}
|
|
|
|
/// Requeue video downloads that were fetched as audio.
|
|
///
|
|
/// Before [`resolve_pending_download_urls`] consulted the item's type, a row
|
|
/// with no `media_type` — which is every row queued from a media card, since
|
|
/// `download_item` does not record one — resolved against
|
|
/// `get_audio_stream_url`. A movie queued that way completed with an audio-only
|
|
/// transcode on disk, so playing it offline could only ever fail. Those rows are
|
|
/// identifiable after the fact (no `media_type`, but a video item), so reset them
|
|
/// to pending with no URL and let the resolver fetch the real video.
|
|
///
|
|
/// Rows carrying an explicit `media_type` were resolved correctly and are left
|
|
/// alone, as are genuine audio downloads.
|
|
///
|
|
/// Returns the number of rows requeued.
|
|
///
|
|
/// TRACES: UR-071 | DR-136 | UT-126
|
|
pub(crate) async fn requeue_mistyped_video_downloads(
|
|
db_service: &Arc<crate::storage::db_service::RusqliteService>,
|
|
) -> Result<usize, String> {
|
|
let video_types = VIDEO_ITEM_TYPES
|
|
.iter()
|
|
.map(|t| format!("'{t}'"))
|
|
.collect::<Vec<_>>()
|
|
.join(", ");
|
|
|
|
let query = Query::new(&format!(
|
|
"UPDATE downloads
|
|
SET status = 'pending', stream_url = NULL, progress = 0,
|
|
bytes_downloaded = 0, started_at = NULL, completed_at = NULL
|
|
WHERE media_type IS NULL
|
|
AND status = 'completed'
|
|
AND item_id IN (SELECT id FROM items WHERE item_type IN ({video_types}))"
|
|
));
|
|
|
|
let n = db_service.execute(query).await.map_err(|e| e.to_string())? as usize;
|
|
|
|
if n > 0 {
|
|
info!(
|
|
"[Catalog] Requeued {} video download(s) that were fetched as audio",
|
|
n
|
|
);
|
|
}
|
|
Ok(n)
|
|
}
|
|
|
|
/// Core of [`resume_queued_downloads`], factored out for testing: select every
|
|
/// `pending`/`stream_url IS NULL` row, resolve each via `resolve` (returning
|
|
/// `None` leaves the row pending), and heal the row so the pump can start it.
|
|
/// The `resolve` closure receives `(item_id, media_type, quality_preset)`.
|
|
pub(crate) async fn resolve_pending_download_urls<F, Fut>(
|
|
db_service: &Arc<crate::storage::db_service::RusqliteService>,
|
|
target_dir: &str,
|
|
resolve: F,
|
|
) -> Result<ResumeQueuedResult, String>
|
|
where
|
|
F: Fn(String, String, String) -> Fut,
|
|
Fut: std::future::Future<Output = Option<String>>,
|
|
{
|
|
// A row's own media_type wins; otherwise the *item's* type decides. Rows
|
|
// queued from a media card never carry one (`download_item` does not record
|
|
// it), and defaulting that NULL to 'audio' resolved movies against
|
|
// `get_audio_stream_url` — the file on disk was an audio-only transcode, so
|
|
// offline video could never play. Falling back to 'audio' only when the item
|
|
// is unknown keeps the historical behaviour for uncached items.
|
|
// TRACES: UR-071, UR-052 | DR-135
|
|
let video_types = VIDEO_ITEM_TYPES
|
|
.iter()
|
|
.map(|t| format!("'{t}'"))
|
|
.collect::<Vec<_>>()
|
|
.join(", ");
|
|
let rows_query = Query::new(&format!(
|
|
"SELECT d.id, d.item_id,
|
|
COALESCE(
|
|
d.media_type,
|
|
CASE WHEN i.item_type IN ({video_types}) THEN 'video'
|
|
WHEN i.item_type IS NOT NULL THEN 'audio'
|
|
END,
|
|
'audio'),
|
|
COALESCE(d.quality_preset, 'original')
|
|
FROM downloads d
|
|
LEFT JOIN items i ON i.id = d.item_id
|
|
WHERE d.status = 'pending' AND d.stream_url IS NULL"
|
|
));
|
|
let rows: Vec<(i64, String, String, String)> = db_service
|
|
.query_many(rows_query, |row| {
|
|
Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
|
|
})
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
|
|
if rows.is_empty() {
|
|
return Ok(ResumeQueuedResult {
|
|
resolved: 0,
|
|
failed: 0,
|
|
});
|
|
}
|
|
|
|
info!(
|
|
"[Catalog] Resolving {} offline-queued downloads on reconnect",
|
|
rows.len()
|
|
);
|
|
|
|
let mut resolved = 0usize;
|
|
let mut failed = 0usize;
|
|
|
|
for (download_id, item_id, media_type, quality) in rows {
|
|
let stream_url = match resolve(item_id.clone(), media_type, quality).await {
|
|
Some(url) => url,
|
|
None => {
|
|
failed += 1;
|
|
continue;
|
|
}
|
|
};
|
|
|
|
// Heal the row so the pump can start it. Guard on stream_url IS NULL so a
|
|
// concurrent resolver doesn't clobber an already-started row.
|
|
let update = Query::with_params(
|
|
"UPDATE downloads SET stream_url = ?, target_dir = ?
|
|
WHERE id = ? AND status = 'pending' AND stream_url IS NULL",
|
|
vec![
|
|
QueryParam::String(stream_url),
|
|
QueryParam::String(target_dir.to_string()),
|
|
QueryParam::Int64(download_id),
|
|
],
|
|
);
|
|
match db_service.execute(update).await {
|
|
Ok(n) if n > 0 => resolved += 1,
|
|
Ok(_) => {} // already resolved by someone else; not a failure
|
|
Err(e) => {
|
|
warn!(
|
|
"[Catalog] Failed to persist URL for download {}: {}",
|
|
download_id, e
|
|
);
|
|
failed += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(ResumeQueuedResult { resolved, failed })
|
|
}
|
|
|
|
/// Resolve the stream URL for every download row that was queued while offline
|
|
/// (`status = 'pending' AND stream_url IS NULL`), then pump the queue so they
|
|
/// start. Call this on reconnect.
|
|
///
|
|
/// Audio rows resolve via `get_audio_stream_url`; video rows (media_type =
|
|
/// 'video') via the pure `get_video_download_url` builder using the row's stored
|
|
/// `quality_preset` — mirroring `enqueue_video_downloads`. Rows whose URL can't
|
|
/// be resolved are left pending (they retry on the next reconnect).
|
|
#[tauri::command]
|
|
#[specta::specta]
|
|
pub async fn resume_queued_downloads(
|
|
repository: State<'_, RepositoryManagerWrapper>,
|
|
db: State<'_, DatabaseWrapper>,
|
|
download_manager: State<'_, DownloadManagerWrapper>,
|
|
app: tauri::AppHandle,
|
|
handle: String,
|
|
) -> Result<ResumeQueuedResult, String> {
|
|
use crate::repository::MediaRepository;
|
|
|
|
let repo = repository.0.get(&handle).ok_or("Repository not found")?;
|
|
|
|
// The pump needs a target_dir; use the same storage root the other download
|
|
// paths use (the database's parent directory — see `storage_get_path`).
|
|
let (db_service, target_dir) = {
|
|
let database = db.0.lock().map_err(|e| e.to_string())?;
|
|
let target_dir = database
|
|
.path()
|
|
.parent()
|
|
.ok_or_else(|| "Database path has no parent directory".to_string())?
|
|
.to_string_lossy()
|
|
.to_string();
|
|
(Arc::new(database.service()), target_dir)
|
|
};
|
|
|
|
// Recover stale downloads: rows left in 'downloading' when the app was killed
|
|
// mid-transfer are orphaned — nothing ever restarts them, so they show as
|
|
// permanently "downloading". Reset them to 'pending' and clear the stale
|
|
// stream_url so they get re-resolved and restarted from scratch below.
|
|
let recover_query = Query::new(
|
|
"UPDATE downloads SET status = 'pending', stream_url = NULL, progress = 0, \
|
|
bytes_downloaded = 0, started_at = NULL \
|
|
WHERE status = 'downloading'",
|
|
);
|
|
match db_service.execute(recover_query).await {
|
|
Ok(n) if n > 0 => info!("[Catalog] Reset {} stale 'downloading' rows to pending", n),
|
|
Ok(_) => {}
|
|
Err(e) => warn!("[Catalog] Failed to reset stale downloads: {}", e),
|
|
}
|
|
|
|
// Repair rows that completed as audio because their media_type was missing;
|
|
// they hold an audio-only transcode where a video should be, so requeue them
|
|
// for the resolver below. TRACES: UR-071 | DR-136
|
|
if let Err(e) = requeue_mistyped_video_downloads(&db_service).await {
|
|
warn!(
|
|
"[Catalog] Failed to requeue mis-typed video downloads: {}",
|
|
e
|
|
);
|
|
}
|
|
|
|
// Resolve each row's URL against the (now reachable) repository.
|
|
let repo_for_resolve = Arc::clone(&repo);
|
|
let outcome = resolve_pending_download_urls(
|
|
&db_service,
|
|
&target_dir,
|
|
move |item_id: String, media_type: String, quality: String| {
|
|
let repo = Arc::clone(&repo_for_resolve);
|
|
async move {
|
|
if media_type == "video" {
|
|
Some(
|
|
crate::repository::resolve_video_download_url(
|
|
repo.as_ref(),
|
|
&item_id,
|
|
&quality,
|
|
None,
|
|
)
|
|
.await,
|
|
)
|
|
} else {
|
|
match repo.get_audio_stream_url(&item_id).await {
|
|
Ok(url) => Some(url),
|
|
Err(e) => {
|
|
warn!(
|
|
"[Catalog] Failed to resolve audio URL for {}: {:?}",
|
|
item_id, e
|
|
);
|
|
None
|
|
}
|
|
}
|
|
}
|
|
}
|
|
},
|
|
)
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
|
|
let ResumeQueuedResult { resolved, failed } = outcome;
|
|
|
|
// Kick the pump so the newly-resolved rows actually start.
|
|
if resolved > 0 {
|
|
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;
|
|
}
|
|
|
|
info!(
|
|
"[Catalog] Resume complete: {} resolved, {} failed",
|
|
resolved, failed
|
|
);
|
|
|
|
Ok(ResumeQueuedResult { resolved, failed })
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::storage::db_service::RusqliteService;
|
|
use rusqlite::Connection;
|
|
use std::sync::Mutex;
|
|
|
|
/// The re-index policy. Pure, so it is testable without a clock, a server or
|
|
/// a database — which is the reason it was factored out of the scheduler.
|
|
///
|
|
/// TRACES: UR-065 | DR-109 | UT-115
|
|
#[test]
|
|
fn test_index_is_due() {
|
|
let ttl = Duration::from_secs(6 * 60 * 60);
|
|
let now = chrono::DateTime::parse_from_rfc3339("2026-08-04T12:00:00+00:00")
|
|
.unwrap()
|
|
.with_timezone(&chrono::Utc);
|
|
|
|
// Never indexed => due. This is the first-run case.
|
|
assert!(index_is_due(None, now, ttl));
|
|
|
|
// Indexed 7 hours ago => past the 6h TTL => due.
|
|
assert!(index_is_due(Some("2026-08-04T05:00:00+00:00"), now, ttl));
|
|
|
|
// Indexed 1 hour ago => fresh => not due. This is what stops the
|
|
// scheduler re-crawling every tick.
|
|
assert!(!index_is_due(Some("2026-08-04T11:00:00+00:00"), now, ttl));
|
|
|
|
// Exactly at the TTL boundary counts as due.
|
|
assert!(index_is_due(Some("2026-08-04T06:00:00+00:00"), now, ttl));
|
|
|
|
// A corrupt stored value must trigger a re-index, not freeze the
|
|
// catalog forever behind an unparseable timestamp.
|
|
assert!(index_is_due(Some("not-a-timestamp"), now, ttl));
|
|
assert!(index_is_due(Some(""), now, ttl));
|
|
|
|
// A timestamp in the future (clock skew, or a restored backup) is not
|
|
// due — it ages into due-ness rather than causing a crawl every tick.
|
|
assert!(!index_is_due(Some("2026-08-05T00:00:00+00:00"), now, ttl));
|
|
|
|
// Offsets other than UTC are compared as instants, not as strings.
|
|
assert!(!index_is_due(Some("2026-08-04T13:30:00+02:00"), now, ttl));
|
|
}
|
|
|
|
fn test_db() -> Arc<RusqliteService> {
|
|
let conn = Connection::open_in_memory().unwrap();
|
|
conn.execute_batch(
|
|
r#"
|
|
CREATE TABLE downloads (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
item_id TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
stream_url TEXT,
|
|
target_dir TEXT,
|
|
media_type TEXT,
|
|
quality_preset TEXT,
|
|
progress REAL DEFAULT 0,
|
|
bytes_downloaded INTEGER DEFAULT 0,
|
|
started_at TEXT,
|
|
completed_at TEXT
|
|
);
|
|
CREATE TABLE items (
|
|
id TEXT PRIMARY KEY,
|
|
item_type TEXT
|
|
);
|
|
"#,
|
|
)
|
|
.unwrap();
|
|
Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
|
|
}
|
|
|
|
async fn insert_item(db: &Arc<RusqliteService>, item_id: &str, item_type: &str) {
|
|
db.execute(Query::with_params(
|
|
"INSERT INTO items (id, item_type) VALUES (?, ?)",
|
|
vec![
|
|
QueryParam::String(item_id.to_string()),
|
|
QueryParam::String(item_type.to_string()),
|
|
],
|
|
))
|
|
.await
|
|
.unwrap();
|
|
}
|
|
|
|
async fn insert_download(
|
|
db: &Arc<RusqliteService>,
|
|
item_id: &str,
|
|
status: &str,
|
|
stream_url: Option<&str>,
|
|
media_type: Option<&str>,
|
|
) {
|
|
let q = Query::with_params(
|
|
"INSERT INTO downloads (item_id, status, stream_url, media_type) VALUES (?, ?, ?, ?)",
|
|
vec![
|
|
QueryParam::String(item_id.to_string()),
|
|
QueryParam::String(status.to_string()),
|
|
stream_url
|
|
.map(|s| QueryParam::String(s.to_string()))
|
|
.unwrap_or(QueryParam::Null),
|
|
media_type
|
|
.map(|s| QueryParam::String(s.to_string()))
|
|
.unwrap_or(QueryParam::Null),
|
|
],
|
|
);
|
|
db.execute(q).await.unwrap();
|
|
}
|
|
|
|
async fn get_row(
|
|
db: &Arc<RusqliteService>,
|
|
item_id: &str,
|
|
) -> (String, Option<String>, Option<String>) {
|
|
let q = Query::with_params(
|
|
"SELECT status, stream_url, target_dir FROM downloads WHERE item_id = ?",
|
|
vec![QueryParam::String(item_id.to_string())],
|
|
);
|
|
db.query_one(q, |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
/// IT-017: a download queued from a greyed-out offline catalog entry
|
|
/// (pending, `stream_url IS NULL`) persists, and on reconnect its URL is
|
|
/// resolved and the row is healed (URL + target dir) so the pump can start
|
|
/// it — while already-resolved rows are left untouched.
|
|
///
|
|
/// TRACES: UR-052, UR-011 | IT-017
|
|
#[tokio::test]
|
|
async fn resolves_offline_queued_row_and_leaves_resolved_rows_untouched() {
|
|
let db = test_db();
|
|
// A row queued offline: pending with no URL yet.
|
|
insert_download(&db, "queued-1", "pending", None, None).await;
|
|
// An already-resolved pending row: must NOT be touched.
|
|
insert_download(&db, "already", "pending", Some("http://existing/url"), None).await;
|
|
// A completed row: irrelevant.
|
|
insert_download(&db, "done", "completed", Some("http://done/url"), None).await;
|
|
|
|
let out =
|
|
resolve_pending_download_urls(&db, "/data/downloads", |item_id, _mt, _q| async move {
|
|
Some(format!("http://resolved/{item_id}"))
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(out.resolved, 1);
|
|
assert_eq!(out.failed, 0);
|
|
|
|
// The offline-queued row now has a URL + target dir and stays pending.
|
|
let (status, url, target) = get_row(&db, "queued-1").await;
|
|
assert_eq!(status, "pending");
|
|
assert_eq!(url.as_deref(), Some("http://resolved/queued-1"));
|
|
assert_eq!(target.as_deref(), Some("/data/downloads"));
|
|
|
|
// The already-resolved row is unchanged (not re-resolved).
|
|
let (_s, url2, _t) = get_row(&db, "already").await;
|
|
assert_eq!(url2.as_deref(), Some("http://existing/url"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn counts_unresolvable_rows_as_failed_and_leaves_them_pending() {
|
|
let db = test_db();
|
|
insert_download(&db, "bad", "pending", None, None).await;
|
|
|
|
// Resolver returns None (e.g. server lookup failed).
|
|
let out = resolve_pending_download_urls(&db, "/data", |_id, _mt, _q| async move { None })
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(out.resolved, 0);
|
|
assert_eq!(out.failed, 1);
|
|
|
|
// Still pending with no URL, so a later reconnect can retry it.
|
|
let (status, url, _t) = get_row(&db, "bad").await;
|
|
assert_eq!(status, "pending");
|
|
assert_eq!(url, None);
|
|
}
|
|
|
|
/// A movie queued from a media card has no `media_type` — `download_item`
|
|
/// never records one. Defaulting that NULL to 'audio' resolved the row
|
|
/// against `get_audio_stream_url`, so the "downloaded movie" on disk was an
|
|
/// audio-only transcode and offline video playback could never work. The
|
|
/// item's own type is the authority.
|
|
///
|
|
/// TRACES: UR-071, UR-052 | DR-135 | UT-125
|
|
#[tokio::test]
|
|
async fn null_media_type_resolves_from_the_item_type_not_audio() {
|
|
let db = test_db();
|
|
insert_item(&db, "movie-1", "Movie").await;
|
|
insert_item(&db, "ep-1", "Episode").await;
|
|
insert_item(&db, "track-1", "Audio").await;
|
|
for id in ["movie-1", "ep-1", "track-1"] {
|
|
insert_download(&db, id, "pending", None, None).await;
|
|
}
|
|
|
|
let seen = Arc::new(Mutex::new(Vec::new()));
|
|
let seen_c = Arc::clone(&seen);
|
|
resolve_pending_download_urls(&db, "/data", move |item_id, media_type, _q| {
|
|
let seen = Arc::clone(&seen_c);
|
|
async move {
|
|
seen.lock().unwrap().push((item_id.clone(), media_type));
|
|
Some(format!("http://resolved/{item_id}"))
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
let seen = seen.lock().unwrap().clone();
|
|
let of = |id: &str| {
|
|
seen.iter()
|
|
.find(|(i, _)| i == id)
|
|
.map(|(_, m)| m.clone())
|
|
.unwrap()
|
|
};
|
|
assert_eq!(of("movie-1"), "video", "a Movie must download as video");
|
|
assert_eq!(of("ep-1"), "video", "an Episode must download as video");
|
|
assert_eq!(of("track-1"), "audio", "a track is still audio");
|
|
}
|
|
|
|
/// An unknown item (never cached locally) has no type to derive from, so it
|
|
/// keeps the historical audio default rather than failing the row.
|
|
///
|
|
/// TRACES: UR-071 | DR-135 | UT-125
|
|
#[tokio::test]
|
|
async fn unknown_item_falls_back_to_audio() {
|
|
let db = test_db();
|
|
insert_download(&db, "ghost", "pending", None, None).await;
|
|
|
|
let seen = Arc::new(Mutex::new(String::new()));
|
|
let seen_c = Arc::clone(&seen);
|
|
resolve_pending_download_urls(&db, "/data", move |_id, media_type, _q| {
|
|
let seen = Arc::clone(&seen_c);
|
|
async move {
|
|
*seen.lock().unwrap() = media_type;
|
|
Some("http://x".to_string())
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(*seen.lock().unwrap(), "audio");
|
|
}
|
|
|
|
/// An explicit `media_type` on the row always wins over the item's type.
|
|
///
|
|
/// TRACES: UR-071 | DR-135 | UT-125
|
|
#[tokio::test]
|
|
async fn explicit_media_type_beats_the_item_type() {
|
|
let db = test_db();
|
|
insert_item(&db, "odd", "Audio").await;
|
|
insert_download(&db, "odd", "pending", None, Some("video")).await;
|
|
|
|
let seen = Arc::new(Mutex::new(String::new()));
|
|
let seen_c = Arc::clone(&seen);
|
|
resolve_pending_download_urls(&db, "/data", move |_id, media_type, _q| {
|
|
let seen = Arc::clone(&seen_c);
|
|
async move {
|
|
*seen.lock().unwrap() = media_type;
|
|
Some("http://x".to_string())
|
|
}
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(*seen.lock().unwrap(), "video");
|
|
}
|
|
|
|
/// Rows already downloaded under the audio default hold an audio-only
|
|
/// transcode on disk, so they play as a broken video forever. They are
|
|
/// identifiable — no `media_type` but a video item — and are requeued so the
|
|
/// resolver fetches the real video. Correctly-typed rows and genuine audio
|
|
/// downloads must be left alone.
|
|
///
|
|
/// TRACES: UR-071 | DR-136 | UT-126
|
|
#[tokio::test]
|
|
async fn requeues_video_downloaded_under_the_audio_default() {
|
|
let db = test_db();
|
|
insert_item(&db, "movie-1", "Movie").await;
|
|
insert_item(&db, "track-1", "Audio").await;
|
|
insert_item(&db, "movie-ok", "Movie").await;
|
|
// Mis-downloaded: completed, no media_type, video item.
|
|
insert_download(&db, "movie-1", "completed", Some("http://audio/url"), None).await;
|
|
// A real audio download: untouched.
|
|
insert_download(&db, "track-1", "completed", Some("http://audio/ok"), None).await;
|
|
// A correctly-typed video download: untouched.
|
|
insert_download(
|
|
&db,
|
|
"movie-ok",
|
|
"completed",
|
|
Some("http://video/ok"),
|
|
Some("video"),
|
|
)
|
|
.await;
|
|
|
|
let requeued = requeue_mistyped_video_downloads(&db).await.unwrap();
|
|
assert_eq!(requeued, 1);
|
|
|
|
let (status, url, _t) = get_row(&db, "movie-1").await;
|
|
assert_eq!(status, "pending", "the mis-typed row must download again");
|
|
assert_eq!(url, None, "its audio URL must be cleared so it re-resolves");
|
|
|
|
let (status, url, _t) = get_row(&db, "track-1").await;
|
|
assert_eq!(status, "completed", "a real audio download is untouched");
|
|
assert_eq!(url.as_deref(), Some("http://audio/ok"));
|
|
|
|
let (status, _u, _t) = get_row(&db, "movie-ok").await;
|
|
assert_eq!(status, "completed", "a correct video download is untouched");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn video_rows_use_media_type_in_resolver() {
|
|
let db = test_db();
|
|
insert_download(&db, "vid-1", "pending", None, Some("video")).await;
|
|
|
|
let out =
|
|
resolve_pending_download_urls(&db, "/data", |item_id, media_type, _q| async move {
|
|
assert_eq!(media_type, "video");
|
|
Some(format!("http://transcode/{item_id}"))
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(out.resolved, 1);
|
|
let (_s, url, _t) = get_row(&db, "vid-1").await;
|
|
assert_eq!(url.as_deref(), Some("http://transcode/vid-1"));
|
|
}
|
|
}
|