mirror of
https://github.com/narrrl/proton-stream.git
synced 2026-11-03 16:41:26 +01:00
nvrl
c5808a5e50
- right-click a library tile to play or resume it, mark it watched or unwatched, download it, change its match, or take it off Continue watching - mpv's demuxer-cache-time is shaded on the seek bar, so it shows whether a jump lands in what is already read ahead - paused-for-cache shows a spinner over the picture and in the controls
2624 lines
100 KiB
Rust
2624 lines
100 KiB
Rust
//! Everything the UI is not allowed to do on its own thread.
|
||
//!
|
||
//! egui redraws at the display's rate; opening a share takes seconds, a crawl
|
||
//! takes minutes, and a thumbnail is a network round-trip plus a PGP decrypt.
|
||
//! So the UI holds an [`Engine`], calls a method that returns immediately, and
|
||
//! learns what happened from an [`Event`] on a channel. Nothing here blocks, and
|
||
//! nothing here draws.
|
||
//!
|
||
//! ```text
|
||
//! ui thread tokio runtime worker threads
|
||
//! ───────── ───────────── ──────────────
|
||
//! engine.crawl() ──spawn──▶ SharedLibrary::crawl ──▶ block decrypt
|
||
//! ▲ │
|
||
//! └──── Event::LibraryLoaded ◀────┘ + ctx.request_repaint()
|
||
//! ```
|
||
//!
|
||
//! The repaint call is the part that is easy to forget: without it an event
|
||
//! lands in the channel and sits there until the viewer happens to move the
|
||
//! mouse.
|
||
|
||
use std::collections::{HashMap, HashSet, VecDeque};
|
||
use std::path::Path;
|
||
use std::sync::Arc;
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::sync::mpsc::{Receiver, Sender, channel};
|
||
|
||
use parking_lot::Mutex;
|
||
use pstr_core::appearance::Appearance;
|
||
use pstr_core::catalog::{
|
||
Catalog, CatalogNode, OfflineFile, TitleTrackPrefs, WatchState, build_rows,
|
||
};
|
||
use pstr_core::config::AppDirs;
|
||
use pstr_core::library::{Library, Title, TitleKind};
|
||
use pstr_core::metadata::{EpisodeGuide, MetadataConfig, MetadataRecord, ProviderId};
|
||
use pstr_core::prefs::PlaybackPrefs;
|
||
use pstr_core::proton_drive_rs::ThumbnailType;
|
||
use pstr_core::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
|
||
use pstr_core::{Share, ShareStore, SharedLibrary};
|
||
use pstr_meta::MetadataService;
|
||
use pstr_stream::{
|
||
BlockSource, DiskCacheConfig, FileBlocks, LibraryOpener, StreamConfig, StreamSource,
|
||
VideoStream,
|
||
};
|
||
use tokio::runtime::Runtime;
|
||
|
||
use crate::playback::PlaybackTarget;
|
||
|
||
/// Stable identity for an episode download. No share secret is included.
|
||
#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
|
||
pub struct DownloadKey {
|
||
pub share_id: String,
|
||
pub link_id: String,
|
||
}
|
||
|
||
impl From<&PlaybackTarget> for DownloadKey {
|
||
fn from(target: &PlaybackTarget) -> Self {
|
||
Self {
|
||
share_id: target.share_id.clone(),
|
||
link_id: target.link_id.clone(),
|
||
}
|
||
}
|
||
}
|
||
|
||
/// A download state suitable for presenting directly to the viewer.
|
||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
pub enum DownloadState {
|
||
Queued,
|
||
Running,
|
||
Paused,
|
||
Completed,
|
||
Failed(String),
|
||
/// Work stopped, but its complete-block partial is retained for Resume.
|
||
Cancelled,
|
||
}
|
||
|
||
/// Current, inspectable state for one episode in a download batch.
|
||
#[derive(Debug, Clone)]
|
||
pub struct DownloadItem {
|
||
pub key: DownloadKey,
|
||
pub target: PlaybackTarget,
|
||
pub state: DownloadState,
|
||
pub downloaded: u64,
|
||
pub total: u64,
|
||
/// Bytes a second, smoothed. Only meaningful while it is running.
|
||
pub rate: f64,
|
||
}
|
||
|
||
impl DownloadItem {
|
||
/// Seconds until done at the current rate, while that is knowable.
|
||
pub fn eta(&self) -> Option<f64> {
|
||
(self.state == DownloadState::Running && self.rate > 0.0 && self.total > self.downloaded)
|
||
.then(|| (self.total - self.downloaded) as f64 / self.rate)
|
||
}
|
||
|
||
pub fn percent(&self) -> f32 {
|
||
if self.total == 0 {
|
||
0.0
|
||
} else {
|
||
(self.downloaded as f64 / self.total as f64).clamp(0.0, 1.0) as f32
|
||
}
|
||
}
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
enum DownloadCommand {
|
||
Run,
|
||
Pause,
|
||
Cancel,
|
||
}
|
||
|
||
struct DownloadControl {
|
||
command: tokio::sync::watch::Sender<DownloadCommand>,
|
||
}
|
||
|
||
impl DownloadControl {
|
||
fn new() -> Self {
|
||
let (command, _) = tokio::sync::watch::channel(DownloadCommand::Run);
|
||
Self { command }
|
||
}
|
||
|
||
fn set(&self, command: DownloadCommand) {
|
||
self.command.send_replace(command);
|
||
}
|
||
|
||
fn subscribe(&self) -> tokio::sync::watch::Receiver<DownloadCommand> {
|
||
self.command.subscribe()
|
||
}
|
||
}
|
||
|
||
struct DownloadJob {
|
||
control: Arc<DownloadControl>,
|
||
handle: tokio::task::JoinHandle<()>,
|
||
live: Arc<AtomicBool>,
|
||
}
|
||
|
||
#[derive(Clone, Copy)]
|
||
enum DownloadEnd {
|
||
Completed,
|
||
Cancelled,
|
||
}
|
||
|
||
#[derive(Default)]
|
||
struct Downloads {
|
||
items: HashMap<DownloadKey, DownloadItem>,
|
||
jobs: HashMap<DownloadKey, DownloadJob>,
|
||
}
|
||
|
||
#[derive(serde::Serialize, serde::Deserialize)]
|
||
struct PartialMarker {
|
||
revision_id: String,
|
||
block_sizes: Vec<u64>,
|
||
#[serde(default)]
|
||
target: Option<PlaybackTarget>,
|
||
}
|
||
|
||
/// Poster thumbnails are decoded to at most this edge before they become a
|
||
/// texture. Proton's preview is larger than any tile this UI draws, and the
|
||
/// difference is VRAM held for every title in the library.
|
||
const THUMBNAIL_MAX_EDGE: u32 = 640;
|
||
|
||
/// What the background side tells the UI.
|
||
pub enum Event {
|
||
/// The share list changed on disk.
|
||
Shares(Vec<Share>),
|
||
/// The share the form asked for is stored; its crawl is starting.
|
||
ShareAdded(String),
|
||
/// The share the form asked for was refused, with why. Nothing was crawled.
|
||
ShareRejected(String),
|
||
/// Shares opened; playback is possible from here.
|
||
Connected {
|
||
/// Shares that did not open, with why.
|
||
failures: Vec<(String, String)>,
|
||
},
|
||
/// Shares could not be opened at all.
|
||
ConnectFailed(String),
|
||
/// The catalog was re-read.
|
||
LibraryLoaded(Library),
|
||
/// A crawl finished for one share.
|
||
Crawled {
|
||
share_id: String,
|
||
nodes: usize,
|
||
files: usize,
|
||
seconds: f64,
|
||
},
|
||
/// A crawl is under way: which share, and how many nodes so far.
|
||
CrawlProgress { share: String, found: usize },
|
||
/// Every requested crawl is done.
|
||
CrawlFinished,
|
||
/// That share's crawl was stopped before it stored anything.
|
||
CrawlStopped { share_id: String },
|
||
/// A decoded Proton thumbnail, keyed as [`thumbnail_key`].
|
||
Thumbnail {
|
||
key: String,
|
||
image: egui::ColorImage,
|
||
},
|
||
/// No thumbnail for this file — remembered, so the grid stops asking.
|
||
ThumbnailMissing { key: String },
|
||
/// Everything the catalog knows about titles, keyed by title key.
|
||
Metadata(HashMap<String, MetadataRecord>),
|
||
/// What the provider said about individual episodes, keyed by title key.
|
||
EpisodeMetadata(HashMap<String, EpisodeGuide>),
|
||
/// The enrichment settings changed, on disk.
|
||
MetadataConfig(MetadataConfig),
|
||
/// A matching run finished, with how it went.
|
||
Matched {
|
||
matched: usize,
|
||
unmatched: usize,
|
||
failed: usize,
|
||
},
|
||
/// Every requested match is done.
|
||
MatchFinished,
|
||
/// What the provider answered a hand search with, unscored and in its own
|
||
/// order. `title_key` says which title asked, because the viewer can close
|
||
/// the search and open another one while a request is still out.
|
||
MatchOptions {
|
||
title_key: String,
|
||
options: Vec<pstr_core::metadata::TitleMetadata>,
|
||
},
|
||
/// A hand search did not get an answer. Distinct from an empty one: nothing
|
||
/// found is a result, and a rate limit is not.
|
||
MatchSearchFailed { title_key: String, error: String },
|
||
/// A decoded piece of provider artwork, keyed by title key.
|
||
Poster {
|
||
key: String,
|
||
image: egui::ColorImage,
|
||
},
|
||
/// The artwork for this title could not be had.
|
||
PosterMissing { key: String },
|
||
/// A stream opened and is ready to hand to mpv.
|
||
PlaybackReady {
|
||
target: Box<PlaybackTarget>,
|
||
stream: VideoStream,
|
||
},
|
||
/// mpv said something about what is playing. `id` is the player it came
|
||
/// from — a player being replaced goes on talking for a moment after the
|
||
/// next one has started.
|
||
Player {
|
||
id: u64,
|
||
event: pstr_player::PlayerEvent,
|
||
},
|
||
/// That player is gone: its window was closed, or playback ended.
|
||
PlayerStopped { id: u64 },
|
||
/// One file's watch state was saved, for the library to patch in place.
|
||
Watched {
|
||
share_id: String,
|
||
link_id: String,
|
||
state: WatchState,
|
||
},
|
||
/// Something the viewer should see, phrased for them.
|
||
Error(String),
|
||
/// Something the viewer might like to see, briefly.
|
||
Status(String),
|
||
/// Full download-manager snapshot. Replacing rather than patching avoids
|
||
/// stale rows when a task is removed.
|
||
Downloads(Vec<DownloadItem>),
|
||
/// One running download moved on. Sent instead of a snapshot for byte
|
||
/// progress, which arrives several times a second per download, while the
|
||
/// list it would copy and sort holds every file ever kept offline.
|
||
DownloadProgress(Box<DownloadItem>),
|
||
/// Completed local copies, for badges and online-only actions.
|
||
OfflineFiles(HashSet<DownloadKey>),
|
||
}
|
||
|
||
/// The share clients and the block layer, once they exist.
|
||
struct Connection {
|
||
library: Arc<SharedLibrary>,
|
||
source: StreamSource,
|
||
}
|
||
|
||
/// The background half of the app.
|
||
///
|
||
/// Cheap to clone: everything inside is shared. The UI keeps one, the tasks it
|
||
/// spawns keep clones.
|
||
#[derive(Clone)]
|
||
pub struct Engine {
|
||
runtime: Arc<Runtime>,
|
||
dirs: AppDirs,
|
||
store: Arc<ShareStore>,
|
||
/// One connection, guarded rather than owned by the UI, because a task that
|
||
/// finishes long after the click that started it still needs it.
|
||
connection: Arc<Mutex<Option<Connection>>>,
|
||
/// Held across the handshake, which a `parking_lot` guard cannot be.
|
||
///
|
||
/// Without it the first frame is a stampede: `connect`, the click that
|
||
/// wanted a stream and every visible poster all find no connection and open
|
||
/// every share at once. One winner, everyone else waits and takes its
|
||
/// result.
|
||
connecting: Arc<tokio::sync::Mutex<()>>,
|
||
/// `rusqlite::Connection` is `Send` but not `Sync`, and both the UI (watch
|
||
/// state) and crawls (rows) write. One lock, held for the length of a
|
||
/// statement.
|
||
catalog: Arc<Mutex<Catalog>>,
|
||
/// Thumbnail fetches in flight. A fast scroll can ask for a hundred, and
|
||
/// each is a round-trip plus a decrypt — unbounded, they crowd out the reads
|
||
/// the player is blocked on.
|
||
thumbnails: Arc<tokio::sync::Semaphore>,
|
||
/// Provider lookups in flight. Far tighter than the thumbnail limit, and
|
||
/// not for our benefit: both providers rate-limit by the minute, and a
|
||
/// library of three hundred titles fired off at once earns a 429 for the
|
||
/// whole run rather than for the tail of it.
|
||
lookups: Arc<tokio::sync::Semaphore>,
|
||
/// The enrichment settings and the provider they resolve to. Rebuilt
|
||
/// whenever the settings change, so nothing holds a client for a provider
|
||
/// the viewer has switched away from.
|
||
metadata: Arc<Mutex<Enrichment>>,
|
||
/// Volume, mute and language preferences. Held here rather than in the UI
|
||
/// because a new player is built from them, and the player is started from
|
||
/// a background event.
|
||
prefs: Arc<Mutex<PlaybackPrefs>>,
|
||
/// Which theme the window wears. Kept beside the playback preferences for
|
||
/// the same reason: it is read from the UI and written to disk off it.
|
||
appearance: Arc<Mutex<Appearance>>,
|
||
/// Preloaded so clicking Play never runs SQLite on egui's thread.
|
||
track_prefs: Arc<Mutex<HashMap<String, TitleTrackPrefs>>>,
|
||
downloads: Arc<Mutex<Downloads>>,
|
||
download_permits: Arc<tokio::sync::Semaphore>,
|
||
/// Whether the player page is on screen. See [`Engine::wake`].
|
||
picture_shown: Arc<AtomicBool>,
|
||
/// Bumped to stop the crawls in flight; each watches it from its start.
|
||
crawl_cancel: Arc<tokio::sync::watch::Sender<u64>>,
|
||
events: Sender<Event>,
|
||
ctx: egui::Context,
|
||
}
|
||
|
||
/// How many posters may be in flight at once.
|
||
const THUMBNAIL_CONCURRENCY: usize = 6;
|
||
|
||
/// How many provider lookups may be in flight at once. See [`Engine::lookups`].
|
||
const LOOKUP_CONCURRENCY: usize = 2;
|
||
const DOWNLOAD_CONCURRENCY: usize = 3;
|
||
/// How many blocks of one download are fetched at once. Four is what the
|
||
/// stream layer itself allows a single read, for the same reason: enough to
|
||
/// hide a round trip, few enough that three downloads do not swamp a player.
|
||
const DOWNLOAD_BLOCKS_IN_FLIGHT: usize = 4;
|
||
|
||
/// Enrichment as currently configured.
|
||
struct Enrichment {
|
||
config: MetadataConfig,
|
||
/// `None` when enrichment is off, or when the provider needs an API key
|
||
/// that is not stored. Both are ordinary states, not errors: the difference
|
||
/// only matters when the viewer asks for a match, which is where it is
|
||
/// reported.
|
||
service: Option<MetadataService>,
|
||
}
|
||
|
||
impl Enrichment {
|
||
/// Resolve a service for `config`, if one can be had.
|
||
fn build(config: MetadataConfig) -> Self {
|
||
let service = config.enabled.then(|| {
|
||
let key = pstr_meta::settings::api_key(config.provider);
|
||
MetadataService::new(&config, key)
|
||
.inspect_err(|error| tracing::debug!("metadata provider unavailable: {error}"))
|
||
.ok()
|
||
});
|
||
|
||
Self {
|
||
config,
|
||
service: service.flatten(),
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Engine {
|
||
/// Build the engine and the channel the UI drains.
|
||
pub fn new(
|
||
runtime: Arc<Runtime>,
|
||
dirs: AppDirs,
|
||
ctx: egui::Context,
|
||
) -> anyhow::Result<(Self, Receiver<Event>)> {
|
||
let catalog = Catalog::open(&dirs.catalog_db())?;
|
||
let (events, receiver) = channel();
|
||
|
||
// An unreadable settings file must not stop the app starting: the
|
||
// library is the point and enrichment is decoration. It is reported and
|
||
// treated as "off", which is also the safe reading — never as "on".
|
||
let config = pstr_meta::settings::load(&dirs).unwrap_or_else(|error| {
|
||
tracing::warn!("read the metadata settings: {error}");
|
||
MetadataConfig::default()
|
||
});
|
||
|
||
// Same reading as the metadata settings above: a preferences file that
|
||
// will not parse is worth a line in the log and nothing more. It is
|
||
// volume and a language, and the defaults play the film.
|
||
let prefs = pstr_core::prefs::load(&dirs).unwrap_or_else(|error| {
|
||
tracing::warn!("read the playback preferences: {error}");
|
||
PlaybackPrefs::default()
|
||
});
|
||
|
||
// And the same again for the theme, which is the least load-bearing
|
||
// file of the three: an unreadable one costs the viewer their colours
|
||
// until they pick them again.
|
||
let appearance = pstr_core::appearance::load(&dirs).unwrap_or_else(|error| {
|
||
tracing::warn!("read the appearance: {error}");
|
||
Appearance::default()
|
||
});
|
||
|
||
let track_prefs = catalog.all_title_track_prefs().unwrap_or_else(|error| {
|
||
tracing::warn!("read show track preferences: {error}");
|
||
HashMap::new()
|
||
});
|
||
let engine = Self {
|
||
runtime,
|
||
store: Arc::new(ShareStore::new(dirs.clone())),
|
||
dirs,
|
||
connection: Arc::new(Mutex::new(None)),
|
||
connecting: Arc::new(tokio::sync::Mutex::new(())),
|
||
catalog: Arc::new(Mutex::new(catalog)),
|
||
thumbnails: Arc::new(tokio::sync::Semaphore::new(THUMBNAIL_CONCURRENCY)),
|
||
lookups: Arc::new(tokio::sync::Semaphore::new(LOOKUP_CONCURRENCY)),
|
||
metadata: Arc::new(Mutex::new(Enrichment::build(config))),
|
||
prefs: Arc::new(Mutex::new(prefs)),
|
||
appearance: Arc::new(Mutex::new(appearance)),
|
||
track_prefs: Arc::new(Mutex::new(track_prefs)),
|
||
downloads: Arc::new(Mutex::new(Downloads::default())),
|
||
download_permits: Arc::new(tokio::sync::Semaphore::new(DOWNLOAD_CONCURRENCY)),
|
||
picture_shown: Arc::new(AtomicBool::new(false)),
|
||
crawl_cancel: Arc::new(tokio::sync::watch::channel(0).0),
|
||
events,
|
||
ctx,
|
||
};
|
||
Ok((engine, receiver))
|
||
}
|
||
|
||
pub fn runtime(&self) -> &Arc<Runtime> {
|
||
&self.runtime
|
||
}
|
||
|
||
/// The playback preferences as they stand.
|
||
pub fn playback_prefs(&self) -> PlaybackPrefs {
|
||
self.prefs.lock().clone()
|
||
}
|
||
|
||
/// Change the preferences, and write them if asked.
|
||
///
|
||
/// `commit` is what keeps a dragged volume slider from being a file write
|
||
/// per frame: every change is kept in memory, so a player started mid-drag
|
||
/// is built from the current value, and the disk is touched when the drag
|
||
/// ends. Losing the last few points of a drag to a crash costs nothing.
|
||
pub fn set_playback_prefs(&self, prefs: PlaybackPrefs, commit: bool) {
|
||
let prefs = prefs.sanitized();
|
||
*self.prefs.lock() = prefs.clone();
|
||
if !commit {
|
||
return;
|
||
}
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
if let Err(error) = pstr_core::prefs::save(&engine.dirs, &prefs) {
|
||
// Not `fail`: the viewer asked to change the volume, not to
|
||
// save a file, and the change did happen. Telling them their
|
||
// config directory is unwritable in a red line under the film
|
||
// helps nobody mid-episode.
|
||
tracing::warn!("save the playback preferences: {error}");
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Which theme the window wears.
|
||
pub fn appearance(&self) -> Appearance {
|
||
*self.appearance.lock()
|
||
}
|
||
|
||
/// Change the theme and write it.
|
||
///
|
||
/// Always committed, unlike the volume: a theme is picked with a click and
|
||
/// there is no drag to coalesce.
|
||
pub fn set_appearance(&self, appearance: Appearance) {
|
||
*self.appearance.lock() = appearance;
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
if let Err(error) = pstr_core::appearance::save(&engine.dirs, &appearance) {
|
||
// Not `fail`, for the same reason the playback preferences are
|
||
// not: the change happened, and only the remembering did not.
|
||
tracing::warn!("save the appearance: {error}");
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Send an event and wake the UI. A closed channel means the window is
|
||
/// gone, which is not an error worth propagating anywhere.
|
||
pub(crate) fn emit(&self, event: Event) {
|
||
// The two that arrive on mpv's clock rather than on anything happening.
|
||
let ticking = matches!(
|
||
event,
|
||
Event::Player {
|
||
event: pstr_player::PlayerEvent::Position(_)
|
||
| pstr_player::PlayerEvent::Buffered(_),
|
||
..
|
||
}
|
||
);
|
||
if self.events.send(event).is_ok() {
|
||
if ticking {
|
||
self.wake();
|
||
} else {
|
||
self.ctx.request_repaint();
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Say whether the player page is the one on screen.
|
||
pub fn set_picture_shown(&self, shown: bool) {
|
||
self.picture_shown.store(shown, Ordering::Relaxed);
|
||
}
|
||
|
||
/// Ask for a frame because playback moved on.
|
||
///
|
||
/// mpv reports the position and offers a new picture once per video frame.
|
||
/// On the player page that is exactly the rate to draw at. Anywhere else
|
||
/// the film is a line of text and a bar in the transport, and redrawing a
|
||
/// whole library page at the film's frame rate to move them is most of the
|
||
/// app's CPU while something plays in the background. Four times a second
|
||
/// keeps the clock honest.
|
||
pub fn wake(&self) {
|
||
/// How often the transport bar moves while the picture is hidden.
|
||
const BACKGROUND_TICK: std::time::Duration = std::time::Duration::from_millis(250);
|
||
|
||
if self.picture_shown.load(Ordering::Relaxed) {
|
||
self.ctx.request_repaint();
|
||
} else {
|
||
self.ctx.request_repaint_after(BACKGROUND_TICK);
|
||
}
|
||
}
|
||
|
||
fn fail(&self, context: &str, error: impl std::fmt::Display) {
|
||
tracing::warn!("{context}: {error}");
|
||
self.emit(Event::Error(format!("{context}: {error}")));
|
||
}
|
||
|
||
// ---------------------------------------------------------------- shares
|
||
|
||
/// Re-read the share list.
|
||
pub fn load_shares(&self) {
|
||
match self.store.list() {
|
||
Ok(shares) => self.emit(Event::Shares(shares)),
|
||
Err(error) => self.fail("read the share list", error),
|
||
}
|
||
}
|
||
|
||
/// Add a share, then open and crawl it — which is what "add" means to
|
||
/// someone who just pasted a link.
|
||
pub fn add_share(&self, name: String, url: String, password: Option<String>) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
let share = match engine.store.add(&name, &url, password.as_deref()) {
|
||
Ok(share) => share,
|
||
Err(error) => return engine.emit(Event::ShareRejected(error.to_string())),
|
||
};
|
||
engine.emit(Event::ShareAdded(share.name.clone()));
|
||
engine.load_shares();
|
||
engine.connect_and_crawl(Some(share.id)).await;
|
||
});
|
||
}
|
||
|
||
/// Forget a share and its stored secrets.
|
||
pub fn remove_share(&self, id: String) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
let keys: Vec<_> = engine
|
||
.downloads
|
||
.lock()
|
||
.jobs
|
||
.keys()
|
||
.filter(|key| key.share_id == id)
|
||
.cloned()
|
||
.collect();
|
||
for key in &keys {
|
||
engine.cancel_and_join(key).await;
|
||
}
|
||
|
||
let offline = match engine.catalog.lock().offline_files_for_share(&id) {
|
||
Ok(files) => files,
|
||
Err(error) => return engine.fail("inspect that share's downloads", error),
|
||
};
|
||
for (link_id, file) in &offline {
|
||
let path = engine.dirs.offline_file(&id, link_id, &file.revision_id);
|
||
if let Err(error) = remove_if_present(&path).await {
|
||
return engine.fail("delete that share's offline bytes", error);
|
||
}
|
||
}
|
||
let item_keys: Vec<_> = engine
|
||
.downloads
|
||
.lock()
|
||
.items
|
||
.keys()
|
||
.filter(|key| key.share_id == id)
|
||
.cloned()
|
||
.collect();
|
||
for key in &item_keys {
|
||
let (partial, marker) = engine.partial_paths(key);
|
||
if let Err(error) = remove_if_present(&partial).await {
|
||
return engine.fail("delete that share's partial download", error);
|
||
}
|
||
if let Err(error) = remove_if_present(&marker).await {
|
||
return engine.fail("delete that share's partial marker", error);
|
||
}
|
||
}
|
||
if let Err(error) = engine.catalog.lock().remove_share(&id) {
|
||
return engine.fail("drop that share's catalog rows", error);
|
||
}
|
||
if let Err(error) = engine.store.remove(&id) {
|
||
return engine.fail("remove that share", error);
|
||
}
|
||
{
|
||
let mut downloads = engine.downloads.lock();
|
||
downloads.items.retain(|key, _| key.share_id != id);
|
||
downloads.jobs.retain(|key, _| key.share_id != id);
|
||
}
|
||
engine.emit_downloads();
|
||
engine.load_shares();
|
||
// Its clients are stale now; the next action reopens what is left.
|
||
engine.connection.lock().take();
|
||
engine.load_library();
|
||
engine.emit(Event::Status("removed".into()));
|
||
});
|
||
}
|
||
|
||
// ------------------------------------------------------------ connection
|
||
|
||
/// The open connection, if there is one. Never held across an await — the
|
||
/// guard is `parking_lot`'s.
|
||
fn opened(&self) -> Option<(Arc<SharedLibrary>, StreamSource)> {
|
||
let connection = self.connection.lock();
|
||
let connection = connection.as_ref()?;
|
||
Some((Arc::clone(&connection.library), connection.source.clone()))
|
||
}
|
||
|
||
/// Open every configured share and build the block layer.
|
||
///
|
||
/// Idempotent: a second call while one connection exists is a no-op, so the
|
||
/// UI can call it whenever it needs the network without tracking whether it
|
||
/// already has it.
|
||
pub fn connect(&self) {
|
||
if self.opened().is_some() {
|
||
return;
|
||
}
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
engine.ensure_connected().await;
|
||
});
|
||
}
|
||
|
||
/// Open the shares if they are not open. Returns the pieces playback needs.
|
||
async fn ensure_connected(&self) -> Option<(Arc<SharedLibrary>, StreamSource)> {
|
||
if let Some(opened) = self.opened() {
|
||
return Some(opened);
|
||
}
|
||
|
||
// One handshake at a time, and re-check after waiting: whoever was
|
||
// ahead in the queue has almost certainly just done it.
|
||
let _connecting = self.connecting.lock().await;
|
||
if let Some(opened) = self.opened() {
|
||
return Some(opened);
|
||
}
|
||
|
||
let (library, failures) = match SharedLibrary::open_all(&self.store).await {
|
||
Ok(opened) => opened,
|
||
Err(error) => {
|
||
self.emit(Event::ConnectFailed(error.to_string()));
|
||
return None;
|
||
}
|
||
};
|
||
let library = Arc::new(library);
|
||
|
||
let opener = Arc::new(LibraryOpener::new(Arc::clone(&library)));
|
||
let config = StreamConfig::default()
|
||
// mpv brings its own read-ahead; a second one under it only competes
|
||
// for the bandwidth mpv is blocked on. See `pstr_player::READAHEAD_BLOCKS`.
|
||
.with_readahead(pstr_player::READAHEAD_BLOCKS)
|
||
.with_disk_cache(DiskCacheConfig::new(self.dirs.block_cache()));
|
||
|
||
let source = match StreamSource::new(opener, config).await {
|
||
Ok(source) => source,
|
||
Err(error) => {
|
||
self.emit(Event::ConnectFailed(error.to_string()));
|
||
return None;
|
||
}
|
||
};
|
||
|
||
*self.connection.lock() = Some(Connection {
|
||
library: Arc::clone(&library),
|
||
source: source.clone(),
|
||
});
|
||
|
||
self.emit(Event::Connected {
|
||
failures: failures
|
||
.into_iter()
|
||
.map(|(share, error)| (share.name, error.to_string()))
|
||
.collect(),
|
||
});
|
||
Some((library, source))
|
||
}
|
||
|
||
// ------------------------------------------------------------- catalogue
|
||
|
||
/// Re-read the catalog into a [`Library`].
|
||
pub fn load_library(&self) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
let result = {
|
||
let catalog = engine.catalog.lock();
|
||
catalog
|
||
.all_files()
|
||
.and_then(|files| Ok((files, catalog.all_watch_states()?)))
|
||
};
|
||
match result {
|
||
Ok((files, watch)) => {
|
||
let library = Library::build(files, &watch);
|
||
let targets = download_targets(&library);
|
||
engine.emit(Event::LibraryLoaded(library));
|
||
engine.refresh_offline(Some(targets));
|
||
}
|
||
Err(error) => engine.fail("read the catalog", error),
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Refresh the lightweight offline index used by title badges and actions.
|
||
pub fn load_offline_files(&self) {
|
||
let engine = self.clone();
|
||
self.runtime
|
||
.spawn_blocking(move || engine.refresh_offline(None));
|
||
}
|
||
|
||
/// The offline index, against `targets` when the caller has just built a
|
||
/// library to take them from — otherwise one is built here, which on a
|
||
/// large catalog is most of what this costs.
|
||
fn refresh_offline(&self, targets: Option<HashMap<DownloadKey, PlaybackTarget>>) {
|
||
let result = (|| {
|
||
let catalog = self.catalog.lock();
|
||
let files = catalog.all_offline_files()?;
|
||
let targets = match targets {
|
||
Some(targets) => targets,
|
||
None => download_targets(&Library::build(
|
||
catalog.all_files()?,
|
||
&catalog.all_watch_states()?,
|
||
)),
|
||
};
|
||
let mut valid = HashSet::new();
|
||
let mut hydrated = Vec::new();
|
||
for ((share_id, link_id), file) in files {
|
||
let key = DownloadKey { share_id, link_id };
|
||
let path = self
|
||
.dirs
|
||
.offline_file(&key.share_id, &key.link_id, &file.revision_id);
|
||
let Some(expected) = file
|
||
.block_sizes
|
||
.iter()
|
||
.try_fold(0_u64, |sum, size| sum.checked_add(*size))
|
||
else {
|
||
catalog.remove_offline_file(&key.share_id, &key.link_id)?;
|
||
continue;
|
||
};
|
||
if std::fs::metadata(path).is_ok_and(|metadata| metadata.len() == expected) {
|
||
valid.insert(key.clone());
|
||
if let Some(target) = targets.get(&key).cloned() {
|
||
hydrated.push(DownloadItem {
|
||
key,
|
||
target,
|
||
state: DownloadState::Completed,
|
||
downloaded: expected,
|
||
total: expected,
|
||
rate: 0.0,
|
||
});
|
||
}
|
||
} else {
|
||
// The catalog is an index, not the bytes. A manually
|
||
// removed or truncated file must stop wearing an
|
||
// offline badge and remain playable from the share.
|
||
catalog.remove_offline_file(&key.share_id, &key.link_id)?;
|
||
}
|
||
}
|
||
Ok::<_, pstr_core::Error>((valid, hydrated))
|
||
})();
|
||
match result {
|
||
Ok((files, hydrated)) => {
|
||
{
|
||
let mut downloads = self.downloads.lock();
|
||
for item in hydrated {
|
||
downloads.items.entry(item.key.clone()).or_insert(item);
|
||
}
|
||
}
|
||
self.hydrate_partial_downloads();
|
||
self.emit_downloads();
|
||
self.emit(Event::OfflineFiles(files));
|
||
}
|
||
Err(error) => self.fail("read offline downloads", error),
|
||
}
|
||
}
|
||
|
||
/// Crawl one share, or all of them, and reload the library.
|
||
pub fn crawl(&self, share: Option<String>) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
engine.connect_and_crawl(share).await;
|
||
});
|
||
}
|
||
|
||
async fn connect_and_crawl(&self, share: Option<String>) {
|
||
let Some((library, _)) = self.ensure_connected().await else {
|
||
self.emit(Event::CrawlFinished);
|
||
return;
|
||
};
|
||
|
||
let targets: Vec<String> = match share {
|
||
Some(id) => vec![id],
|
||
None => library.share_ids().map(str::to_string).collect(),
|
||
};
|
||
|
||
// Every share at once: each is its own session against its own
|
||
// tree, and one large share no longer holds the rest back behind it.
|
||
let cancel = self.crawl_cancel.subscribe();
|
||
futures::future::join_all(
|
||
targets
|
||
.into_iter()
|
||
.map(|share_id| self.crawl_share(&library, share_id, cancel.clone())),
|
||
)
|
||
.await;
|
||
self.emit(Event::CrawlFinished);
|
||
}
|
||
|
||
/// Stop every crawl still listing. Shares already past their listing
|
||
/// finish storing what they found.
|
||
pub fn stop_crawl(&self) {
|
||
self.crawl_cancel.send_modify(|generation| *generation += 1);
|
||
}
|
||
|
||
/// Crawl one share and store what it holds.
|
||
async fn crawl_share(
|
||
&self,
|
||
library: &SharedLibrary,
|
||
share_id: String,
|
||
mut cancel: tokio::sync::watch::Receiver<u64>,
|
||
) {
|
||
let started = std::time::Instant::now();
|
||
let name = self
|
||
.store
|
||
.list()
|
||
.ok()
|
||
.and_then(|shares| shares.into_iter().find(|share| share.id == share_id))
|
||
.map_or_else(|| share_id.clone(), |share| share.name);
|
||
self.emit(Event::CrawlProgress {
|
||
share: name.clone(),
|
||
found: 0,
|
||
});
|
||
// At most a few times a second: a wide level comes back as a burst
|
||
// of listings, and a frame per listing is a frame per folder.
|
||
let last = parking_lot::Mutex::new(std::time::Instant::now());
|
||
let report = |found: usize| {
|
||
let mut last = last.lock();
|
||
if last.elapsed() >= std::time::Duration::from_millis(250) {
|
||
*last = std::time::Instant::now();
|
||
self.emit(Event::CrawlProgress {
|
||
share: name.clone(),
|
||
found,
|
||
});
|
||
}
|
||
};
|
||
// Only the listing can be stopped. Past it the crawl is replacing the
|
||
// share's rows and its stale downloads, and stopping halfway through
|
||
// that would leave the two disagreeing.
|
||
let nodes = tokio::select! {
|
||
result = library.crawl_reporting(&share_id, report) => match result {
|
||
Ok(nodes) => nodes,
|
||
Err(error) => {
|
||
self.fail(&format!("crawl {share_id}"), error);
|
||
return;
|
||
}
|
||
},
|
||
Ok(()) = cancel.changed() => {
|
||
self.emit(Event::CrawlStopped { share_id });
|
||
return;
|
||
}
|
||
};
|
||
let rows = build_rows(&share_id, &nodes);
|
||
let files = rows.len();
|
||
|
||
// A writer opened before the crawl may be producing the revision
|
||
// this crawl is about to supersede. Quiesce all writers for the
|
||
// share before inspecting or deleting any revision paths.
|
||
let active_downloads: Vec<_> = self
|
||
.downloads
|
||
.lock()
|
||
.jobs
|
||
.keys()
|
||
.filter(|key| key.share_id == share_id)
|
||
.cloned()
|
||
.collect();
|
||
for key in &active_downloads {
|
||
self.cancel_and_join(key).await;
|
||
}
|
||
|
||
// Paths are derived from the old revision id, so inspect and
|
||
// remove stale bytes before changing or dropping their index rows.
|
||
let retained = match self.catalog.lock().offline_files_for_share(&share_id) {
|
||
Ok(files) => files,
|
||
Err(error) => {
|
||
self.fail(&format!("inspect downloads for {share_id}"), error);
|
||
return;
|
||
}
|
||
};
|
||
let revisions: HashMap<&str, Option<&str>> = rows
|
||
.iter()
|
||
.map(|row| (row.link_id.as_str(), row.active_revision_id.as_deref()))
|
||
.collect();
|
||
let stale: Vec<_> = retained
|
||
.iter()
|
||
.filter(|(link_id, file)| {
|
||
revisions.get(link_id.as_str()).copied().flatten()
|
||
!= Some(file.revision_id.as_str())
|
||
})
|
||
.map(|(link_id, file)| (link_id.clone(), file.clone()))
|
||
.collect();
|
||
let mut cleanup_failed = false;
|
||
for (link_id, file) in &stale {
|
||
let path = self
|
||
.dirs
|
||
.offline_file(&share_id, link_id, &file.revision_id);
|
||
if let Err(error) = remove_if_present(&path).await {
|
||
self.fail(&format!("delete stale download for {share_id}"), error);
|
||
cleanup_failed = true;
|
||
break;
|
||
}
|
||
}
|
||
if cleanup_failed {
|
||
return;
|
||
}
|
||
|
||
let stored = {
|
||
let mut catalog = self.catalog.lock();
|
||
catalog
|
||
.replace_share_retaining_offline(&share_id, &rows)
|
||
.and_then(|()| {
|
||
for (link_id, _) in &stale {
|
||
catalog.remove_offline_file(&share_id, link_id)?;
|
||
}
|
||
Ok(())
|
||
})
|
||
};
|
||
if let Err(error) = stored {
|
||
self.fail(&format!("store {share_id}"), error);
|
||
return;
|
||
}
|
||
|
||
self.emit(Event::Crawled {
|
||
share_id,
|
||
nodes: nodes.len(),
|
||
files,
|
||
seconds: started.elapsed().as_secs_f64(),
|
||
});
|
||
self.load_library();
|
||
}
|
||
|
||
/// Record where playback got to.
|
||
///
|
||
/// Fire and forget: a watch position that fails to save is worth a log line,
|
||
/// not an interruption to what the viewer is watching.
|
||
pub fn save_watch_state(&self, share_id: String, link_id: String, state: WatchState) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
engine.save_watch_state_now(&share_id, &link_id, &state);
|
||
});
|
||
}
|
||
|
||
/// The same write, on the calling thread.
|
||
///
|
||
/// What the player thread uses for its last save: a task spawned as the
|
||
/// window is closing may never be polled, and the position at the moment
|
||
/// the viewer quit is exactly the one worth keeping.
|
||
pub fn save_watch_state_now(&self, share_id: &str, link_id: &str, state: &WatchState) {
|
||
if let Err(error) = self
|
||
.catalog
|
||
.lock()
|
||
.set_watch_state(share_id, link_id, state)
|
||
{
|
||
tracing::warn!("save watch state for {link_id}: {error}");
|
||
return;
|
||
}
|
||
self.emit(Event::Watched {
|
||
share_id: share_id.to_owned(),
|
||
link_id: link_id.to_owned(),
|
||
state: *state,
|
||
});
|
||
}
|
||
|
||
pub fn title_track_prefs(&self, key: &str) -> Option<TitleTrackPrefs> {
|
||
self.track_prefs.lock().get(key).cloned()
|
||
}
|
||
pub fn set_title_track_prefs(&self, key: String, prefs: TitleTrackPrefs) {
|
||
self.track_prefs.lock().insert(key.clone(), prefs.clone());
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
if let Err(e) = engine.catalog.lock().set_title_track_prefs(&key, &prefs) {
|
||
tracing::warn!("save show track preferences: {e}");
|
||
}
|
||
});
|
||
}
|
||
|
||
// ------------------------------------------------------------- playback
|
||
|
||
/// Open a stream on `node` and hand it back through
|
||
/// [`Event::PlaybackReady`].
|
||
///
|
||
/// Opening is done here rather than in mpv's open callback for the reason
|
||
/// `pstr-player` documents: opening can take seconds and fail for reasons a
|
||
/// person needs to read, and mpv's callback can only say "loading failed".
|
||
pub fn play(&self, target: PlaybackTarget) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
// A completed local copy is authoritative until the catalog sees a
|
||
// new revision; opening it never needs a public-link session.
|
||
let offline = {
|
||
engine
|
||
.catalog
|
||
.lock()
|
||
.offline_file(&target.share_id, &target.link_id)
|
||
.ok()
|
||
.flatten()
|
||
};
|
||
if let Some(file) = offline {
|
||
let path =
|
||
engine
|
||
.dirs
|
||
.offline_file(&target.share_id, &target.link_id, &file.revision_id);
|
||
if tokio::fs::metadata(&path)
|
||
.await
|
||
.map(|m| m.len() == file.block_sizes.iter().sum::<u64>())
|
||
.unwrap_or(false)
|
||
{
|
||
let blocks: std::sync::Arc<dyn BlockSource> = std::sync::Arc::new(
|
||
FileBlocks::new(file.revision_id, path, file.block_sizes),
|
||
);
|
||
let uid = node_uid(&target.volume_id, &target.link_id);
|
||
engine.emit(Event::PlaybackReady {
|
||
target: Box::new(target),
|
||
stream: VideoStream::offline(uid, blocks, pstr_stream::DEFAULT_RING_BYTES),
|
||
});
|
||
return;
|
||
}
|
||
}
|
||
let Some((_, source)) = engine.ensure_connected().await else {
|
||
return;
|
||
};
|
||
let uid = node_uid(&target.volume_id, &target.link_id);
|
||
match source.open(&target.share_id, &uid).await {
|
||
Ok(stream) if stream.size() == 0 => {
|
||
engine.emit(Event::Error(format!("{} has no content", target.name)));
|
||
}
|
||
Ok(stream) => {
|
||
// Before mpv is even built: building it takes the UI thread
|
||
// a moment, and the network can spend that moment on the
|
||
// two reads every file opens with.
|
||
stream.warm_edges();
|
||
engine.emit(Event::PlaybackReady {
|
||
target: Box::new(target),
|
||
stream,
|
||
})
|
||
}
|
||
Err(error) => engine.fail(&format!("open {}", target.name), error),
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Fetch the blocks a file opens with into the cache, without playing it.
|
||
///
|
||
/// For the episode after the one playing, so that autoplay — or a click on
|
||
/// "Play now" — starts from the cache rather than from two round trips. A
|
||
/// local copy needs none of this, and a failure here costs nothing but the
|
||
/// head start, so both are quietly skipped.
|
||
pub fn warm(&self, target: &PlaybackTarget) {
|
||
let engine = self.clone();
|
||
let share_id = target.share_id.clone();
|
||
let link_id = target.link_id.clone();
|
||
let uid = node_uid(&target.volume_id, &target.link_id);
|
||
self.runtime.spawn(async move {
|
||
let offline = engine
|
||
.catalog
|
||
.lock()
|
||
.offline_file(&share_id, &link_id)
|
||
.ok()
|
||
.flatten()
|
||
.is_some();
|
||
if offline {
|
||
return;
|
||
}
|
||
let Some(source) = engine.opened().map(|(_, source)| source) else {
|
||
return;
|
||
};
|
||
match source.open(&share_id, &uid).await {
|
||
Ok(stream) => stream.warm_edges(),
|
||
Err(error) => tracing::debug!(%error, "warming the next episode failed"),
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Queue complete plaintext copies. Jobs for the same episode are
|
||
/// de-duplicated; a failed or cancelled row is a resumable retry.
|
||
pub fn make_offline(&self, targets: Vec<PlaybackTarget>) {
|
||
for target in targets {
|
||
let key = DownloadKey::from(&target);
|
||
if !self.job_is_live(&key) {
|
||
self.spawn_download(target);
|
||
}
|
||
}
|
||
}
|
||
|
||
fn job_is_live(&self, key: &DownloadKey) -> bool {
|
||
self.downloads
|
||
.lock()
|
||
.jobs
|
||
.get(key)
|
||
.is_some_and(|job| job.live.load(Ordering::Acquire))
|
||
}
|
||
|
||
fn spawn_download(&self, target: PlaybackTarget) {
|
||
let key = DownloadKey::from(&target);
|
||
let control = Arc::new(DownloadControl::new());
|
||
let live = Arc::new(AtomicBool::new(true));
|
||
{
|
||
let mut downloads = self.downloads.lock();
|
||
let previous = downloads.items.get(&key).cloned();
|
||
downloads.items.insert(
|
||
key.clone(),
|
||
DownloadItem {
|
||
key: key.clone(),
|
||
target: target.clone(),
|
||
state: DownloadState::Queued,
|
||
downloaded: previous.as_ref().map_or(0, |item| item.downloaded),
|
||
total: previous.as_ref().map_or(0, |item| item.total),
|
||
rate: 0.0,
|
||
},
|
||
);
|
||
}
|
||
self.emit_downloads();
|
||
let engine = self.clone();
|
||
let task_live = Arc::clone(&live);
|
||
let task_target = target.clone();
|
||
let task_control = Arc::clone(&control);
|
||
let handle = self.runtime.spawn(async move {
|
||
let result = engine.download_one(task_target.clone(), task_control).await;
|
||
task_live.store(false, Ordering::Release);
|
||
let task_key = DownloadKey::from(&task_target);
|
||
match result {
|
||
Ok(DownloadEnd::Completed) => {
|
||
engine.update_download(&task_key, |item| {
|
||
item.downloaded = item.total;
|
||
item.state = DownloadState::Completed;
|
||
});
|
||
engine.load_offline_files();
|
||
engine.emit(Event::Status(format!(
|
||
"{} is available offline",
|
||
task_target.name
|
||
)));
|
||
}
|
||
Ok(DownloadEnd::Cancelled) => {
|
||
engine.update_download(&task_key, |item| item.state = DownloadState::Cancelled);
|
||
}
|
||
Err(error) => {
|
||
engine.update_download(&task_key, |item| {
|
||
item.state = DownloadState::Failed(error.to_string());
|
||
});
|
||
}
|
||
}
|
||
});
|
||
self.downloads.lock().jobs.insert(
|
||
key,
|
||
DownloadJob {
|
||
control,
|
||
handle,
|
||
live,
|
||
},
|
||
);
|
||
}
|
||
|
||
pub fn pause_download(&self, key: &DownloadKey) {
|
||
if let Some(control) = self.live_control(key) {
|
||
control.set(DownloadCommand::Pause);
|
||
self.update_download(key, |item| item.state = DownloadState::Paused);
|
||
}
|
||
}
|
||
|
||
pub fn resume_download(&self, key: &DownloadKey) {
|
||
let control = self.live_control(key);
|
||
if let Some(control) = control {
|
||
control.set(DownloadCommand::Run);
|
||
self.update_download(key, |item| item.state = DownloadState::Running);
|
||
return;
|
||
}
|
||
if let Some(target) = self
|
||
.downloads
|
||
.lock()
|
||
.items
|
||
.get(key)
|
||
.map(|item| item.target.clone())
|
||
{
|
||
self.make_offline(vec![target]);
|
||
}
|
||
}
|
||
|
||
/// Stop network work after the current complete block. The partial remains
|
||
/// valid and Resume starts at its next block.
|
||
pub fn cancel_download(&self, key: &DownloadKey) {
|
||
if let Some(control) = self.live_control(key) {
|
||
control.set(DownloadCommand::Cancel);
|
||
}
|
||
}
|
||
|
||
fn live_control(&self, key: &DownloadKey) -> Option<Arc<DownloadControl>> {
|
||
self.downloads
|
||
.lock()
|
||
.jobs
|
||
.get(key)
|
||
.filter(|job| job.live.load(Ordering::Acquire))
|
||
.map(|job| Arc::clone(&job.control))
|
||
}
|
||
|
||
/// Remove completed offline bytes and the catalog record. With
|
||
/// `remove_partial`, also discard resumable work after cancelling it.
|
||
pub fn remove_download(&self, key: DownloadKey, remove_partial: bool) {
|
||
self.cancel_download(&key);
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
engine.cancel_and_join(&key).await;
|
||
let offline = {
|
||
engine
|
||
.catalog
|
||
.lock()
|
||
.offline_file(&key.share_id, &key.link_id)
|
||
.ok()
|
||
.flatten()
|
||
};
|
||
if let Some(file) = offline {
|
||
let path = engine
|
||
.dirs
|
||
.offline_file(&key.share_id, &key.link_id, &file.revision_id);
|
||
if let Err(error) = remove_if_present(&path).await {
|
||
return engine.fail("delete offline bytes", error);
|
||
}
|
||
}
|
||
if remove_partial {
|
||
let (partial, marker) = engine.partial_paths(&key);
|
||
if let Err(error) = remove_if_present(&partial).await {
|
||
return engine.fail("delete partial download", error);
|
||
}
|
||
if let Err(error) = remove_if_present(&marker).await {
|
||
return engine.fail("delete partial marker", error);
|
||
}
|
||
}
|
||
if let Err(error) = engine
|
||
.catalog
|
||
.lock()
|
||
.remove_offline_file(&key.share_id, &key.link_id)
|
||
{
|
||
return engine.fail("make online-only", error);
|
||
}
|
||
{
|
||
let mut downloads = engine.downloads.lock();
|
||
downloads.jobs.remove(&key);
|
||
downloads.items.remove(&key);
|
||
}
|
||
engine.emit_downloads();
|
||
engine.load_offline_files();
|
||
engine.emit(Event::Status(
|
||
"download removed; online source was kept".to_owned(),
|
||
));
|
||
});
|
||
}
|
||
|
||
async fn cancel_and_join(&self, key: &DownloadKey) {
|
||
let job = self.downloads.lock().jobs.remove(key);
|
||
if let Some(job) = job {
|
||
job.control.set(DownloadCommand::Cancel);
|
||
if let Err(error) = job.handle.await
|
||
&& !error.is_cancelled()
|
||
{
|
||
tracing::warn!("join offline writer: {error}");
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Ask every writer to stop at its next durable block boundary and wait a
|
||
/// bounded time. If a network read outlives the bound, the marker and last
|
||
/// `sync_data` boundary remain a coherent handoff for the next launch.
|
||
pub fn shutdown_downloads(&self, timeout: std::time::Duration) {
|
||
let jobs: Vec<_> = {
|
||
let mut downloads = self.downloads.lock();
|
||
downloads
|
||
.jobs
|
||
.drain()
|
||
.map(|(_, job)| {
|
||
job.control.set(DownloadCommand::Cancel);
|
||
job.handle
|
||
})
|
||
.collect()
|
||
};
|
||
self.runtime.block_on(async move {
|
||
let wait = async move {
|
||
for handle in jobs {
|
||
let _ = handle.await;
|
||
}
|
||
};
|
||
let _ = tokio::time::timeout(timeout, wait).await;
|
||
});
|
||
}
|
||
|
||
fn emit_downloads(&self) {
|
||
let mut items: Vec<_> = self.downloads.lock().items.values().cloned().collect();
|
||
items.sort_by(|a, b| {
|
||
a.target
|
||
.title_key
|
||
.cmp(&b.target.title_key)
|
||
.then(a.target.season.cmp(&b.target.season))
|
||
.then(a.target.number.cmp(&b.target.number))
|
||
});
|
||
self.emit(Event::Downloads(items));
|
||
}
|
||
|
||
fn update_download(&self, key: &DownloadKey, update: impl FnOnce(&mut DownloadItem)) {
|
||
if let Some(item) = self.downloads.lock().items.get_mut(key) {
|
||
update(item);
|
||
}
|
||
self.emit_downloads();
|
||
}
|
||
|
||
/// [`Self::update_download`] for a change to one row only: the row goes
|
||
/// to the UI on its own rather than in a fresh snapshot of every row.
|
||
fn progress_download(&self, key: &DownloadKey, update: impl FnOnce(&mut DownloadItem)) {
|
||
let item = {
|
||
let mut downloads = self.downloads.lock();
|
||
let Some(item) = downloads.items.get_mut(key) else {
|
||
return;
|
||
};
|
||
update(item);
|
||
item.clone()
|
||
};
|
||
self.emit(Event::DownloadProgress(Box::new(item)));
|
||
}
|
||
|
||
fn partial_paths(&self, key: &DownloadKey) -> (std::path::PathBuf, std::path::PathBuf) {
|
||
let partial = self
|
||
.dirs
|
||
.offline_file(&key.share_id, &key.link_id, "partial");
|
||
let marker = partial.with_extension("partial.json");
|
||
(partial, marker)
|
||
}
|
||
|
||
fn hydrate_partial_downloads(&self) {
|
||
let Ok(entries) = std::fs::read_dir(self.dirs.offline_content()) else {
|
||
return;
|
||
};
|
||
let mut found = Vec::new();
|
||
for entry in entries.flatten() {
|
||
let path = entry.path();
|
||
if !path
|
||
.file_name()
|
||
.and_then(|name| name.to_str())
|
||
.is_some_and(|name| name.ends_with(".partial.json"))
|
||
{
|
||
continue;
|
||
}
|
||
let Some(marker) = std::fs::read(&path)
|
||
.ok()
|
||
.and_then(|bytes| serde_json::from_slice::<PartialMarker>(&bytes).ok())
|
||
else {
|
||
continue;
|
||
};
|
||
let Some(target) = marker.target else {
|
||
continue;
|
||
};
|
||
let key = DownloadKey::from(&target);
|
||
let (partial, expected_marker) = self.partial_paths(&key);
|
||
if expected_marker != path {
|
||
continue;
|
||
}
|
||
let existing = std::fs::metadata(partial).map_or(0, |metadata| metadata.len());
|
||
let (_, downloaded) = resume_position(existing, &marker.block_sizes);
|
||
let Some(total) = marker
|
||
.block_sizes
|
||
.iter()
|
||
.try_fold(0_u64, |sum, size| sum.checked_add(*size))
|
||
else {
|
||
continue;
|
||
};
|
||
found.push(DownloadItem {
|
||
key,
|
||
target,
|
||
state: DownloadState::Cancelled,
|
||
downloaded,
|
||
total,
|
||
rate: 0.0,
|
||
});
|
||
}
|
||
let mut downloads = self.downloads.lock();
|
||
for item in found {
|
||
downloads.items.entry(item.key.clone()).or_insert(item);
|
||
}
|
||
}
|
||
|
||
async fn download_one(
|
||
&self,
|
||
target: PlaybackTarget,
|
||
control: Arc<DownloadControl>,
|
||
) -> anyhow::Result<DownloadEnd> {
|
||
use tokio::io::{AsyncSeekExt, AsyncWriteExt};
|
||
|
||
let key = DownloadKey::from(&target);
|
||
let mut command = control.subscribe();
|
||
if !wait_until_runnable(&mut command).await {
|
||
return Ok(DownloadEnd::Cancelled);
|
||
}
|
||
|
||
let _permit = loop {
|
||
self.update_download(&key, |item| item.state = DownloadState::Queued);
|
||
let permits = Arc::clone(&self.download_permits);
|
||
tokio::select! {
|
||
permit = permits.acquire_owned() => break permit?,
|
||
changed = command.changed() => {
|
||
if changed.is_err() || !wait_until_runnable(&mut command).await {
|
||
return Ok(DownloadEnd::Cancelled);
|
||
}
|
||
}
|
||
}
|
||
};
|
||
self.update_download(&key, |item| item.state = DownloadState::Running);
|
||
let Some((_, source)) = self.ensure_connected().await else {
|
||
anyhow::bail!("share is not connected");
|
||
};
|
||
let stream = source
|
||
.open_for_copy(
|
||
&target.share_id,
|
||
&node_uid(&target.volume_id, &target.link_id),
|
||
)
|
||
.await?;
|
||
let revision_id = stream.revision_id().to_owned();
|
||
let block_sizes = stream.block_sizes().to_vec();
|
||
let total = block_sizes
|
||
.iter()
|
||
.try_fold(0_u64, |sum, size| sum.checked_add(*size))
|
||
.ok_or_else(|| anyhow::anyhow!("offline file size overflow"))?;
|
||
self.update_download(&key, |item| item.total = total);
|
||
|
||
let destination = self
|
||
.dirs
|
||
.offline_file(&target.share_id, &target.link_id, &revision_id);
|
||
let parent = destination
|
||
.parent()
|
||
.ok_or_else(|| anyhow::anyhow!("offline file has no parent"))?;
|
||
tokio::fs::create_dir_all(parent).await?;
|
||
if tokio::fs::metadata(&destination)
|
||
.await
|
||
.is_ok_and(|metadata| metadata.len() == total)
|
||
{
|
||
let (partial, marker) = self.partial_paths(&key);
|
||
remove_if_present(&partial).await?;
|
||
self.finish_download(&key, &target, revision_id, block_sizes, total, &marker)
|
||
.await?;
|
||
return Ok(DownloadEnd::Completed);
|
||
}
|
||
|
||
let (partial, marker_path) = self.partial_paths(&key);
|
||
let marker = PartialMarker {
|
||
revision_id: revision_id.clone(),
|
||
block_sizes: block_sizes.clone(),
|
||
target: Some(target.clone()),
|
||
};
|
||
let matches = tokio::fs::read(&marker_path)
|
||
.await
|
||
.ok()
|
||
.and_then(|bytes| serde_json::from_slice::<PartialMarker>(&bytes).ok())
|
||
.is_some_and(|stored| {
|
||
stored.revision_id == revision_id && stored.block_sizes == block_sizes
|
||
});
|
||
if !matches {
|
||
remove_if_present(&partial).await?;
|
||
// The old marker describes bytes we just discarded. Remove it
|
||
// before publishing the new journal so rename never relies on
|
||
// platform-specific replace-existing behaviour.
|
||
remove_if_present(&marker_path).await?;
|
||
write_marker_atomically(&marker_path, &marker).await?;
|
||
}
|
||
|
||
let existing = tokio::fs::metadata(&partial)
|
||
.await
|
||
.map(|metadata| metadata.len())
|
||
.unwrap_or(0);
|
||
let (mut block_index, mut offset) = resume_position(existing, &block_sizes);
|
||
let mut output = tokio::fs::OpenOptions::new()
|
||
.create(true)
|
||
.truncate(false)
|
||
.read(true)
|
||
.write(true)
|
||
.open(&partial)
|
||
.await?;
|
||
output.set_len(offset).await?;
|
||
output.seek(std::io::SeekFrom::Start(offset)).await?;
|
||
self.update_download(&key, |item| item.downloaded = offset);
|
||
|
||
use futures::TryStreamExt as _;
|
||
let mut fetches = None;
|
||
let mut last_snapshot = std::time::Instant::now();
|
||
let mut snapshot_bytes = offset;
|
||
while block_index < block_sizes.len() {
|
||
// Dropped before a pause rather than held across it: a fetch left
|
||
// half-done for as long as the viewer stays paused comes back as a
|
||
// timed-out request, and a timed-out request fails the download.
|
||
if matches!(*command.borrow(), DownloadCommand::Pause) {
|
||
fetches = None;
|
||
}
|
||
if !wait_until_runnable(&mut command).await {
|
||
output.sync_all().await?;
|
||
return Ok(DownloadEnd::Cancelled);
|
||
}
|
||
let size = block_sizes[block_index];
|
||
let pending =
|
||
fetches.get_or_insert_with(|| fetch_in_order(&stream, &block_sizes, block_index));
|
||
let Some(bytes) = pending.try_next().await? else {
|
||
anyhow::bail!("block {block_index} never arrived");
|
||
};
|
||
if bytes.len() as u64 != size {
|
||
anyhow::bail!(
|
||
"short block {block_index}: received {}, expected {size}",
|
||
bytes.len()
|
||
);
|
||
}
|
||
output.write_all(&bytes).await?;
|
||
// The reported boundary must survive a crash; otherwise the next
|
||
// run could trust a file length whose trailing block never reached
|
||
// storage.
|
||
output.sync_data().await?;
|
||
offset += size;
|
||
block_index += 1;
|
||
if last_snapshot.elapsed() >= std::time::Duration::from_millis(100)
|
||
|| block_index == block_sizes.len()
|
||
{
|
||
let seconds = last_snapshot.elapsed().as_secs_f64();
|
||
let instant = (offset - snapshot_bytes) as f64 / seconds.max(f64::EPSILON);
|
||
self.progress_download(&key, |item| {
|
||
item.state = DownloadState::Running;
|
||
item.downloaded = offset;
|
||
// Smoothed, or the time left jumps about with every block
|
||
// that took a little longer than the last.
|
||
item.rate = if item.rate > 0.0 {
|
||
0.8 * item.rate + 0.2 * instant
|
||
} else {
|
||
instant
|
||
};
|
||
});
|
||
last_snapshot = std::time::Instant::now();
|
||
snapshot_bytes = offset;
|
||
}
|
||
}
|
||
output.sync_all().await?;
|
||
drop(output);
|
||
if !wait_until_runnable(&mut command).await {
|
||
return Ok(DownloadEnd::Cancelled);
|
||
}
|
||
remove_if_present(&destination).await?;
|
||
tokio::fs::rename(&partial, &destination).await?;
|
||
sync_parent(&destination).await?;
|
||
self.finish_download(&key, &target, revision_id, block_sizes, total, &marker_path)
|
||
.await?;
|
||
Ok(DownloadEnd::Completed)
|
||
}
|
||
|
||
async fn finish_download(
|
||
&self,
|
||
key: &DownloadKey,
|
||
target: &PlaybackTarget,
|
||
revision_id: String,
|
||
block_sizes: Vec<u64>,
|
||
total: u64,
|
||
marker_path: &std::path::Path,
|
||
) -> anyhow::Result<()> {
|
||
self.catalog.lock().set_offline_file_durable(
|
||
&target.share_id,
|
||
&target.link_id,
|
||
&OfflineFile {
|
||
revision_id,
|
||
block_sizes,
|
||
},
|
||
)?;
|
||
remove_if_present(marker_path).await?;
|
||
sync_parent(marker_path).await?;
|
||
self.update_download(key, |item| item.total = total);
|
||
Ok(())
|
||
}
|
||
|
||
/// Drop a stream the player is done with, so its reader and blocks are not
|
||
/// held for the rest of the session.
|
||
pub fn release(&self, share_id: String, volume_id: String, link_id: String) {
|
||
if let Some(connection) = self.connection.lock().as_ref() {
|
||
connection
|
||
.source
|
||
.close(&share_id, &node_uid(&volume_id, &link_id));
|
||
}
|
||
}
|
||
|
||
// ------------------------------------------------------------ thumbnails
|
||
|
||
/// Fetch, decode and post a poster for `node`.
|
||
///
|
||
/// The disk cache is checked first and written on success, so a restart
|
||
/// paints the grid without touching the network. A file with no thumbnail
|
||
/// answers [`Event::ThumbnailMissing`] — the grid must remember that, or
|
||
/// every render pays a round-trip per unmatched title.
|
||
pub fn request_thumbnail(&self, node: &CatalogNode) {
|
||
let engine = self.clone();
|
||
let key = thumbnail_key(node);
|
||
let share_id = node.share_id.clone();
|
||
let uid = node_uid(&node.volume_id, &node.link_id);
|
||
let path = self.dirs.thumbnail_cache().join(format!("{key}.bin"));
|
||
|
||
self.runtime.spawn(async move {
|
||
if let Ok(bytes) = tokio::fs::read(&path).await {
|
||
match decode_thumbnail(&bytes) {
|
||
Ok(decoded) => {
|
||
// A cache written before entries were stored shrunk
|
||
// holds the full preview; shrink it once, on its way out.
|
||
if let Some(smaller) = &decoded.smaller {
|
||
store_image(&path, smaller).await;
|
||
}
|
||
return engine.emit(Event::Thumbnail {
|
||
key,
|
||
image: decoded.image,
|
||
});
|
||
}
|
||
// A truncated or corrupt cache entry is not worth reporting;
|
||
// fall through and fetch it again.
|
||
Err(error) => tracing::debug!("cached thumbnail {key}: {error}"),
|
||
}
|
||
}
|
||
|
||
// Past the cache, so this one costs the network. Wait for a permit
|
||
// before taking any of the bandwidth playback might want.
|
||
let permits = Arc::clone(&engine.thumbnails);
|
||
let Ok(_permit) = permits.acquire().await else {
|
||
return;
|
||
};
|
||
|
||
let Some((library, _)) = engine.ensure_connected().await else {
|
||
return;
|
||
};
|
||
let Some(client) = library.client(&share_id) else {
|
||
return engine.emit(Event::ThumbnailMissing { key });
|
||
};
|
||
|
||
// Preview first: it is the one sized for a card. Not every file has
|
||
// one, and the small thumbnail is better than a grey rectangle.
|
||
let mut bytes = None;
|
||
for kind in [ThumbnailType::Preview, ThumbnailType::Thumbnail] {
|
||
match client.download_thumbnail(&uid, kind).await {
|
||
Ok(Some(found)) => {
|
||
bytes = Some(found);
|
||
break;
|
||
}
|
||
Ok(None) => {}
|
||
Err(error) => tracing::debug!("thumbnail {key} ({kind:?}): {error}"),
|
||
}
|
||
}
|
||
let Some(bytes) = bytes else {
|
||
// Common, not exceptional: Proton renders thumbnails on upload,
|
||
// and a share filled by a client that does not attach any — the
|
||
// FUSE client among them — has none at all. The tile falls back
|
||
// to initials until a metadata provider gives it a poster.
|
||
tracing::debug!("no thumbnail for {key}");
|
||
return engine.emit(Event::ThumbnailMissing { key });
|
||
};
|
||
|
||
match decode_thumbnail(&bytes) {
|
||
Ok(decoded) => {
|
||
store_image(&path, decoded.smaller.as_deref().unwrap_or(&bytes)).await;
|
||
engine.emit(Event::Thumbnail {
|
||
key,
|
||
image: decoded.image,
|
||
})
|
||
}
|
||
Err(error) => {
|
||
tracing::debug!("decode thumbnail {key}: {error}");
|
||
engine.emit(Event::ThumbnailMissing { key });
|
||
}
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------- enrichment
|
||
|
||
impl Engine {
|
||
/// The enrichment settings as they stand.
|
||
pub fn metadata_config(&self) -> MetadataConfig {
|
||
self.metadata.lock().config.clone()
|
||
}
|
||
|
||
/// Whether a provider is configured *and* usable right now.
|
||
///
|
||
/// False for "switched off" and for "TMDB with no API key stored" alike —
|
||
/// the UI distinguishes them from the config, not from here.
|
||
pub fn enrichment_ready(&self) -> bool {
|
||
self.metadata.lock().service.is_some()
|
||
}
|
||
|
||
/// Change the settings, write them, and rebuild the provider.
|
||
///
|
||
/// Turning enrichment *off* clears the stored answers as well. That is the
|
||
/// point of the switch: someone who turns this off has decided they would
|
||
/// rather the third party's answers were not on their disk either, and
|
||
/// leaving them there would make "off" mean only "stop asking".
|
||
pub fn set_metadata_config(&self, config: MetadataConfig) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
let previous = engine.metadata_config();
|
||
if let Err(error) = pstr_meta::settings::save(&engine.dirs, &config) {
|
||
return engine.fail("save the metadata settings", error);
|
||
}
|
||
|
||
let forget = !config.enabled || config.provider != previous.provider;
|
||
if forget && let Err(error) = engine.catalog.lock().clear_metadata() {
|
||
tracing::warn!("clear stored metadata: {error}");
|
||
}
|
||
|
||
*engine.metadata.lock() = Enrichment::build(config.clone());
|
||
engine.emit(Event::MetadataConfig(config));
|
||
engine.load_metadata();
|
||
});
|
||
}
|
||
|
||
/// Store an API key for a provider and rebuild against it.
|
||
pub fn set_api_key(&self, provider: ProviderId, key: String) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
if let Err(error) = pstr_meta::settings::set_api_key(provider, &key) {
|
||
return engine.fail("store the API key", error);
|
||
}
|
||
let config = engine.metadata_config();
|
||
*engine.metadata.lock() = Enrichment::build(config);
|
||
engine.emit(Event::Status(if key.trim().is_empty() {
|
||
format!("cleared the {} key", provider.label())
|
||
} else {
|
||
format!("saved the {} key", provider.label())
|
||
}));
|
||
});
|
||
}
|
||
|
||
/// Re-read every stored provider answer.
|
||
///
|
||
/// Read whole rather than per tile, for the same reason watch state is —
|
||
/// see [`Catalog::all_metadata`].
|
||
pub fn load_metadata(&self) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
// One lock, both reads: they are drawn together, and a UI that has
|
||
// the titles but not their episodes for a frame flickers the rows.
|
||
let answers = {
|
||
let catalog = engine.catalog.lock();
|
||
catalog
|
||
.all_metadata()
|
||
.and_then(|records| Ok((records, catalog.all_episode_metadata()?)))
|
||
};
|
||
match answers {
|
||
Ok((records, episodes)) => {
|
||
engine.emit(Event::Metadata(records));
|
||
engine.emit(Event::EpisodeMetadata(episodes));
|
||
}
|
||
Err(error) => engine.fail("read stored metadata", error),
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Look up every title that has no fresh answer.
|
||
///
|
||
/// `force` re-asks about titles that already have one, which is what
|
||
/// "match again" means. Otherwise a run over an already-matched library
|
||
/// makes no requests at all, which is what makes it safe to call whenever
|
||
/// the library changes.
|
||
pub fn match_titles(&self, titles: Vec<Title>, force: bool) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
engine.run_match(titles, force).await;
|
||
engine.emit(Event::MatchFinished);
|
||
});
|
||
}
|
||
|
||
async fn run_match(&self, titles: Vec<Title>, force: bool) {
|
||
let Some(service) = self.metadata.lock().service.clone() else {
|
||
let config = self.metadata_config();
|
||
return self.emit(Event::Error(if !config.enabled {
|
||
"turn on metadata enrichment first".into()
|
||
} else {
|
||
format!("{} needs an API key", config.provider.label())
|
||
}));
|
||
};
|
||
let provider = service.provider();
|
||
|
||
let (stored, listed) = {
|
||
let catalog = self.catalog.lock();
|
||
match catalog
|
||
.all_metadata()
|
||
.and_then(|stored| Ok((stored, catalog.episode_metadata_ages()?)))
|
||
{
|
||
Ok(pair) => pair,
|
||
Err(error) => return self.fail("read stored metadata", error),
|
||
}
|
||
};
|
||
|
||
// Two kinds of work, and the difference is a request saved: a title
|
||
// that already has a good match but no episode list needs the episode
|
||
// request only — searching for it again would ask the provider
|
||
// something it has already answered.
|
||
let mut pending: Vec<Work> = Vec::new();
|
||
for title in titles {
|
||
let record = stored.get(&title.key);
|
||
// A hand-picked entry is never searched for again, not even by
|
||
// "match again" — that button means "the automatic answers are
|
||
// wrong", and re-deciding the one title the viewer already fixed by
|
||
// hand is the opposite of what they asked for. Its episode list is
|
||
// still fetched below if it is missing.
|
||
let pinned = record.is_some_and(|record| record.manual && record.provider == provider);
|
||
if !pinned && (force || !pstr_meta::service::is_usable(record, provider)) {
|
||
pending.push(Work::Match(title));
|
||
continue;
|
||
}
|
||
let has_episodes = listed
|
||
.get(&title.key)
|
||
.is_some_and(|(asked, _)| *asked == provider);
|
||
if let Some(found) = record.and_then(|record| record.metadata.clone())
|
||
&& !has_episodes
|
||
{
|
||
pending.push(Work::Episodes(title, Box::new(found)));
|
||
}
|
||
}
|
||
|
||
if pending.is_empty() {
|
||
return self.emit(Event::Status("everything is already matched".into()));
|
||
}
|
||
self.emit(Event::Status(format!(
|
||
"matching {} titles against {}…",
|
||
pending.len(),
|
||
provider.label()
|
||
)));
|
||
|
||
let (mut matched, mut unmatched, mut failed) = (0usize, 0usize, 0usize);
|
||
let mut lookups = futures::stream::FuturesUnordered::new();
|
||
for work in &pending {
|
||
let service = service.clone();
|
||
let permits = Arc::clone(&self.lookups);
|
||
lookups.push(async move {
|
||
let Ok(_permit) = permits.acquire().await else {
|
||
return None;
|
||
};
|
||
Some(match work {
|
||
Work::Match(title) => match service.record(title).await {
|
||
Ok(record) => {
|
||
// Only for a title that matched: there is no id to
|
||
// ask about otherwise.
|
||
let episodes = match &record.metadata {
|
||
Some(found) => service.title_episodes(title, found).await,
|
||
None => Vec::new(),
|
||
};
|
||
Ok((record.title_key.clone(), Some(record), episodes))
|
||
}
|
||
Err(error) => Err(error),
|
||
},
|
||
Work::Episodes(title, found) => Ok((
|
||
title.key.clone(),
|
||
None,
|
||
service.title_episodes(title, found).await,
|
||
)),
|
||
})
|
||
});
|
||
}
|
||
|
||
use futures::StreamExt as _;
|
||
let mut episodes_found = 0usize;
|
||
while let Some(result) = lookups.next().await {
|
||
match result {
|
||
Some(Ok((title_key, record, episodes))) => {
|
||
if let Some(record) = &record {
|
||
if record.metadata.is_some() {
|
||
matched += 1;
|
||
} else {
|
||
unmatched += 1;
|
||
}
|
||
// Misses are stored on purpose; failures below are not.
|
||
// See `pstr_meta::service`.
|
||
if let Err(error) = self.catalog.lock().set_metadata(record) {
|
||
tracing::warn!("store metadata for {}: {error}", record.title_key);
|
||
}
|
||
}
|
||
if !episodes.is_empty() {
|
||
episodes_found += episodes.len();
|
||
let stored = self.catalog.lock().set_episode_metadata(
|
||
&title_key,
|
||
provider,
|
||
now(),
|
||
&episodes,
|
||
);
|
||
if let Err(error) = stored {
|
||
tracing::warn!("store episodes for {title_key}: {error}");
|
||
}
|
||
}
|
||
}
|
||
Some(Err(error)) => {
|
||
failed += 1;
|
||
tracing::warn!("metadata lookup: {error}");
|
||
}
|
||
None => {}
|
||
}
|
||
}
|
||
|
||
if episodes_found > 0 {
|
||
self.emit(Event::Status(format!("{episodes_found} episodes named")));
|
||
}
|
||
self.emit(Event::Matched {
|
||
matched,
|
||
unmatched,
|
||
failed,
|
||
});
|
||
self.load_metadata();
|
||
}
|
||
|
||
/// Ask the provider what `term` might be, for a viewer choosing by hand.
|
||
///
|
||
/// Unscored and unfiltered — see [`MetadataService::search`]. Nothing is
|
||
/// stored: this is a look, and only [`Engine::choose_match`] writes.
|
||
pub fn search_matches(&self, title_key: String, term: String, kind: TitleKind) {
|
||
let Some(service) = self.metadata.lock().service.clone() else {
|
||
let config = self.metadata_config();
|
||
return self.emit(Event::MatchSearchFailed {
|
||
title_key,
|
||
error: if !config.enabled {
|
||
"turn on metadata enrichment first".into()
|
||
} else {
|
||
format!("{} needs an API key", config.provider.label())
|
||
},
|
||
});
|
||
};
|
||
|
||
let engine = self.clone();
|
||
let permits = Arc::clone(&self.lookups);
|
||
self.runtime.spawn(async move {
|
||
// The same permit the batch run takes: a viewer typing in the search
|
||
// box while a match run is going must not be what earns the 429.
|
||
let Ok(_permit) = permits.acquire().await else {
|
||
return;
|
||
};
|
||
match service.search(&term, kind).await {
|
||
Ok(options) => engine.emit(Event::MatchOptions { title_key, options }),
|
||
Err(error) => {
|
||
tracing::warn!("search {term:?}: {error}");
|
||
engine.emit(Event::MatchSearchFailed {
|
||
title_key,
|
||
error: error.to_string(),
|
||
});
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Pin one title to the entry the viewer picked, and take its episodes.
|
||
///
|
||
/// Stored as [`MetadataRecord::manual`], so it outlives every TTL and
|
||
/// survives "match again". The episode list is fetched here rather than left
|
||
/// to the next match run, because the point of choosing an entry by hand is
|
||
/// usually that its episode names were wrong too.
|
||
pub fn choose_match(&self, title: Title, found: pstr_core::metadata::TitleMetadata) {
|
||
let Some(service) = self.metadata.lock().service.clone() else {
|
||
return self.emit(Event::Error("turn on metadata enrichment first".into()));
|
||
};
|
||
|
||
let engine = self.clone();
|
||
self.runtime.spawn(async move {
|
||
let name = found.name.clone();
|
||
let record = service.chosen(title.key.clone(), found.clone());
|
||
if let Err(error) = engine.catalog.lock().set_metadata(&record) {
|
||
return engine.fail("store the match", error);
|
||
}
|
||
|
||
let episodes = service.title_episodes(&title, &found).await;
|
||
if !episodes.is_empty() {
|
||
let stored = engine.catalog.lock().set_episode_metadata(
|
||
&title.key,
|
||
service.provider(),
|
||
now(),
|
||
&episodes,
|
||
);
|
||
if let Err(error) = stored {
|
||
tracing::warn!("store episodes for {}: {error}", title.key);
|
||
}
|
||
}
|
||
|
||
engine.emit(Event::Status(format!("{} is now {name}", title.name)));
|
||
engine.load_metadata();
|
||
});
|
||
}
|
||
|
||
/// Drop what is stored for one title, so it is matched from scratch again.
|
||
///
|
||
/// The undo for [`Engine::choose_match`], and the repair for a bad automatic
|
||
/// match: the row goes entirely rather than becoming a stored miss, which is
|
||
/// what makes the next match run treat the title as one it has never asked
|
||
/// about.
|
||
pub fn forget_match(&self, title_key: String) {
|
||
let engine = self.clone();
|
||
self.runtime.spawn_blocking(move || {
|
||
if let Err(error) = engine.catalog.lock().forget_metadata(&title_key) {
|
||
return engine.fail("forget the match", error);
|
||
}
|
||
engine.emit(Event::Status("match cleared".into()));
|
||
engine.load_metadata();
|
||
});
|
||
}
|
||
|
||
/// Fetch, decode and post one title's artwork.
|
||
///
|
||
/// Cached on disk under a hash of the URL, so a provider that moves its CDN
|
||
/// paths invalidates by itself and the same picture shared by two titles is
|
||
/// fetched once.
|
||
pub fn request_poster(&self, title_key: String, url: String) {
|
||
let engine = self.clone();
|
||
let path = self
|
||
.dirs
|
||
.poster_cache()
|
||
.join(format!("{}.img", digest(&url)));
|
||
|
||
self.runtime.spawn(async move {
|
||
if let Ok(bytes) = tokio::fs::read(&path).await {
|
||
match decode_thumbnail(&bytes) {
|
||
Ok(decoded) => {
|
||
if let Some(smaller) = &decoded.smaller {
|
||
store_image(&path, smaller).await;
|
||
}
|
||
return engine.emit(Event::Poster {
|
||
key: title_key,
|
||
image: decoded.image,
|
||
});
|
||
}
|
||
Err(error) => tracing::debug!("cached poster {title_key}: {error}"),
|
||
}
|
||
}
|
||
|
||
let Some(service) = engine.metadata.lock().service.clone() else {
|
||
// Enrichment was turned off between the request and now.
|
||
return engine.emit(Event::PosterMissing { key: title_key });
|
||
};
|
||
|
||
let bytes = match service.artwork(&url).await {
|
||
Ok(bytes) => bytes,
|
||
Err(error) => {
|
||
tracing::debug!("fetch poster for {title_key}: {error}");
|
||
return engine.emit(Event::PosterMissing { key: title_key });
|
||
}
|
||
};
|
||
|
||
match decode_thumbnail(&bytes) {
|
||
Ok(decoded) => {
|
||
store_image(&path, decoded.smaller.as_deref().unwrap_or(&bytes)).await;
|
||
engine.emit(Event::Poster {
|
||
key: title_key,
|
||
image: decoded.image,
|
||
})
|
||
}
|
||
Err(error) => {
|
||
tracing::debug!("decode poster for {title_key}: {error}");
|
||
engine.emit(Event::PosterMissing { key: title_key })
|
||
}
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
/// Every file in `library`, by the key its download is filed under.
|
||
fn download_targets(library: &Library) -> HashMap<DownloadKey, PlaybackTarget> {
|
||
let mut targets = HashMap::new();
|
||
for title in &library.titles {
|
||
for episode in title.episodes() {
|
||
let target = PlaybackTarget::new(title, episode);
|
||
targets.insert(DownloadKey::from(&target), target);
|
||
}
|
||
}
|
||
targets
|
||
}
|
||
|
||
/// A short, filesystem-safe name for a URL.
|
||
fn digest(url: &str) -> String {
|
||
use sha2::{Digest as _, Sha256};
|
||
let hash = Sha256::digest(url.as_bytes());
|
||
hash.iter()
|
||
.take(16)
|
||
.map(|byte| format!("{byte:02x}"))
|
||
.collect()
|
||
}
|
||
|
||
/// The cache and texture key for a file's poster.
|
||
/// One title's worth of work in a match run.
|
||
enum Work {
|
||
/// Search for it, and take its episodes if it matches.
|
||
Match(Title),
|
||
/// It is already matched; only the episode list is missing.
|
||
Episodes(Title, Box<pstr_core::metadata::TitleMetadata>),
|
||
}
|
||
|
||
/// Unix seconds, for the "when was this asked" columns.
|
||
fn now() -> i64 {
|
||
std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.map(|since| since.as_secs() as i64)
|
||
.unwrap_or_default()
|
||
}
|
||
|
||
pub fn thumbnail_key(node: &CatalogNode) -> String {
|
||
format!("{}-{}", node.share_id, node.link_id)
|
||
}
|
||
|
||
fn node_uid(volume_id: &str, link_id: &str) -> NodeUid {
|
||
NodeUid::new(
|
||
VolumeId::new(volume_id.to_string()),
|
||
LinkId::new(link_id.to_string()),
|
||
)
|
||
}
|
||
|
||
/// Resume only after the last completely durable Proton block. A process may
|
||
/// die during a write, so trailing bytes inside a block are deliberately
|
||
/// discarded rather than trusted.
|
||
fn resume_position(existing: u64, block_sizes: &[u64]) -> (usize, u64) {
|
||
let mut offset = 0_u64;
|
||
for (index, size) in block_sizes.iter().copied().enumerate() {
|
||
let Some(next) = offset.checked_add(size) else {
|
||
break;
|
||
};
|
||
if next > existing {
|
||
return (index, offset);
|
||
}
|
||
offset = next;
|
||
}
|
||
(block_sizes.len(), offset)
|
||
}
|
||
|
||
/// A file's blocks from `first` on, fetched several at a time and yielded in
|
||
/// file order — which is what the resume marker's "the length is the
|
||
/// boundary" rule needs. One at a time, a download runs at one block per round
|
||
/// trip whatever the link can carry.
|
||
fn fetch_in_order(
|
||
stream: &VideoStream,
|
||
block_sizes: &[u64],
|
||
first: usize,
|
||
) -> futures::stream::BoxStream<'static, pstr_stream::Result<Vec<u8>>> {
|
||
use futures::StreamExt as _;
|
||
let start: u64 = block_sizes[..first].iter().sum();
|
||
let ranges: Vec<(u64, u64)> = block_sizes[first..]
|
||
.iter()
|
||
.scan(start, |offset, size| {
|
||
let range = (*offset, *size);
|
||
*offset += size;
|
||
Some(range)
|
||
})
|
||
.collect();
|
||
let stream = stream.clone();
|
||
futures::stream::iter(ranges)
|
||
.map(move |(offset, size)| {
|
||
let stream = stream.clone();
|
||
async move { stream.read_range(offset, size).await }
|
||
})
|
||
.buffered(DOWNLOAD_BLOCKS_IN_FLIGHT)
|
||
.boxed()
|
||
}
|
||
|
||
async fn wait_until_runnable(command: &mut tokio::sync::watch::Receiver<DownloadCommand>) -> bool {
|
||
loop {
|
||
let current = *command.borrow_and_update();
|
||
match current {
|
||
DownloadCommand::Run => return true,
|
||
DownloadCommand::Cancel => return false,
|
||
DownloadCommand::Pause => {
|
||
if command.changed().await.is_err() {
|
||
return false;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
async fn remove_if_present(path: &std::path::Path) -> std::io::Result<()> {
|
||
match tokio::fs::remove_file(path).await {
|
||
Ok(()) => Ok(()),
|
||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
||
Err(error) => Err(error),
|
||
}
|
||
}
|
||
|
||
#[cfg(unix)]
|
||
async fn sync_parent(path: &std::path::Path) -> std::io::Result<()> {
|
||
let Some(parent) = path.parent() else {
|
||
return Ok(());
|
||
};
|
||
tokio::fs::File::open(parent).await?.sync_all().await
|
||
}
|
||
|
||
#[cfg(not(unix))]
|
||
async fn sync_parent(_path: &std::path::Path) -> std::io::Result<()> {
|
||
// Windows does not expose directory handles through std/tokio. The file
|
||
// itself is synced before publication and rename is still atomic.
|
||
Ok(())
|
||
}
|
||
|
||
async fn write_marker_atomically(
|
||
path: &std::path::Path,
|
||
marker: &PartialMarker,
|
||
) -> anyhow::Result<()> {
|
||
use tokio::io::AsyncWriteExt;
|
||
let temporary = path.with_extension("partial.json.tmp");
|
||
let bytes = serde_json::to_vec(marker)?;
|
||
let mut output = tokio::fs::File::create(&temporary).await?;
|
||
output.write_all(&bytes).await?;
|
||
output.sync_all().await?;
|
||
drop(output);
|
||
tokio::fs::rename(temporary, path).await?;
|
||
sync_parent(path).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// A picture ready to upload, and — when it had to be scaled down — the
|
||
/// scaled copy encoded again, for the disk cache to keep instead.
|
||
struct Decoded {
|
||
image: egui::ColorImage,
|
||
smaller: Option<Vec<u8>>,
|
||
}
|
||
|
||
/// Decode image bytes into something egui can upload, scaled down to
|
||
/// [`THUMBNAIL_MAX_EDGE`].
|
||
///
|
||
/// Runs on the runtime rather than the UI thread: a JPEG decode per tile during
|
||
/// the first paint of a large library is visible as dropped frames.
|
||
///
|
||
/// A provider's backdrop is often 1920 px or more, and decoding and resizing
|
||
/// that every launch was most of what painting a cached grid cost. The scaled
|
||
/// copy is handed back so the cache holds that instead, and the next launch
|
||
/// decodes a picture already the size it is drawn at.
|
||
fn decode_thumbnail(bytes: &[u8]) -> anyhow::Result<Decoded> {
|
||
let decoded = image::load_from_memory(bytes)?;
|
||
let (decoded, smaller) = if decoded.width().max(decoded.height()) > THUMBNAIL_MAX_EDGE {
|
||
let scaled = decoded.thumbnail(THUMBNAIL_MAX_EDGE, THUMBNAIL_MAX_EDGE);
|
||
let encoded = encode_for_cache(&scaled);
|
||
(scaled, encoded)
|
||
} else {
|
||
(decoded, None)
|
||
};
|
||
let rgba = decoded.to_rgba8();
|
||
let size = [rgba.width() as usize, rgba.height() as usize];
|
||
Ok(Decoded {
|
||
image: egui::ColorImage::from_rgba_unmultiplied(size, rgba.as_flat_samples().as_slice()),
|
||
smaller,
|
||
})
|
||
}
|
||
|
||
/// Encode a scaled picture for the cache: PNG when it has transparency to keep
|
||
/// (a logo), JPEG otherwise. `None` if encoding failed, which leaves the cache
|
||
/// holding the original — slower to load, never wrong.
|
||
fn encode_for_cache(image: &image::DynamicImage) -> Option<Vec<u8>> {
|
||
let mut bytes = std::io::Cursor::new(Vec::new());
|
||
let encoded = if image.color().has_alpha() {
|
||
image.write_to(&mut bytes, image::ImageFormat::Png)
|
||
} else {
|
||
image::codecs::jpeg::JpegEncoder::new_with_quality(&mut bytes, 88)
|
||
.encode_image(&image.to_rgb8())
|
||
};
|
||
encoded.ok().map(|()| bytes.into_inner())
|
||
}
|
||
|
||
/// Write a cached picture. Best effort: the cache is a convenience, and a
|
||
/// write that fails only means the next launch fetches it again.
|
||
async fn store_image(path: &Path, bytes: &[u8]) {
|
||
if let Some(parent) = path.parent() {
|
||
let _ = tokio::fs::create_dir_all(parent).await;
|
||
}
|
||
let _ = tokio::fs::write(path, bytes).await;
|
||
}
|
||
|
||
/// Watch state as of now, for a position in a file.
|
||
pub fn watch_state(position: f64, duration: Option<f64>, watched: bool) -> WatchState {
|
||
WatchState {
|
||
position_secs: position,
|
||
duration_secs: duration,
|
||
watched,
|
||
updated_at: std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.map(|since| since.as_secs() as i64)
|
||
.unwrap_or(0),
|
||
}
|
||
}
|
||
|
||
/// Shares that failed to open, formatted for one status line.
|
||
pub fn describe_failures(failures: &[(String, String)]) -> Option<String> {
|
||
if failures.is_empty() {
|
||
return None;
|
||
}
|
||
let names: Vec<&str> = failures.iter().map(|(name, _)| name.as_str()).collect();
|
||
Some(format!("could not open: {}", names.join(", ")))
|
||
}
|
||
|
||
/// Pictures the UI has, plus the ones it has already asked for.
|
||
///
|
||
/// Two instances exist and they hold different things: one keyed by file for
|
||
/// Proton's own thumbnails, one keyed by title for a provider's artwork. What
|
||
/// they share is the part that matters — a fast scroll must not re-ask for the
|
||
/// same picture sixty times a second, and a picture that turned out not to exist
|
||
/// must be remembered as not existing. Without that second half, every render
|
||
/// pays a round-trip per unmatched tile, which is exactly what
|
||
/// `proton-drive-linux`'s photo grid had to learn.
|
||
///
|
||
/// It is also bounded. Scrolling a large library once used to keep a texture
|
||
/// for every title in it for the rest of the session; past [`IMAGE_BUDGET`]
|
||
/// bytes the picture asked for longest ago goes, and comes back from the disk
|
||
/// cache if it is asked for again.
|
||
pub struct ImageCache {
|
||
textures: HashMap<String, Entry>,
|
||
requested: HashSet<String>,
|
||
missing: HashSet<String>,
|
||
/// Decoded, waiting for a frame with upload budget left.
|
||
pending: VecDeque<(String, egui::ColorImage)>,
|
||
/// A logical clock: one tick per lookup, so "asked for longest ago" needs
|
||
/// no wall time.
|
||
clock: u64,
|
||
bytes: usize,
|
||
budget: usize,
|
||
}
|
||
|
||
struct Entry {
|
||
texture: egui::TextureHandle,
|
||
used: u64,
|
||
bytes: usize,
|
||
}
|
||
|
||
/// Texture memory one [`ImageCache`] may hold. A 640 px card picture is about
|
||
/// a megabyte, so this is a few hundred tiles — several screens of grid.
|
||
const IMAGE_BUDGET: usize = 256 << 20;
|
||
|
||
/// Textures uploaded per frame. A cached grid answers dozens of pictures at
|
||
/// once, and uploading them all in one frame is a visible hitch; spread over a
|
||
/// few frames they arrive faster than the eye follows.
|
||
const UPLOADS_PER_FRAME: usize = 6;
|
||
|
||
impl Default for ImageCache {
|
||
fn default() -> Self {
|
||
Self::with_budget(IMAGE_BUDGET)
|
||
}
|
||
}
|
||
|
||
impl ImageCache {
|
||
fn with_budget(budget: usize) -> Self {
|
||
Self {
|
||
textures: HashMap::new(),
|
||
requested: HashSet::new(),
|
||
missing: HashSet::new(),
|
||
pending: VecDeque::new(),
|
||
clock: 0,
|
||
bytes: 0,
|
||
budget,
|
||
}
|
||
}
|
||
|
||
/// The texture for `key`, calling `fetch` the first time it is asked for.
|
||
///
|
||
/// `None` while a fetch is in flight, and forever for a picture that turned
|
||
/// out not to exist — the negative answer is remembered deliberately.
|
||
///
|
||
/// Borrows the key: this is asked once per tile per frame, and a hit — the
|
||
/// usual answer — should not cost an allocation.
|
||
pub fn texture(&mut self, key: &str, fetch: impl FnOnce()) -> Option<egui::TextureHandle> {
|
||
self.clock += 1;
|
||
if let Some(entry) = self.textures.get_mut(key) {
|
||
entry.used = self.clock;
|
||
return Some(entry.texture.clone());
|
||
}
|
||
if !self.missing.contains(key) && !self.requested.contains(key) {
|
||
self.requested.insert(key.to_owned());
|
||
fetch();
|
||
}
|
||
None
|
||
}
|
||
|
||
/// Queue a decoded picture. It becomes a texture in [`Self::upload`].
|
||
pub fn insert(&mut self, key: String, image: egui::ColorImage) {
|
||
// Forgotten while it was being fetched: the answer is for a question
|
||
// nobody is asking any more.
|
||
if self.requested.contains(&key) {
|
||
self.pending.push_back((key, image));
|
||
}
|
||
}
|
||
|
||
/// Upload up to this frame's share of queued pictures. Returns whether any
|
||
/// are still waiting, so the caller can ask for another frame.
|
||
pub fn upload(&mut self, ctx: &egui::Context) -> bool {
|
||
for _ in 0..UPLOADS_PER_FRAME {
|
||
let Some((key, image)) = self.pending.pop_front() else {
|
||
break;
|
||
};
|
||
let bytes = image.pixels.len() * 4;
|
||
let texture = ctx.load_texture(&key, image, egui::TextureOptions::LINEAR);
|
||
self.clock += 1;
|
||
let entry = Entry {
|
||
texture,
|
||
used: self.clock,
|
||
bytes,
|
||
};
|
||
if let Some(old) = self.textures.insert(key, entry) {
|
||
self.bytes -= old.bytes;
|
||
}
|
||
self.bytes += bytes;
|
||
self.evict();
|
||
}
|
||
!self.pending.is_empty()
|
||
}
|
||
|
||
/// Drop the pictures asked for longest ago until the rest fit the budget.
|
||
fn evict(&mut self) {
|
||
while self.bytes > self.budget && self.textures.len() > 1 {
|
||
let Some(oldest) = self
|
||
.textures
|
||
.iter()
|
||
.min_by_key(|(_, entry)| entry.used)
|
||
.map(|(key, _)| key.clone())
|
||
else {
|
||
break;
|
||
};
|
||
if let Some(entry) = self.textures.remove(&oldest) {
|
||
self.bytes -= entry.bytes;
|
||
}
|
||
// Asked for again, it is fetched again — from the disk cache.
|
||
self.requested.remove(&oldest);
|
||
}
|
||
}
|
||
|
||
pub fn mark_missing(&mut self, key: String) {
|
||
self.missing.insert(key);
|
||
}
|
||
|
||
/// Forget one picture, so the next ask fetches it again.
|
||
pub fn forget(&mut self, key: &str) {
|
||
if let Some(entry) = self.textures.remove(key) {
|
||
self.bytes -= entry.bytes;
|
||
}
|
||
self.requested.remove(key);
|
||
self.missing.remove(key);
|
||
self.pending.retain(|(pending, _)| pending != key);
|
||
}
|
||
|
||
/// Forget everything. Used when the catalog is replaced, so a recrawl that
|
||
/// changed link ids does not leave stale pictures behind, and when the
|
||
/// metadata provider changes, so the old provider's artwork goes with it.
|
||
pub fn clear(&mut self) {
|
||
self.textures.clear();
|
||
self.requested.clear();
|
||
self.missing.clear();
|
||
self.pending.clear();
|
||
self.bytes = 0;
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod image_cache_tests {
|
||
use super::*;
|
||
|
||
fn picture() -> egui::ColorImage {
|
||
egui::ColorImage::filled([4, 4], egui::Color32::WHITE)
|
||
}
|
||
|
||
#[test]
|
||
fn a_full_cache_drops_the_picture_asked_for_longest_ago() {
|
||
let ctx = egui::Context::default();
|
||
// Room for two 4×4 pictures.
|
||
let mut cache = ImageCache::with_budget(2 * 64);
|
||
for key in ["a", "b"] {
|
||
cache.texture(key, || {});
|
||
cache.insert(key.into(), picture());
|
||
}
|
||
cache.upload(&ctx);
|
||
// "a" is looked at again, so "b" is now the stale one.
|
||
assert!(cache.texture("a", || {}).is_some());
|
||
cache.texture("c", || {});
|
||
cache.insert("c".into(), picture());
|
||
cache.upload(&ctx);
|
||
|
||
assert!(cache.texture("a", || {}).is_some());
|
||
assert!(cache.texture("c", || {}).is_some());
|
||
let mut fetched = false;
|
||
assert!(cache.texture("b", || fetched = true).is_none());
|
||
assert!(
|
||
fetched,
|
||
"an evicted picture is fetched again when asked for"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn uploads_are_spread_over_frames() {
|
||
let ctx = egui::Context::default();
|
||
let mut cache = ImageCache::default();
|
||
for index in 0..UPLOADS_PER_FRAME + 1 {
|
||
let key = format!("{index}");
|
||
cache.texture(&key, || {});
|
||
cache.insert(key, picture());
|
||
}
|
||
assert!(cache.upload(&ctx), "one picture is left for the next frame");
|
||
assert!(!cache.upload(&ctx));
|
||
}
|
||
|
||
#[test]
|
||
fn an_answer_for_a_forgotten_picture_is_dropped() {
|
||
let ctx = egui::Context::default();
|
||
let mut cache = ImageCache::default();
|
||
cache.texture("a", || {});
|
||
cache.forget("a");
|
||
cache.insert("a".into(), picture());
|
||
cache.upload(&ctx);
|
||
assert!(cache.textures.is_empty());
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod download_tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn a_partial_download_resumes_only_after_a_complete_block() {
|
||
let blocks = [3, 5, 2];
|
||
assert_eq!(resume_position(0, &blocks), (0, 0));
|
||
assert_eq!(resume_position(3, &blocks), (1, 3));
|
||
assert_eq!(resume_position(7, &blocks), (1, 3));
|
||
assert_eq!(resume_position(8, &blocks), (2, 8));
|
||
assert_eq!(resume_position(99, &blocks), (3, 10));
|
||
}
|
||
|
||
#[test]
|
||
fn pause_resume_and_cancel_controls_are_distinct() {
|
||
let control = DownloadControl::new();
|
||
let command = control.subscribe();
|
||
assert_eq!(*command.borrow(), DownloadCommand::Run);
|
||
control.set(DownloadCommand::Pause);
|
||
assert_eq!(*command.borrow(), DownloadCommand::Pause);
|
||
control.set(DownloadCommand::Run);
|
||
assert_eq!(*command.borrow(), DownloadCommand::Run);
|
||
control.set(DownloadCommand::Cancel);
|
||
assert_eq!(*command.borrow(), DownloadCommand::Cancel);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn resume_before_wait_cannot_be_lost() {
|
||
let control = DownloadControl::new();
|
||
let mut command = control.subscribe();
|
||
control.set(DownloadCommand::Pause);
|
||
control.set(DownloadCommand::Run);
|
||
assert!(wait_until_runnable(&mut command).await);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn cancellation_wakes_a_paused_writer_deterministically() {
|
||
let control = Arc::new(DownloadControl::new());
|
||
let mut command = control.subscribe();
|
||
control.set(DownloadCommand::Pause);
|
||
let waiter = tokio::spawn(async move { wait_until_runnable(&mut command).await });
|
||
tokio::task::yield_now().await;
|
||
control.set(DownloadCommand::Cancel);
|
||
assert!(
|
||
!tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
|
||
.await
|
||
.expect("the paused waiter woke")
|
||
.expect("waiter did not panic")
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn startup_hydrates_a_durable_partial_as_resumable() {
|
||
let unique = format!(
|
||
"pstr-download-hydration-{}-{}",
|
||
std::process::id(),
|
||
std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.expect("clock")
|
||
.as_nanos()
|
||
);
|
||
let root = std::env::temp_dir().join(unique);
|
||
let dirs = AppDirs::from_paths(root.join("config"), root.join("data"), root.join("cache"))
|
||
.expect("temporary app dirs");
|
||
let runtime = Arc::new(Runtime::new().expect("runtime"));
|
||
let (engine, _) =
|
||
Engine::new(Arc::clone(&runtime), dirs, egui::Context::default()).expect("engine");
|
||
let target = PlaybackTarget {
|
||
share_id: "share".to_owned(),
|
||
volume_id: "volume".to_owned(),
|
||
link_id: "episode".to_owned(),
|
||
name: "Show.S01E01.mkv".to_owned(),
|
||
title_key: "show".to_owned(),
|
||
title_name: "Show".to_owned(),
|
||
subtitle: "S01E01".to_owned(),
|
||
season: Some(1),
|
||
number: Some(1),
|
||
episode_name: None,
|
||
resume_at: None,
|
||
track_prefs: None,
|
||
};
|
||
let key = DownloadKey::from(&target);
|
||
let (partial, marker_path) = engine.partial_paths(&key);
|
||
std::fs::create_dir_all(partial.parent().expect("offline parent"))
|
||
.expect("create offline parent");
|
||
std::fs::write(&partial, [0_u8; 7]).expect("write one block and a fragment");
|
||
runtime
|
||
.block_on(write_marker_atomically(
|
||
&marker_path,
|
||
&PartialMarker {
|
||
revision_id: "revision".to_owned(),
|
||
block_sizes: vec![4, 5],
|
||
target: Some(target),
|
||
},
|
||
))
|
||
.expect("durable marker");
|
||
|
||
engine.hydrate_partial_downloads();
|
||
let downloads = engine.downloads.lock();
|
||
let item = downloads.items.get(&key).expect("hydrated row");
|
||
assert_eq!(item.state, DownloadState::Cancelled);
|
||
assert_eq!(item.downloaded, 4, "only a complete block is resumable");
|
||
assert_eq!(item.total, 9);
|
||
drop(downloads);
|
||
std::fs::remove_dir_all(root).expect("remove temporary app dirs");
|
||
}
|
||
}
|