dtourolle
1a9805f0f3
An album download put a handful of its tracks on the device while the button reported the album as downloaded. Two independent gaps, one shared cause. - `download_album` read its track list from `items WHERE album_id = ?` — the local catalog cache. Jellyfin does not return `AlbumId` on every listing endpoint, so tracks cached from one of those sit in `items` with a NULL `album_id` and are invisible to that query. On the reported database three whole albums (18, 12 and 9 tracks) had it NULL on every track; a partially linked album queued only the linked subset. - The frontend then resolved one stream URL per track from its own list and paired it with the returned row ids by position. The ids came back in the backend's `index_number` order over a different set of rows, so a row could be handed another track's URL and any track past the end of the shorter list was never started. On Android that loop also stopped wherever the webview was suspended. - `album_id` is what `OfflineRepository::get_items` joins a track to its album on, so a track that did download stayed invisible under its album offline — the same missing link seen from the other side. The operation now belongs to Rust end to end: - `HybridRepository::get_album_tracks` asks the server what the album contains. Cache-first `get_items` is right for browsing and wrong for deciding what to download; it errors offline so the caller falls back to the ungated local catalog, keeping the queue-while-offline flow. - `queue_album_tracks` writes the album link onto every track it queues, and creates an `items` row for tracks the cache has never seen. - Stream URLs resolve here, through the existing reconnect resolver, now scoped to the rows just queued so one album cannot start every unrelated pending row. Only the album id crosses the IPC boundary. - `album_file_names` gives each track its own file. A title repeated inside one album (deluxe edition, two discs) mapped to one path, so those downloads overwrote each other. Re-tapping download on a broken album heals it: missing tracks are queued and the tracks already on disk get their link. `download_series`/`download_season` still derive their episode lists from the cache the same way and want the same treatment. DR-173, UT-170..172. Rust 673 tests, frontend 975 tests, svelte-check and check:boundary clean. Note: this tree is shared with a concurrent session. Only the files above are committed; docs/traceability.md is left to be regenerated once that work lands.
1146 lines
42 KiB
Rust
1146 lines
42 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)`.
|
|
///
|
|
/// `only_ids` restricts the sweep to specific download rows. Reconnect passes
|
|
/// `None` and heals everything; a bulk enqueue (an album, say) passes the rows
|
|
/// it just created, so clicking download on one album cannot also start every
|
|
/// unrelated row that has been sitting pending.
|
|
pub(crate) async fn resolve_pending_download_urls<F, Fut>(
|
|
db_service: &Arc<crate::storage::db_service::RusqliteService>,
|
|
target_dir: &str,
|
|
only_ids: Option<&[i64]>,
|
|
resolve: F,
|
|
) -> Result<ResumeQueuedResult, String>
|
|
where
|
|
F: Fn(String, String, String) -> Fut,
|
|
Fut: std::future::Future<Output = Option<String>>,
|
|
{
|
|
if only_ids.is_some_and(|ids| ids.is_empty()) {
|
|
return Ok(ResumeQueuedResult {
|
|
resolved: 0,
|
|
failed: 0,
|
|
});
|
|
}
|
|
// 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 id_filter = match only_ids {
|
|
Some(ids) => format!(
|
|
" AND d.id IN ({})",
|
|
ids.iter()
|
|
.map(|id| id.to_string())
|
|
.collect::<Vec<_>>()
|
|
.join(", ")
|
|
),
|
|
None => String::new(),
|
|
};
|
|
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{id_filter}"
|
|
));
|
|
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,
|
|
None,
|
|
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",
|
|
None,
|
|
|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"));
|
|
}
|
|
|
|
/// A bulk enqueue resolves only the rows it just created. Downloading one
|
|
/// album must not also start every unrelated row that has been sitting
|
|
/// pending with no URL (the smart cache leaves plenty of those).
|
|
///
|
|
/// TRACES: UR-018, UR-055 | DR-173 | UT-171
|
|
#[tokio::test]
|
|
async fn only_ids_restricts_the_sweep_to_the_given_rows() {
|
|
let db = test_db();
|
|
insert_download(&db, "mine", "pending", None, Some("audio")).await;
|
|
insert_download(&db, "someone-elses", "pending", None, Some("audio")).await;
|
|
|
|
let mine: i64 = db
|
|
.query_one(
|
|
Query::new("SELECT id FROM downloads WHERE item_id = 'mine'"),
|
|
|row| row.get(0),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
let out = resolve_pending_download_urls(
|
|
&db,
|
|
"/data",
|
|
Some(&[mine]),
|
|
|item_id, _mt, _q| async move { Some(format!("http://resolved/{item_id}")) },
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(out.resolved, 1);
|
|
assert_eq!(out.failed, 0);
|
|
|
|
let (_s, url, _t) = get_row(&db, "mine").await;
|
|
assert_eq!(url.as_deref(), Some("http://resolved/mine"));
|
|
|
|
let (status, other_url, _t) = get_row(&db, "someone-elses").await;
|
|
assert_eq!(status, "pending");
|
|
assert_eq!(
|
|
other_url, None,
|
|
"a scoped resolve must leave unrelated pending rows alone"
|
|
);
|
|
}
|
|
|
|
/// An empty id list resolves nothing — it must not fall through to "sweep
|
|
/// everything", which is what an unguarded `IN ()` would amount to.
|
|
///
|
|
/// TRACES: UR-018, UR-055 | DR-173 | UT-171
|
|
#[tokio::test]
|
|
async fn an_empty_id_list_resolves_nothing() {
|
|
let db = test_db();
|
|
insert_download(&db, "untouched", "pending", None, Some("audio")).await;
|
|
|
|
let out =
|
|
resolve_pending_download_urls(&db, "/data", Some(&[]), |item_id, _mt, _q| async move {
|
|
Some(format!("http://resolved/{item_id}"))
|
|
})
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(out.resolved, 0);
|
|
let (_s, url, _t) = get_row(&db, "untouched").await;
|
|
assert_eq!(url, None);
|
|
}
|
|
|
|
#[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", None, |_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", None, 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", None, 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", None, 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",
|
|
None,
|
|
|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"));
|
|
}
|
|
}
|