nvrl/proton-stream
1
0 Fork 0
mirror of https://github.com/narrrl/proton-stream.git synced 2026-11-03 16:41:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

feat(core,app,android,cli,ios): sync the library and settings, verify a share before adding it

Browse Source 
Sync carried only watch positions, so a second device started empty. The
per-installation history file now also carries per-key registers
(pstr-core::profile): every share with its link and custom password, the
playback, appearance and metadata settings, the TMDB key, hand-picked
matches and per-title track choices. Newer updated_at wins per key, a
removal travels as an empty register, and a first sync stamps at 0 so a
fresh install takes the library the other devices have instead of
overwriting it with defaults. Volume, mute and the device-only settings
(cache budget, Wi-Fi only, background audio, hardware decoding) stay out.
The secrets live only in the end-to-end encrypted Drive file, never in
shares.json. FORMAT_VERSION stays 1; an older build ignores the registers.

Adding a share stored it before opening the link, so a first failure left a
broken entry and every retry said "already configured". add_verified opens
the link first and stores nothing on failure, and only the new share is
crawled afterwards, so another broken share cannot fail the add.

Android drops picture-in-picture: the window sized the video wrongly and
background audio already covers leaving the app.

iOS sizes the Metal drawable on every layout, so the picture no longer
stretches after a rotation or a Split View resize. Settings show a save at
once rather than after the match run, and a failed save reverts and says
why. Both phone clients gain an activity banner naming what is running, and
the add-share, metadata and change-match forms show progress and errors in
place.
This commit is contained in:
nvrl committed 2026-10-09 03:15:39 +02:00
1 parent dc70cafbb7
commit 8e42906a7f
37 files changed
+2018 -594
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/pstr-android/src/lib.rs
+151 -41
View File
@@ -187,6 +187,16 @@ pub struct SyncRecord {
pub uploaded: bool,
/// Other installations whose history was read.
pub devices: u32,
/// Shares another device added or changed, configured here and not yet
/// crawled. Crawl each.
pub shares_added: Vec<String>,
/// Shares another device removed, gone here with their offline files.
/// Drop any download state kept for them.
pub shares_removed: Vec<String>,
/// Whether settings changed: repaint, and re-read the metadata settings.
pub settings_changed: bool,
/// Hand-picked matches and per-show track choices taken.
pub titles: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, uniffi::Enum)]
@@ -704,6 +714,22 @@ impl<T> Drop for AbortOnDrop<T> {
}
}
/// Why a link could not be added, in words a viewer can act on.
///
/// A link that is already in the library and one that is not a link at all
/// say so themselves; anything else failed at the handshake, where a wrong
/// custom password and an expired link look the same from here.
fn add_share_failure(error: &pstr_core::Error) -> String {
match error {
pstr_core::Error::Config(reason) => reason.clone(),
pstr_core::Error::Drive(_) => format!(
"Could not open that link: {error}. Check that it is complete, \
still shared, and that the password is right."
),
other => other.to_string(),
}
}
/// `runtime.spawn(task)`, with the handle wrapped so the task is aborted if the
/// caller stops waiting. Awaits to exactly what `JoinHandle` does.
fn spawned<T: Send + 'static>(
@@ -1143,12 +1169,31 @@ impl AndroidEngine {
else {
return Ok(None);
};
match sync.sync(&account, &self.catalog).await {
Ok(report) => Ok(Some(SyncRecord {
applied: u32::try_from(report.applied).unwrap_or(u32::MAX),
uploaded: report.uploaded,
devices: u32::try_from(report.devices).unwrap_or(u32::MAX),
})),
match sync.sync(&account, &self.catalog, &self.store).await {
Ok(report) => {
let profile = report.profile;
for id in profile.shares_added.iter().chain(&profile.shares_changed) {
self.invalidate_connection(id);
}
for id in &profile.shares_removed {
if let Err(error) = self.forget_synced_share(id) {
log::warn!("files of {id}, a share removed elsewhere: {error}");
}
}
Ok(Some(SyncRecord {
applied: u32::try_from(report.applied).unwrap_or(u32::MAX),
uploaded: report.uploaded,
devices: u32::try_from(report.devices).unwrap_or(u32::MAX),
shares_added: profile
.shares_added
.into_iter()
.chain(profile.shares_changed)
.collect(),
shares_removed: profile.shares_removed,
settings_changed: profile.settings,
titles: u32::try_from(profile.titles).unwrap_or(u32::MAX),
}))
}
Err(error) if error.is_signed_out() => {
let _ = self.accounts.sign_out(None).await;
self.signed_out();
@@ -1223,6 +1268,26 @@ impl AndroidEngine {
Ok(share_record(share))
}
/// Configured shares with nothing in the catalog: brought in by sync,
/// possibly while the app was in the background, or added here and never
/// crawled to the end. Crawl each. An empty share is crawled every time,
/// which costs one listing.
pub fn uncrawled_shares(&self) -> Result<Vec<String>, BridgeError> {
let shares = self.store.list().map_err(BridgeError::from_display)?;
let catalog = self.catalog.lock();
let mut uncrawled = Vec::new();
for share in shares {
if catalog
.files(&share.id)
.map_err(BridgeError::from_display)?
.is_empty()
{
uncrawled.push(share.id);
}
}
Ok(uncrawled)
}
pub fn shares(&self) -> Result<Vec<ShareRecord>, BridgeError> {
self.store
.list()
@@ -1230,18 +1295,34 @@ impl AndroidEngine {
.map(|shares| shares.into_iter().map(share_record).collect())
}
pub fn add_share(
&self,
/// Add a public link, once it has opened with the password given. A link
/// that will not open is refused and leaves nothing behind. Crawl the
/// returned share next.
pub async fn add_share(
self: Arc<Self>,
name: String,
url: String,
custom_password: Option<String>,
) -> Result<ShareRecord, BridgeError> {
let share = self
.store
.add(&name, &url, custom_password.as_deref())
.map_err(BridgeError::from_display)?;
self.invalidate_connection(&share.id);
Ok(share_record(share))
let runtime = Arc::clone(&self.runtime);
spawned(&runtime, async move {
let (share, client) = self
.store
.add_verified(&name, &url, custom_password.as_deref())
.await
.map_err(|error| BridgeError::Failure {
reason: add_share_failure(&error),
})?;
self.invalidate_connection(&share.id);
// The handshake that proved the link is the one the crawl uses.
self.reusable_clients
.lock()
.get_or_insert_with(Default::default)
.insert(share.id.clone(), client);
Ok(share_record(share))
})
.await
.map_err(BridgeError::from_display)?
}
/// Re-supply the link behind a share whose stored secret is unreadable.
@@ -1291,28 +1372,7 @@ impl AndroidEngine {
// when secret cleanup fails, old authenticated clients and catalog
// rows must not remain usable.
self.invalidate_connection(&share_id);
for ((stored_share_id, link_id), file) in offline {
if stored_share_id == share_id {
let path = self
.dirs
.offline_file(&share_id, &link_id, &file.revision_id);
match std::fs::remove_file(&path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(BridgeError::from_display(error)),
}
}
}
for link_id in links {
let (partial, marker) = self.partial_paths(&share_id, &link_id);
let _ = std::fs::remove_file(partial);
let _ = std::fs::remove_file(marker);
}
self.catalog
.lock()
.remove_share(&share_id)
.map_err(BridgeError::from_display)?;
self.forget_share_files(&share_id, offline, links)?;
store_result.map_err(BridgeError::from_display)
}
@@ -1901,6 +1961,56 @@ impl AndroidEngine {
}
impl AndroidEngine {
/// Delete a removed share's offline and partial files, then its catalog
/// rows. Watch state stays; see `Catalog::remove_share`.
fn forget_share_files(
&self,
share_id: &str,
offline: HashMap<(String, String), OfflineFile>,
links: Vec<String>,
) -> Result<(), BridgeError> {
for ((stored_share_id, link_id), file) in offline {
if stored_share_id == share_id {
let path = self
.dirs
.offline_file(share_id, &link_id, &file.revision_id);
match std::fs::remove_file(&path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(BridgeError::from_display(error)),
}
}
}
for link_id in links {
let (partial, marker) = self.partial_paths(share_id, &link_id);
let _ = std::fs::remove_file(partial);
let _ = std::fs::remove_file(marker);
}
self.catalog
.lock()
.remove_share(share_id)
.map_err(BridgeError::from_display)
}
/// [`Self::forget_share_files`] for a share another device removed, which
/// sync has already taken out of the configuration.
fn forget_synced_share(&self, share_id: &str) -> Result<(), BridgeError> {
let _publication = self.share_publication.lock();
let catalog = self.catalog.lock();
let offline = catalog
.all_offline_files()
.map_err(BridgeError::from_display)?;
let links: Vec<String> = catalog
.files(share_id)
.map_err(BridgeError::from_display)?
.into_iter()
.map(|node| node.link_id)
.collect();
drop(catalog);
self.invalidate_connection(share_id);
self.forget_share_files(share_id, offline, links)
}
/// One enrichment pass over the whole library.
///
/// What to do per title is decided by `pstr_meta::service::plan`, the same
@@ -3765,13 +3875,13 @@ mod tests {
)
.expect("engine");
// Stored directly: `add_share` opens the link first, and a test has
// no network to open it over.
let share = engine
.add_share(
"test".to_owned(),
"https://drive.proton.me/urls/ABC123#s3cr3t".to_owned(),
None,
)
.store
.add("test", "https://drive.proton.me/urls/ABC123#s3cr3t", None)
.expect("add share");
engine.invalidate_connection(&share.id);
assert_eq!(engine.share_generation.load(Ordering::Acquire), 1);
assert!(
engine
crates/pstr-app/src/app.rs
+4
View File
@@ -793,6 +793,10 @@ impl App {
self.view.stale = true;
}
Event::EpisodeMetadata(episodes) => self.episodes = episodes,
Event::Appearance(appearance) => {
theme::apply(ctx, appearance);
self.view.stale = true;
}
Event::MetadataConfig(config) => {
self.settings = config;
self.posters.clear();
crates/pstr-app/src/engine.rs
+84 -11
View File
@@ -34,6 +34,7 @@ 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::profile::ProfileChanges;
use pstr_core::proton_drive_rs::ThumbnailType;
use pstr_core::proton_sdk::api::HumanVerificationCredential;
use pstr_core::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
@@ -215,6 +216,8 @@ pub enum Event {
EpisodeMetadata(HashMap<String, EpisodeGuide>),
/// The enrichment settings changed, on disk.
MetadataConfig(MetadataConfig),
/// Another device's appearance arrived through sync.
Appearance(Appearance),
/// A matching run finished, with how it went.
Matched {
matched: usize,
@@ -279,7 +282,7 @@ pub enum Event {
SignInFailed(String),
/// Watch history synced with the account's Drive.
Synced(SyncReport),
/// Watch history did not sync, with why. Not a toast: it is retried on a
/// Sync did not run, with why. Not a toast: it is retried on a
/// timer, and an offline evening would be a stack of them.
SyncFailed(String),
/// The account's top-level places, for the Drive browser.
@@ -640,12 +643,17 @@ impl Engine {
}
/// Add a share, then open and crawl it — which is what "add" means to
/// someone who just pasted a link.
/// someone who just pasted a link. A link that will not open is refused
/// before anything is stored, so trying again is not "already configured".
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,
let share = match engine
.store
.add_verified(&name, &url, password.as_deref())
.await
{
Ok((share, _)) => share,
Err(error) => return engine.emit(Event::ShareRejected(error.to_string())),
};
engine.emit(Event::ShareAdded(share.name.clone()));
@@ -836,9 +844,10 @@ impl Engine {
};
let engine = self.clone();
self.runtime.spawn(async move {
match sync.sync(&account, &engine.catalog).await {
match sync.sync(&account, &engine.catalog, &engine.store).await {
Ok(report) => {
if report.applied > 0 {
engine.take_synced_profile(&report.profile).await;
if report.applied > 0 || report.profile.titles > 0 {
engine.load_library();
}
engine.emit(Event::Synced(report));
@@ -875,8 +884,9 @@ impl Engine {
return;
};
let catalog = Arc::clone(&self.catalog);
let store = Arc::clone(&self.store);
self.runtime.block_on(async move {
match tokio::time::timeout(timeout, sync.sync(&account, &catalog)).await {
match tokio::time::timeout(timeout, sync.sync(&account, &catalog, &store)).await {
Ok(Ok(_)) => {}
Ok(Err(error)) => tracing::warn!("sync watch history on exit: {error}"),
Err(_) => tracing::warn!("sync watch history on exit: timed out"),
@@ -884,6 +894,58 @@ impl Engine {
});
}
/// Act on what sync brought from other devices: reread the settings it
/// rewrote, crawl the shares it added, drop the ones it removed.
async fn take_synced_profile(&self, profile: &ProfileChanges) {
if profile.settings {
match pstr_core::prefs::load(&self.dirs) {
// The volume is this device's and was not synced; the file
// still holds it.
Ok(prefs) => *self.prefs.lock() = prefs,
Err(error) => tracing::warn!("reread the playback preferences: {error}"),
}
match pstr_core::appearance::load(&self.dirs) {
Ok(appearance) => {
*self.appearance.lock() = appearance;
self.emit(Event::Appearance(appearance));
}
Err(error) => tracing::warn!("reread the appearance: {error}"),
}
match pstr_meta::settings::load(&self.dirs) {
Ok(config) => {
let engine = self.clone();
let _ = self
.runtime
.spawn_blocking(move || {
engine.install_enrichment(Enrichment::build(config.clone()), false);
engine.emit(Event::MetadataConfig(config));
})
.await;
}
Err(error) => tracing::warn!("reread the metadata settings: {error}"),
}
}
if profile.titles > 0 {
match self.catalog.lock().all_title_track_prefs() {
Ok(prefs) => *self.track_prefs.lock() = prefs,
Err(error) => tracing::warn!("reread show track preferences: {error}"),
}
self.load_metadata();
}
for id in &profile.shares_removed {
self.forget_share(id.clone(), false).await;
}
if !profile.shares() {
return;
}
self.connection.lock().take();
self.load_shares();
self.load_library();
for id in profile.shares_added.iter().chain(&profile.shares_changed) {
self.connect_and_crawl(Some(id.clone())).await;
}
}
fn sync_periodically(&self) {
let engine = self.clone();
self.runtime.spawn(async move {
@@ -929,6 +991,19 @@ impl Engine {
pub fn remove_share(&self, id: String) {
let engine = self.clone();
self.runtime.spawn(async move {
engine.forget_share(id, true).await;
engine.connection.lock().take();
engine.load_library();
engine.emit(Event::Status("removed".into()));
});
}
/// Stop a share's downloads and delete its files and catalog rows, and,
/// when it is still `configured` — not already removed by sync — its
/// entry and secrets.
async fn forget_share(&self, id: String, configured: bool) {
let engine = self;
{
let keys: Vec<_> = engine
.downloads
.lock()
@@ -971,7 +1046,7 @@ impl Engine {
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) {
if configured && let Err(error) = engine.store.remove(&id) {
return engine.fail("remove that share", error);
}
{
@@ -983,9 +1058,7 @@ impl Engine {
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
crates/pstr-app/src/ui/account.rs
+8 -5
View File
@@ -133,9 +133,10 @@ impl DriveBrowser {
pub fn section(ui: &mut egui::Ui, panel: &mut AccountPanel, actions: &mut Vec<Action>) {
ui::section(ui, "Proton account");
ui.label(ui::muted(
"Optional. Signed in, your watch history is kept in your own Drive and every device \
resumes where another left off, and folders of your Drive — your files, other \
devices, what was shared with you — can be added to the library.",
"Optional. Signed in, your shares, settings and watch history are kept in your own \
Drive, so every device shows the same library and resumes where another left off, \
and folders of your Drive — your files, other devices, what was shared with you — \
can be added to the library.",
));
ui.add_space(theme::space::L);
@@ -281,9 +282,11 @@ fn signed_in(
.strong(),
);
let status = match (&panel.sync_error, &panel.synced) {
(Some(error), _) => format!("Watch history did not sync: {error}"),
(Some(error), _) => format!("Did not sync: {error}"),
(None, Some((at, report))) => sync_line(*at, report, unix_now()),
(None, None) => "Watch history syncs with your Drive".to_owned(),
(None, None) => {
"Library, settings and watch history sync through your Drive".to_owned()
}
};
ui.label(ui::muted(status));
});
crates/pstr-cli/src/main.rs
+23 -4
View File
@@ -256,8 +256,9 @@ async fn run(runtime: std::sync::Arc<tokio::runtime::Runtime>) -> Result<()> {
} else {
None
};
let share = store
.add(&name, &url, password.as_deref())
let (share, _) = store
.add_verified(&name, &url, password.as_deref())
.await
.context("add share")?;
println!("added {} ({})", share.name, share.id);
}
@@ -323,11 +324,29 @@ async fn run(runtime: std::sync::Arc<tokio::runtime::Runtime>) -> Result<()> {
let sync = WatchSync::open(&dirs).context("open sync state")?;
let catalog =
parking_lot::Mutex::new(Catalog::open(&dirs.catalog_db()).context("open catalog")?);
let report = sync.sync(&account, &catalog).await.context("sync")?;
let report = sync
.sync(&account, &catalog, &store)
.await
.context("sync")?;
let profile = &report.profile;
for id in &profile.shares_removed {
catalog
.lock()
.remove_share(id)
.context("drop a removed share")?;
}
println!(
"read {} other device(s), took {} newer position(s), {}",
"read {} other device(s), took {} newer position(s), \
{} share(s) added, {} removed, settings {}, {}",
report.devices,
report.applied,
profile.shares_added.len() + profile.shares_changed.len(),
profile.shares_removed.len(),
if profile.settings {
"updated"
} else {
"unchanged"
},
if report.uploaded {
"uploaded this device's history"
} else {
crates/pstr-core/src/lib.rs
+1
View File
@@ -17,6 +17,7 @@ pub mod library;
pub mod metadata;
pub mod naming;
pub mod prefs;
pub mod profile;
pub mod shares;
pub mod shelves;
pub mod sync;
crates/pstr-core/src/profile.rs
+808
View File
@@ -0,0 +1,808 @@
//! The rest of what makes the app look the same on every device: the shares,
//! the settings, the hand-picked metadata matches and the per-show track
//! choices. Synced next to watch history by [`crate::sync`], in the same file.
//!
//! ## Registers
//!
//! Each synced thing is one *register*: a key (`share/<id>`,
//! `settings/playback`, `tracks/<title>`, …), its latest value, and when it
//! last changed. Merging is the same rule as watch history — the newer
//! `updated_at` wins, per key — so it is idempotent and order-free across any
//! number of devices. A removed share is a register with no value, so a
//! removal travels like any other change.
//!
//! ## Noticing changes
//!
//! Nothing that writes a setting or a share knows about sync. Instead
//! `sync-state.json` remembers every register as last synced, and each sync
//! compares it with what is on this device: whatever differs changed since,
//! and is stamped with the time of the sync. That costs a change made between
//! syncs its exact time, which matters only when two devices change the same
//! setting within one sync interval of each other.
//!
//! The first sync of an installation stamps what it finds at zero rather than
//! now. A fresh install holds only defaults, and those must not outrank the
//! library every other device already agreed on; what it has that nobody else
//! does is still added, since nothing older competes with it.
//!
//! ## What does not travel
//!
//! The volume and mute, which belong to the speakers in front of the viewer,
//! and every setting the front ends keep themselves: the cache budget, Wi-Fi
//! only, background audio, hardware decoding, window geometry.
//!
//! ## Secrets
//!
//! A share's register carries its link, fragment included, and its custom
//! password, and the TMDB API key travels too: without them another device
//! could list the share but never open it. The file holding them is in the
//! viewer's own Drive, end-to-end encrypted with their account's keys, which is
//! the same trust the credential store gives them locally; they are never
//! written anywhere else in plain text.
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::catalog::{Catalog, TitleTrackPrefs};
use crate::config::{AppDirs, read_json, write_json};
use crate::error::Result;
use crate::metadata::{MetadataConfig, MetadataRecord, ProviderId, TitleMetadata};
use crate::shares::{AccountFolder, Share, ShareSecrets, ShareStore};
const SHARE: &str = "share/";
const TRACKS: &str = "tracks/";
const MATCH: &str = "match/";
/// The settings files that travel, by register, and the fields in each that
/// stay on the device that wrote them.
const SETTINGS: [(&str, &str, &[&str]); 3] = [
("settings/playback", "playback.json", &["volume", "muted"]),
("settings/appearance", "appearance.json", &[]),
("settings/metadata", "metadata.json", &[]),
];
const METADATA_SETTINGS: &str = "settings/metadata";
/// One synced value and when it last changed.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(crate) struct Register {
/// `None` is a removal.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub value: Option<Value>,
pub updated_at: i64,
}
pub(crate) type Registers = BTreeMap<String, Register>;
/// What this installation last synced, kept beside its sync identity.
#[derive(Debug, Default, Serialize, Deserialize)]
struct State {
registers: Registers,
}
/// A share as it travels: everything another device needs to open it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct SyncedShare {
name: String,
#[serde(default, skip_serializing_if = "String::is_empty")]
token: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
url: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
custom_password: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
folder: Option<AccountFolder>,
}
/// A match the viewer picked by hand. Automatic ones are not synced: every
/// device with enrichment on finds them again, and they expire anyway.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct SyncedMatch {
provider: ProviderId,
metadata: Option<TitleMetadata>,
fetched_at: i64,
}
/// What applying other devices' registers changed here, for the front end to
/// act on: reopen and crawl shares, repaint, reload the library.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProfileChanges {
/// Shares this device did not have, now configured. Not yet crawled.
pub shares_added: Vec<String>,
/// Shares whose name or secrets changed.
pub shares_changed: Vec<String>,
/// Shares removed elsewhere, now gone from the configuration here. Their
/// catalog rows and offline files are the front end's to drop, the same
/// way it does for a share removed on this device.
pub shares_removed: Vec<String>,
/// Whether any settings file or the TMDB key changed.
pub settings: bool,
/// Hand-picked matches and per-show track choices taken.
pub titles: usize,
}
impl ProfileChanges {
pub fn is_empty(&self) -> bool {
*self == Self::default()
}
pub fn shares(&self) -> bool {
!(self.shares_added.is_empty()
&& self.shares_changed.is_empty()
&& self.shares_removed.is_empty())
}
}
/// The local side of profile sync: where the state file lives, and the stores
/// registers are read from and applied to.
pub(crate) struct Profile<'a> {
pub store: &'a ShareStore,
pub catalog: &'a Mutex<Catalog>,
}
/// What is on this device, register by register.
#[derive(Debug, Default)]
struct Observed {
values: BTreeMap<String, Value>,
/// Keys that exist but could not be read — a share whose secret will not
/// decrypt. Neither a change nor a removal.
unknown: BTreeSet<String>,
/// Whether the share list itself could be read. Without it, a share
/// missing from `values` says nothing about whether it was removed.
shares_listed: bool,
}
fn state_file(dirs: &AppDirs) -> PathBuf {
dirs.config.join("sync-state.json")
}
fn tmdb_secret() -> String {
format!("metadata-{}", ProviderId::Tmdb.as_str())
}
const TMDB_KEY: &str = "secret/metadata-tmdb";
impl Profile<'_> {
fn dirs(&self) -> &AppDirs {
&self.store.dirs
}
/// Read the state, notice what changed here since, take what other
/// devices have newer, and write the state back. Returns the registers to
/// upload and what changed on this device.
pub fn sync<'r>(
&self,
remote: impl IntoIterator<Item = &'r Registers>,
now: i64,
) -> Result<(Registers, ProfileChanges)> {
let path = state_file(self.dirs());
let stored = read_json::<State>(&path)?;
let first = stored.is_none();
let mut state = stored.unwrap_or_default();
let observed = self.observe();
stamp(&mut state.registers, observed, if first { 0 } else { now });
let winners = merge(&state.registers, remote);
let changes = self.apply(&mut state.registers, winners);
write_json(&path, &state)?;
Ok((state.registers, changes))
}
fn observe(&self) -> Observed {
let mut observed = Observed::default();
for (key, file, local) in SETTINGS {
match read_json::<Value>(&self.dirs().config.join(file)) {
Ok(Some(value)) => {
observed
.values
.insert(key.to_owned(), without(value, local));
}
Ok(None) => {}
Err(error) => {
tracing::warn!(%error, file, "a settings file was not synced");
observed.unknown.insert(key.to_owned());
}
}
}
match self.store.secrets.get(&tmdb_secret()) {
Ok(Some(key)) => {
observed
.values
.insert(TMDB_KEY.to_owned(), Value::String(key));
}
Ok(None) => {}
Err(_) => {
observed.unknown.insert(TMDB_KEY.to_owned());
}
}
if let Ok(shares) = self.store.list() {
observed.shares_listed = true;
for share in shares {
let key = format!("{SHARE}{}", share.id);
let secrets = if share.folder.is_some() {
None
} else {
match self.store.secrets_of(&share.id) {
Ok(secrets) => Some(secrets),
Err(_) => {
observed.unknown.insert(key);
continue;
}
}
};
let synced = SyncedShare {
name: share.name,
token: share.token,
url: secrets.as_ref().map(|secrets| secrets.url.clone()),
custom_password: secrets.and_then(|secrets| secrets.custom_password),
folder: share.folder,
};
if let Ok(value) = serde_json::to_value(synced) {
observed.values.insert(key, value);
}
}
}
let catalog = self.catalog.lock();
if let Ok(tracks) = catalog.all_title_track_prefs() {
for (title, prefs) in tracks {
if let Ok(value) = serde_json::to_value(prefs) {
observed.values.insert(format!("{TRACKS}{title}"), value);
}
}
}
if let Ok(records) = catalog.all_metadata() {
for (title, record) in records {
if !record.manual {
continue;
}
let synced = SyncedMatch {
provider: record.provider,
metadata: record.metadata,
fetched_at: record.fetched_at,
};
if let Ok(value) = serde_json::to_value(synced) {
observed.values.insert(format!("{MATCH}{title}"), value);
}
}
}
observed
}
/// Apply each winning register here, recording it in `registers` once it
/// took. One that fails is left out, so the next sync tries it again.
fn apply(&self, registers: &mut Registers, winners: Registers) -> ProfileChanges {
let mut changes = ProfileChanges::default();
for (key, register) in winners {
match self.apply_one(&key, register.value.as_ref(), &mut changes) {
Ok(()) => {
registers.insert(key, register);
}
Err(error) => tracing::warn!(%error, key, "a synced change was not applied"),
}
}
changes
}
fn apply_one(
&self,
key: &str,
value: Option<&Value>,
changes: &mut ProfileChanges,
) -> Result<()> {
if let Some(id) = key.strip_prefix(SHARE) {
return self.apply_share(id, value, changes);
}
if let Some(title) = key.strip_prefix(TRACKS) {
if let Some(value) = value {
let prefs: TitleTrackPrefs = parse(value)?;
self.catalog.lock().set_title_track_prefs(title, &prefs)?;
changes.titles += 1;
}
return Ok(());
}
if let Some(title) = key.strip_prefix(MATCH) {
if let Some(value) = value {
let synced: SyncedMatch = parse(value)?;
self.catalog.lock().set_metadata(&MetadataRecord {
title_key: title.to_owned(),
provider: synced.provider,
metadata: synced.metadata,
fetched_at: synced.fetched_at,
manual: true,
})?;
changes.titles += 1;
}
return Ok(());
}
if key == TMDB_KEY {
match value.and_then(Value::as_str) {
Some(secret) => self.store.secrets.set(&tmdb_secret(), secret)?,
None => self.store.secrets.delete(&tmdb_secret())?,
}
changes.settings = true;
return Ok(());
}
if let Some((_, file, local)) = SETTINGS.iter().find(|(name, ..)| *name == key) {
let Some(Value::Object(incoming)) = value else {
return Ok(());
};
let path = self.dirs().config.join(file);
let before = read_json::<Value>(&path)?;
let mut merged = match &before {
Some(Value::Object(existing)) => existing.clone(),
_ => serde_json::Map::new(),
};
for (field, field_value) in incoming {
if !local.contains(&field.as_str()) {
merged.insert(field.clone(), field_value.clone());
}
}
let merged = Value::Object(merged);
if key == METADATA_SETTINGS {
self.metadata_settings_changing(before.as_ref(), &merged)?;
}
write_json(&path, &merged)?;
changes.settings = true;
return Ok(());
}
// A register a newer build writes. Kept, so it travels on.
Ok(())
}
fn apply_share(
&self,
id: &str,
value: Option<&Value>,
changes: &mut ProfileChanges,
) -> Result<()> {
let existing = self.store.list()?.into_iter().find(|share| share.id == id);
let Some(value) = value else {
if existing.is_some() {
self.store.remove(id)?;
changes.shares_removed.push(id.to_owned());
}
return Ok(());
};
let synced: SyncedShare = parse(value)?;
let expected = match (&synced.folder, &synced.url) {
(Some(folder), _) => folder.share_id(),
(None, Some(_)) => format!("share-{}", synced.token),
(None, None) => return Err(invalid("a synced link has no URL")),
};
if expected != id {
return Err(invalid("a synced share's id does not match what it opens"));
}
let secrets = synced.url.map(|url| ShareSecrets {
url,
custom_password: synced.custom_password,
});
let share = Share {
id: id.to_owned(),
name: synced.name,
token: synced.token,
has_custom_password: secrets
.as_ref()
.is_some_and(|secrets| secrets.custom_password.is_some()),
folder: synced.folder,
};
if existing.as_ref() == Some(&share)
&& secrets.as_ref().is_none_or(|secrets| {
self.store
.secrets_of(id)
.is_ok_and(|stored| &stored == secrets)
})
{
return Ok(());
}
self.store.put_synced(share, secrets.as_ref())?;
if existing.is_some() {
changes.shares_changed.push(id.to_owned());
} else {
changes.shares_added.push(id.to_owned());
}
Ok(())
}
/// Turning enrichment off or switching provider drops every stored answer
/// on the device that did it, which the settings page also does here.
fn metadata_settings_changing(&self, before: Option<&Value>, after: &Value) -> Result<()> {
let config = |value: Option<&Value>| -> MetadataConfig {
value
.and_then(|value| serde_json::from_value(value.clone()).ok())
.unwrap_or_default()
};
let (before, after) = (config(before), config(Some(after)));
if !after.enabled || before.provider != after.provider {
self.catalog.lock().clear_metadata()?;
}
Ok(())
}
}
/// Record what changed on this device since the last sync, stamped `now`.
fn stamp(registers: &mut Registers, observed: Observed, now: i64) {
for (key, register) in registers.iter_mut() {
if register.value.is_some()
&& key.starts_with(SHARE)
&& observed.shares_listed
&& !observed.values.contains_key(key)
&& !observed.unknown.contains(key)
{
*register = Register {
value: None,
updated_at: now,
};
}
}
for (key, value) in observed.values {
let unchanged = registers
.get(&key)
.is_some_and(|register| same(&key, register.value.as_ref(), Some(&value)));
if !unchanged {
registers.insert(
key,
Register {
value: Some(value),
updated_at: now,
},
);
}
}
}
/// Whether two values of one register are the same choice.
///
/// A match is compared by what it points at: its stored details move every
/// time the provider is asked again, and a register re-stamped by every
/// refresh would bounce between devices for ever.
fn same(key: &str, a: Option<&Value>, b: Option<&Value>) -> bool {
if key.starts_with(MATCH) {
let pointer = |value: Option<&Value>| {
value.map(|value| {
(
value.get("provider").cloned(),
value
.get("metadata")
.and_then(|metadata| metadata.get("remote_id"))
.cloned(),
)
})
};
return pointer(a) == pointer(b);
}
a == b
}
/// The registers from other devices that are newer than this one's.
fn merge<'r>(local: &Registers, remote: impl IntoIterator<Item = &'r Registers>) -> Registers {
let mut winners = Registers::new();
for registers in remote {
for (key, register) in registers {
let current = winners.get(key).or_else(|| local.get(key));
if current.is_none_or(|current| newer(register, current)) {
winners.insert(key.clone(), register.clone());
}
}
}
winners
}
/// Newer wins; a tie between two different values goes the same way on every
/// device, or two of them would each keep their own.
fn newer(candidate: &Register, current: &Register) -> bool {
if candidate.updated_at != current.updated_at {
return candidate.updated_at > current.updated_at;
}
let order = |register: &Register| {
register
.value
.as_ref()
.map(Value::to_string)
.unwrap_or_default()
};
order(candidate) > order(current)
}
/// `value` without the fields that stay on this device.
fn without(value: Value, local: &[&str]) -> Value {
match value {
Value::Object(mut fields) => {
for field in local {
fields.remove(*field);
}
Value::Object(fields)
}
other => other,
}
}
fn parse<T: serde::de::DeserializeOwned>(value: &Value) -> Result<T> {
serde_json::from_value(value.clone()).map_err(|e| invalid(&e.to_string()))
}
fn invalid(reason: &str) -> crate::Error {
crate::Error::Config(format!("synced setting: {reason}"))
}
#[cfg(test)]
pub(crate) mod tests {
use std::collections::HashMap;
use std::sync::Arc;
use super::*;
use crate::shares::SecretStore;
#[derive(Default)]
struct MemorySecrets(Mutex<HashMap<String, String>>);
impl SecretStore for MemorySecrets {
fn set(&self, key: &str, value: &str) -> Result<()> {
self.0.lock().insert(key.to_owned(), value.to_owned());
Ok(())
}
fn get(&self, key: &str) -> Result<Option<String>> {
Ok(self.0.lock().get(key).cloned())
}
fn delete(&self, key: &str) -> Result<()> {
self.0.lock().remove(key);
Ok(())
}
}
/// One installation's configuration and catalog.
pub(crate) struct Device {
root: PathBuf,
pub store: ShareStore,
pub catalog: Mutex<Catalog>,
}
impl Device {
pub(crate) fn new(tag: &str) -> Self {
let mut bytes = [0_u8; 8];
getrandom::fill(&mut bytes).unwrap();
let unique: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect();
let root = std::env::temp_dir().join(format!("pstr-profile-{tag}-{unique}"));
let dirs = AppDirs::from_paths(root.join("c"), root.join("d"), root.join("k")).unwrap();
Self {
root,
store: ShareStore::with_secret_store(dirs, Arc::new(MemorySecrets::default())),
catalog: Mutex::new(Catalog::in_memory().unwrap()),
}
}
pub(crate) fn lock(&self) -> parking_lot::MutexGuard<'_, Catalog> {
self.catalog.lock()
}
pub(crate) fn profile(&self) -> Profile<'_> {
Profile {
store: &self.store,
catalog: &self.catalog,
}
}
fn sync(&self, remote: &[&Registers], now: i64) -> (Registers, ProfileChanges) {
self.profile().sync(remote.iter().copied(), now).unwrap()
}
fn share_names(&self) -> Vec<String> {
self.store
.list()
.unwrap()
.into_iter()
.map(|share| share.name)
.collect()
}
}
impl Drop for Device {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.root);
}
}
const LINK: &str = "https://drive.proton.me/urls/ABCDEFGHIJ#secretpart";
#[test]
fn a_share_added_on_one_device_arrives_on_another_with_its_secrets() {
let laptop = Device::new("laptop");
let phone = Device::new("phone");
laptop.sync(&[], 10);
laptop.store.add("Anime", LINK, Some("hunter2")).unwrap();
let (sent, _) = laptop.sync(&[], 20);
let (_, changes) = phone.sync(&[&sent], 30);
assert_eq!(changes.shares_added, ["share-ABCDEFGHIJ"]);
assert_eq!(phone.share_names(), ["Anime"]);
let secrets = phone.store.secrets_of("share-ABCDEFGHIJ").unwrap();
assert_eq!(secrets.url, LINK);
assert_eq!(secrets.custom_password.as_deref(), Some("hunter2"));
}
#[test]
fn a_share_removed_on_one_device_is_removed_on_the_others() {
let laptop = Device::new("laptop");
let phone = Device::new("phone");
laptop.store.add("Anime", LINK, None).unwrap();
let (sent, _) = laptop.sync(&[], 10);
let (phone_sent, _) = phone.sync(&[&sent], 20);
laptop.store.remove("share-ABCDEFGHIJ").unwrap();
let (sent, _) = laptop.sync(&[&phone_sent], 30);
let (_, changes) = phone.sync(&[&sent], 40);
assert_eq!(changes.shares_removed, ["share-ABCDEFGHIJ"]);
assert!(phone.share_names().is_empty());
}
#[test]
fn a_fresh_installation_takes_the_library_rather_than_its_defaults() {
let laptop = Device::new("laptop");
let phone = Device::new("phone");
laptop.sync(&[], 10);
let mut prefs = crate::prefs::load(&laptop.store.dirs).unwrap();
prefs.autoplay_next = false;
prefs.speed = 1.5;
crate::prefs::save(&laptop.store.dirs, &prefs).unwrap();
let (sent, _) = laptop.sync(&[], 20);
// The new phone has written its defaults before it ever synced.
crate::prefs::save(&phone.store.dirs, &crate::prefs::PlaybackPrefs::default()).unwrap();
let (_, changes) = phone.sync(&[&sent], 30);
assert!(changes.settings);
let taken = crate::prefs::load(&phone.store.dirs).unwrap();
assert!(!taken.autoplay_next);
assert_eq!(taken.speed, 1.5);
}
#[test]
fn volume_and_mute_stay_on_the_device_that_set_them() {
let laptop = Device::new("laptop");
let phone = Device::new("phone");
laptop.sync(&[], 10);
let mut prefs = crate::prefs::PlaybackPrefs {
volume: 20.0,
muted: true,
..Default::default()
};
crate::prefs::save(&laptop.store.dirs, &prefs).unwrap();
let (sent, _) = laptop.sync(&[], 20);
prefs.volume = 80.0;
prefs.muted = false;
prefs.subtitles = false;
crate::prefs::save(&phone.store.dirs, &prefs).unwrap();
phone.sync(&[], 5);
phone.sync(&[&sent], 30);
let taken = crate::prefs::load(&phone.store.dirs).unwrap();
assert_eq!(taken.volume, 80.0);
assert!(!taken.muted);
assert!(taken.subtitles, "the laptop's later choice wins");
}
#[test]
fn the_newer_change_wins_whichever_device_syncs_first() {
let laptop = Device::new("laptop");
let phone = Device::new("phone");
laptop.sync(&[], 1);
phone.sync(&[], 1);
let mut appearance = crate::appearance::load(&laptop.store.dirs).unwrap();
appearance.gradients = false;
crate::appearance::save(&laptop.store.dirs, &appearance).unwrap();
let (from_laptop, _) = laptop.sync(&[], 10);
appearance.gradients = true;
appearance.accent = crate::appearance::Accent::Peach;
crate::appearance::save(&phone.store.dirs, &appearance).unwrap();
let (from_phone, _) = phone.sync(&[&from_laptop], 20);
laptop.sync(&[&from_phone], 30);
for device in [&laptop, &phone] {
let seen = crate::appearance::load(&device.store.dirs).unwrap();
assert_eq!(seen.accent, crate::appearance::Accent::Peach);
assert!(seen.gradients);
}
}
#[test]
fn an_unchanged_device_sends_the_same_registers_again() {
let laptop = Device::new("laptop");
laptop.store.add("Anime", LINK, None).unwrap();
let (first, _) = laptop.sync(&[], 10);
let (second, changes) = laptop.sync(&[&first], 20);
assert_eq!(first, second);
assert!(changes.is_empty());
}
#[test]
fn a_share_whose_secret_cannot_be_read_is_not_removed_elsewhere() {
let laptop = Device::new("laptop");
laptop.store.add("Anime", LINK, None).unwrap();
laptop.sync(&[], 10);
laptop.store.secrets.delete("share-ABCDEFGHIJ").unwrap();
let (sent, _) = laptop.sync(&[], 20);
assert!(sent["share/share-ABCDEFGHIJ"].value.is_some());
}
#[test]
fn a_hand_picked_match_and_a_shows_track_choice_travel() {
let laptop = Device::new("laptop");
let phone = Device::new("phone");
laptop.sync(&[], 1);
let metadata = TitleMetadata {
provider: ProviderId::AniList,
remote_id: "21".to_owned(),
name: "One Piece".to_owned(),
original_name: None,
overview: None,
year: Some(1999),
kind: crate::library::TitleKind::Series,
poster_url: None,
backdrop_url: None,
rating: None,
genres: Vec::new(),
episodes: None,
url: None,
details: None,
};
laptop
.catalog
.lock()
.set_metadata(&MetadataRecord {
title_key: "one piece".to_owned(),
provider: ProviderId::AniList,
metadata: Some(metadata),
fetched_at: 5,
manual: true,
})
.unwrap();
let tracks = TitleTrackPrefs {
audio_language: Some("jpn".to_owned()),
..Default::default()
};
laptop
.catalog
.lock()
.set_title_track_prefs("one piece", &tracks)
.unwrap();
let (sent, _) = laptop.sync(&[], 10);
let (_, changes) = phone.sync(&[&sent], 20);
assert_eq!(changes.titles, 2);
let catalog = phone.catalog.lock();
let record = catalog.metadata("one piece").unwrap().unwrap();
assert!(record.manual);
assert_eq!(record.metadata.unwrap().remote_id, "21");
assert_eq!(
catalog.title_track_prefs("one piece").unwrap(),
Some(tracks)
);
}
#[test]
fn a_synced_share_that_names_another_id_is_refused() {
let phone = Device::new("phone");
let mut forged = Registers::new();
forged.insert(
"share/share-OTHER".to_owned(),
Register {
value: Some(serde_json::json!({
"name": "x", "token": "ABCDEFGHIJ", "url": LINK
})),
updated_at: 10,
},
);
let (registers, changes) = phone.sync(&[&forged], 20);
assert!(changes.shares_added.is_empty());
assert!(!registers.contains_key("share/share-OTHER"));
}
}
crates/pstr-core/src/shares.rs
+68 -20
View File
@@ -355,19 +355,19 @@ impl AccountFolder {
}
/// The secrets for one share, as held in the credential store.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ShareSecrets {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct ShareSecrets {
/// The full share URL, fragment included.
url: String,
pub(crate) url: String,
/// The custom password, when the link has one.
#[serde(default)]
custom_password: Option<String>,
pub(crate) custom_password: Option<String>,
}
/// The configured shares, and their secrets.
pub struct ShareStore {
dirs: AppDirs,
secrets: Arc<dyn SecretStore>,
pub(crate) dirs: AppDirs,
pub(crate) secrets: Arc<dyn SecretStore>,
}
impl ShareStore {
@@ -396,9 +396,9 @@ impl ShareStore {
let mut shares = self.list()?;
if shares.iter().any(|share| share.id == id) {
return Err(Error::Config(format!(
"a share for token {token} is already configured"
)));
return Err(Error::Config(
"that link is already in the library".to_owned(),
));
}
let secrets = ShareSecrets {
@@ -421,6 +421,53 @@ impl ShareStore {
Ok(share)
}
/// [`Self::add`], but only once the link has been opened with the
/// password given.
///
/// A link that is mistyped, expired or behind the wrong password used to
/// be stored anyway, so the first try failed on the crawl and every retry
/// failed on "already configured" — the only way out was to remove a share
/// that had never worked. Opening first means a refused link leaves
/// nothing behind, and the client it opened is handed back for the crawl.
pub async fn add_verified(
&self,
name: &str,
url: &str,
custom_password: Option<&str>,
) -> Result<(Share, ShareClient)> {
let token = token_from_url(url)?;
let id = format!("share-{token}");
if self.list()?.iter().any(|share| share.id == id) {
return Err(Error::Config(
"that link is already in the library".to_owned(),
));
}
let custom_password = custom_password.filter(|password| !password.is_empty());
let client = open_link(url, custom_password).await?;
let share = self.add(name, url, custom_password)?;
Ok((share, client))
}
/// Record a share another installation has, with the secrets it opens
/// with, replacing what this one holds under the same id. See
/// [`crate::profile`].
pub(crate) fn put_synced(&self, share: Share, secrets: Option<&ShareSecrets>) -> Result<()> {
if let Some(secrets) = secrets {
self.store_secrets(&share.id, secrets)?;
}
let mut shares = self.list()?;
match shares.iter_mut().find(|existing| existing.id == share.id) {
Some(existing) => *existing = share,
None => shares.push(share),
}
write_json(&self.dirs.shares_file(), &shares)
}
/// A share's secrets, for syncing them to another installation.
pub(crate) fn secrets_of(&self, id: &str) -> Result<ShareSecrets> {
self.load_secrets(id)
}
/// Record a folder of the signed-in account as a share.
///
/// Nothing goes in the credential store: the folder opens through the
@@ -526,17 +573,7 @@ impl ShareStore {
});
}
let secrets = self.load_secrets(&share.id)?;
let client = ProtonDrivePublicLinkClient::open(
client_configuration(),
&secrets.url,
secrets.custom_password.as_deref(),
)
.await?
// The SDK default is sized for a background sync daemon. A player that
// seeks wants more blocks in flight; at 4 MiB each this is a 192 MiB
// ceiling, which is unremarkable for a desktop app.
.with_max_inflight_blocks(48);
Ok(ShareClient::PublicLink(Box::new(client)))
open_link(&secrets.url, secrets.custom_password.as_deref()).await
}
fn store_secrets(&self, id: &str, secrets: &ShareSecrets) -> Result<()> {
@@ -556,6 +593,17 @@ impl ShareStore {
}
}
/// Open a public link as a visitor.
async fn open_link(url: &str, custom_password: Option<&str>) -> Result<ShareClient> {
let client = ProtonDrivePublicLinkClient::open(client_configuration(), url, custom_password)
.await?
// The SDK default is sized for a background sync daemon. A player that
// seeks wants more blocks in flight; at 4 MiB each this is a 192 MiB
// ceiling, which is unremarkable for a desktop app.
.with_max_inflight_blocks(48);
Ok(ShareClient::PublicLink(Box::new(client)))
}
/// Several opened shares, presented as one library.
pub struct SharedLibrary {
clients: BTreeMap<String, ShareClient>,
crates/pstr-core/src/sync.rs
+151 -74
View File
@@ -1,5 +1,7 @@
//! Watch history kept in the viewer's own Drive, so every device resumes where
//! another left off.
//! another left off — and, in the same file, the shares, settings and
//! hand-picked matches that make the library look the same on every device
//! (see [`crate::profile`]).
//!
//! ## Layout
//!
@@ -17,7 +19,8 @@
//! nothing is overwritten and there is no write conflict to resolve.
//!
//! Each file carries the installation's whole watch table, and merging is per
//! episode: the newer `updated_at` wins. That is idempotent and order-free, so
//! episode: the newer `updated_at` wins. The profile's registers merge the same
//! way, per key. That is idempotent and order-free, so
//! it does not matter how many devices there are, which synced first, or that
//! a device writes back entries it learned from another. A device whose clock
//! runs ahead wins ties it should not, which is the price of not having a
@@ -25,11 +28,11 @@
//!
//! The keys travel: `share-{token}` is derived from the link and a folder's
//! `drive-…` id from its node, so the same library added on two devices has
//! the same `(share_id, link_id)` on both. Nothing secret is in a file — the
//! link token decrypts nothing without the fragment — and the file is
//! end-to-end encrypted by Drive regardless.
//! the same `(share_id, link_id)` on both. The watch entries carry nothing
//! secret; the profile's share registers do carry each link and its password,
//! which is why they live only here, end-to-end encrypted by Drive.
use std::collections::HashMap;
use std::collections::{BTreeMap, HashMap};
use parking_lot::Mutex;
use proton_drive_rs::{DeviceType, NodeKind, ProtonDriveClient};
@@ -40,6 +43,8 @@ use crate::account::Account;
use crate::catalog::{Catalog, WatchState};
use crate::config::{AppDirs, read_json, write_json};
use crate::error::{Error, Result};
use crate::profile::{Profile, ProfileChanges, Registers};
use crate::shares::ShareStore;
/// The Drive device watch history is kept on.
pub const DEVICE_NAME: &str = "proton-stream";
@@ -48,7 +53,8 @@ pub const DEVICE_NAME: &str = "proton-stream";
const FOLDER: &str = "watch-history";
/// Bumped only for a change an older build would misread. A file of a newer
/// version is skipped rather than half-understood.
/// version is skipped rather than half-understood. The profile's registers
/// were added without a bump: a build that predates them ignores the field.
const FORMAT_VERSION: u32 = 1;
/// One installation's watch history, as stored in Drive.
@@ -57,6 +63,15 @@ struct HistoryFile {
version: u32,
installation: String,
entries: Vec<HistoryEntry>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
registers: Registers,
}
/// What one history file holds.
#[derive(Debug, Clone, Default, PartialEq)]
struct Contents {
entries: Vec<HistoryEntry>,
registers: Registers,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -112,7 +127,7 @@ struct Identity {
}
/// What one [`WatchSync::sync`] did.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct SyncReport {
/// Episodes whose position another device had newer.
pub applied: usize,
@@ -123,6 +138,8 @@ pub struct SyncReport {
/// Files of installations gone quiet that were moved to the trash, their
/// history now carried in this installation's own file.
pub retired: usize,
/// What other devices' shares and settings changed here.
pub profile: ProfileChanges,
}
/// How long an installation's file may go unwritten before another
@@ -243,10 +260,10 @@ impl HistoryDrive for ProtonDriveClient {
}
/// A history file as last read, by the revision it was read at. `None`
/// entries for one this build could not read.
/// contents for one this build could not read.
struct Seen {
revision: Option<String>,
entries: Option<Vec<HistoryEntry>>,
contents: Option<Contents>,
}
/// Syncs the catalog's watch table with the account's Drive.
@@ -257,8 +274,9 @@ pub struct WatchSync {
seen: Mutex<HashMap<String, Seen>>,
/// What this installation's own file holds, once known: what was read
/// back at the first sync, then what was last written. Compared against
/// the catalog so an unchanged history is not uploaded as a new revision.
own: Mutex<Option<Vec<HistoryEntry>>>,
/// the catalog and the profile so an unchanged history is not uploaded as
/// a new revision.
own: Mutex<Option<Contents>>,
/// The history folders, resolved once per account. Usually one; see
/// [`HistoryDrive::history_folders`].
folders: Mutex<Option<Vec<NodeUid>>>,
@@ -303,25 +321,34 @@ impl WatchSync {
*self.folders.lock() = None;
}
/// Pull every other installation's history into `catalog`, write this
/// one's if it changed, then retire the files of installations gone quiet.
/// Pull every other installation's history, shares and settings in, write
/// this one's if it changed, then retire the files of installations gone
/// quiet.
///
/// The catalog is locked only around its own reads and writes, never
/// across a request, so playback can keep saving its position while this
/// runs.
pub async fn sync(&self, account: &Account, catalog: &Mutex<Catalog>) -> Result<SyncReport> {
/// runs. Shares that arrive are configured but not opened or crawled; see
/// [`SyncReport::profile`] for what the front end has to do about them.
pub async fn sync(
&self,
account: &Account,
catalog: &Mutex<Catalog>,
store: &ShareStore,
) -> Result<SyncReport> {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |since| since.as_secs() as i64);
self.sync_with(account.drive(), catalog, now).await
let profile = Profile { store, catalog };
self.sync_with(account.drive(), &profile, now).await
}
pub(crate) async fn sync_with<D: HistoryDrive>(
&self,
drive: &D,
catalog: &Mutex<Catalog>,
profile: &Profile<'_>,
now: i64,
) -> Result<SyncReport> {
let catalog = profile.catalog;
let _running = self.running.lock().await;
let folders = self.folders(drive).await?;
let own_name = format!("{}.json", self.installation);
@@ -339,12 +366,14 @@ impl WatchSync {
}
let mut remote = Vec::new();
let mut registers = Vec::new();
// Files whose every entry has been read, and so can be retired once
// this installation's file carries them.
let mut readable = Vec::new();
for file in &others {
if let Some(entries) = self.read(drive, file).await {
remote.extend(entries);
if let Some(contents) = self.read(drive, file).await {
remote.extend(contents.entries);
registers.push(contents.registers);
readable.push(file);
}
}
@@ -355,19 +384,22 @@ impl WatchSync {
// less than it holds, and read again next time.
let mut own_unreadable = false;
if self.own.lock().is_none() {
let entries = match &own_file {
let contents = match &own_file {
Some(file) => self.read(drive, file).await,
None => Some(Vec::new()),
None => Some(Contents::default()),
};
match entries {
Some(entries) => {
remote.extend(entries.iter().cloned());
*self.own.lock() = Some(entries);
match contents {
Some(contents) => {
remote.extend(contents.entries.iter().cloned());
registers.push(contents.registers.clone());
*self.own.lock() = Some(contents);
}
None => own_unreadable = true,
}
}
let (registers, profile_changes) = profile.sync(&registers, now)?;
let incoming: Vec<(String, String, WatchState)> = remote
.iter()
.filter(|entry| entry.is_sane())
@@ -388,15 +420,17 @@ impl WatchSync {
// A file that is missing is written even when what it would hold has
// not changed: another installation retired it, or it was deleted by
// hand, and either way the history it carried lives only here now.
let unchanged = own_file.is_some() && self.own.lock().as_ref() == Some(&entries);
let uploaded = if own_unreadable || unchanged || (entries.is_empty() && own_file.is_none())
{
let contents = Contents { entries, registers };
let unchanged = own_file.is_some() && self.own.lock().as_ref() == Some(&contents);
let empty = contents.entries.is_empty() && contents.registers.is_empty();
let uploaded = if own_unreadable || unchanged || (empty && own_file.is_none()) {
false
} else {
let file = HistoryFile {
version: FORMAT_VERSION,
installation: self.installation.clone(),
entries: entries.clone(),
entries: contents.entries.clone(),
registers: contents.registers.clone(),
};
let bytes = serde_json::to_vec(&file)
.map_err(|e| Error::Config(format!("serialize watch history: {e}")))?;
@@ -404,7 +438,7 @@ impl WatchSync {
Some(file) => drive.replace(&file.uid, &bytes).await?,
None => drive.create(&folders[0], &own_name, &bytes).await?,
}
*self.own.lock() = Some(entries);
*self.own.lock() = Some(contents);
true
};
@@ -435,34 +469,31 @@ impl WatchSync {
uploaded,
devices: others.len(),
retired: retired.len(),
profile: profile_changes,
})
}
/// The entries of one history file, from the cache when its revision has
/// The contents of one history file, from the cache when its revision has
/// not moved, or `None` when it cannot be read.
///
/// A file that will not download or parse is logged and skipped: it is
/// another device's, and one bad file must not stop the rest syncing.
async fn read<D: HistoryDrive>(
&self,
drive: &D,
file: &RemoteFile,
) -> Option<Vec<HistoryEntry>> {
async fn read<D: HistoryDrive>(&self, drive: &D, file: &RemoteFile) -> Option<Contents> {
let key = file.uid.link_id.as_str().to_owned();
if let Some(seen) = self.seen.lock().get(&key)
&& file.revision.is_some()
&& seen.revision == file.revision
{
return seen.entries.clone();
return seen.contents.clone();
}
let uid = &file.uid;
let entries = match drive.download(uid).await {
let contents = match drive.download(uid).await {
Ok(bytes) => {
let entries = parse(&bytes);
if entries.is_none() {
let contents = parse(&bytes);
if contents.is_none() {
tracing::warn!(%uid, "skipping a watch-history file this build cannot read");
}
entries
contents
}
Err(error) => {
// Not cached: a download that failed is tried again next time.
@@ -474,10 +505,10 @@ impl WatchSync {
key,
Seen {
revision: file.revision.clone(),
entries: entries.clone(),
contents: contents.clone(),
},
);
entries
contents
}
/// The history folders, resolved once per account.
@@ -491,10 +522,13 @@ impl WatchSync {
}
}
/// A history file's entries, or `None` for one this build cannot read.
fn parse(bytes: &[u8]) -> Option<Vec<HistoryEntry>> {
/// A history file's contents, or `None` for one this build cannot read.
fn parse(bytes: &[u8]) -> Option<Contents> {
let file: HistoryFile = serde_json::from_slice(bytes).ok()?;
(file.version <= FORMAT_VERSION).then_some(file.entries)
(file.version <= FORMAT_VERSION).then_some(Contents {
entries: file.entries,
registers: file.registers,
})
}
/// The kind of device to register, which is only the icon the web app draws.
@@ -518,6 +552,7 @@ fn new_installation_id() -> Result<String> {
#[cfg(test)]
mod tests {
use super::*;
use crate::profile::tests::Device;
fn entry(link: &str, position: f64, updated_at: i64) -> HistoryEntry {
HistoryEntry {
@@ -655,11 +690,8 @@ mod tests {
}
/// An installation: its sync and its catalog.
fn device(id: &str) -> (WatchSync, Mutex<Catalog>) {
(
WatchSync::with_installation(id.to_owned()),
Mutex::new(Catalog::in_memory().unwrap()),
)
fn device(id: &str) -> (WatchSync, Device) {
(WatchSync::with_installation(id.to_owned()), Device::new(id))
}
fn watched_at(position_secs: f64, updated_at: i64) -> WatchState {
@@ -684,11 +716,14 @@ mod tests {
.unwrap();
let first = laptop
.sync_with(&drive, &laptop_catalog, 100)
.sync_with(&drive, &laptop_catalog.profile(), 100)
.await
.unwrap();
assert!(first.uploaded);
let second = phone.sync_with(&drive, &phone_catalog, 110).await.unwrap();
let second = phone
.sync_with(&drive, &phone_catalog.profile(), 110)
.await
.unwrap();
assert_eq!(second.applied, 1);
assert_eq!(second.devices, 1);
@@ -714,12 +749,15 @@ mod tests {
.unwrap();
laptop
.sync_with(&drive, &laptop_catalog, 300)
.sync_with(&drive, &laptop_catalog.profile(), 300)
.await
.unwrap();
phone
.sync_with(&drive, &phone_catalog.profile(), 300)
.await
.unwrap();
phone.sync_with(&drive, &phone_catalog, 300).await.unwrap();
laptop
.sync_with(&drive, &laptop_catalog, 300)
.sync_with(&drive, &laptop_catalog.profile(), 300)
.await
.unwrap();
@@ -740,8 +778,14 @@ mod tests {
.set_watch_state("share-a", "ep1", &watched_at(60.0, 100))
.unwrap();
laptop.sync_with(&drive, &catalog, 100).await.unwrap();
let again = laptop.sync_with(&drive, &catalog, 200).await.unwrap();
laptop
.sync_with(&drive, &catalog.profile(), 100)
.await
.unwrap();
let again = laptop
.sync_with(&drive, &catalog.profile(), 200)
.await
.unwrap();
assert!(!again.uploaded);
assert_eq!(drive.uploads(), 1);
@@ -751,7 +795,10 @@ mod tests {
async fn nothing_is_uploaded_before_anything_was_watched() {
let drive = FakeDrive::default();
let (laptop, catalog) = device("laptop");
let report = laptop.sync_with(&drive, &catalog, 100).await.unwrap();
let report = laptop
.sync_with(&drive, &catalog.profile(), 100)
.await
.unwrap();
assert!(!report.uploaded);
assert!(drive.names().is_empty());
}
@@ -764,11 +811,17 @@ mod tests {
.lock()
.set_watch_state("share-a", "ep1", &watched_at(60.0, 100))
.unwrap();
laptop.sync_with(&drive, &catalog, 100).await.unwrap();
laptop
.sync_with(&drive, &catalog.profile(), 100)
.await
.unwrap();
let own: Vec<NodeUid> = drive.files.lock().iter().map(|f| f.uid.clone()).collect();
drive.trash(&own).await.unwrap();
let report = laptop.sync_with(&drive, &catalog, 200).await.unwrap();
let report = laptop
.sync_with(&drive, &catalog.profile(), 200)
.await
.unwrap();
assert!(report.uploaded);
assert_eq!(drive.names(), ["laptop.json"]);
@@ -782,11 +835,17 @@ mod tests {
.lock()
.set_watch_state("share-a", "ep1", &watched_at(60.0, 100))
.unwrap();
laptop.sync_with(&drive, &catalog, 100).await.unwrap();
laptop
.sync_with(&drive, &catalog.profile(), 100)
.await
.unwrap();
// The catalog was deleted; the config, and with it the id, was not.
let (relaunched, empty) = device("laptop");
let report = relaunched.sync_with(&drive, &empty, 200).await.unwrap();
let report = relaunched
.sync_with(&drive, &empty.profile(), 200)
.await
.unwrap();
assert_eq!(report.applied, 1);
assert!(!report.uploaded);
@@ -804,7 +863,10 @@ mod tests {
.lock()
.set_watch_state("share-a", "ep1", &watched_at(60.0, 100))
.unwrap();
laptop.sync_with(&drive, &catalog, 100).await.unwrap();
laptop
.sync_with(&drive, &catalog.profile(), 100)
.await
.unwrap();
let (relaunched, other) = device("laptop");
other
@@ -814,13 +876,19 @@ mod tests {
drive
.offline
.store(true, std::sync::atomic::Ordering::SeqCst);
let report = relaunched.sync_with(&drive, &other, 200).await.unwrap();
let report = relaunched
.sync_with(&drive, &other.profile(), 200)
.await
.unwrap();
assert!(!report.uploaded);
drive
.offline
.store(false, std::sync::atomic::Ordering::SeqCst);
relaunched.sync_with(&drive, &other, 300).await.unwrap();
relaunched
.sync_with(&drive, &other.profile(), 300)
.await
.unwrap();
let state = other.lock().all_watch_states().unwrap();
assert_eq!(state.len(), 2, "both episodes survive: {state:?}");
}
@@ -832,6 +900,7 @@ mod tests {
version: FORMAT_VERSION + 1,
installation: "future".to_owned(),
entries: vec![entry("ep1", 60.0, 100)],
registers: Registers::new(),
};
drive.put("future.json", serde_json::to_vec(&future).unwrap());
let (laptop, catalog) = device("laptop");
@@ -841,7 +910,7 @@ mod tests {
.unwrap();
let report = laptop
.sync_with(&drive, &catalog, RETIRE_AFTER_SECS + DAY)
.sync_with(&drive, &catalog.profile(), RETIRE_AFTER_SECS + DAY)
.await
.unwrap();
@@ -860,21 +929,24 @@ mod tests {
.unwrap();
drive.at(100);
old_phone
.sync_with(&drive, &old_catalog, 100)
.sync_with(&drive, &old_catalog.profile(), 100)
.await
.unwrap();
let (phone, catalog) = device("phone");
let later = 100 + RETIRE_AFTER_SECS;
drive.at(later);
let report = phone.sync_with(&drive, &catalog, later).await.unwrap();
let report = phone
.sync_with(&drive, &catalog.profile(), later)
.await
.unwrap();
assert_eq!(report.retired, 1);
assert_eq!(drive.names(), ["phone.json"]);
// Nothing was lost: the newcomer's file carries it now.
let (fresh, fresh_catalog) = device("fresh");
fresh
.sync_with(&drive, &fresh_catalog, later)
.sync_with(&drive, &fresh_catalog.profile(), later)
.await
.unwrap();
assert_eq!(
@@ -893,13 +965,13 @@ mod tests {
.unwrap();
drive.at(100);
laptop
.sync_with(&drive, &laptop_catalog, 100)
.sync_with(&drive, &laptop_catalog.profile(), 100)
.await
.unwrap();
let (phone, catalog) = device("phone");
let report = phone
.sync_with(&drive, &catalog, 100 + RETIRE_AFTER_SECS - DAY)
.sync_with(&drive, &catalog.profile(), 100 + RETIRE_AFTER_SECS - DAY)
.await
.unwrap();
@@ -925,9 +997,13 @@ mod tests {
version: FORMAT_VERSION,
installation: "abc".to_owned(),
entries: vec![entry("a", 12.5, 100)],
registers: Registers::new(),
};
let bytes = serde_json::to_vec(&file).unwrap();
assert_eq!(parse(&bytes), Some(file.entries));
assert_eq!(
parse(&bytes).map(|contents| contents.entries),
Some(file.entries)
);
}
#[test]
@@ -936,6 +1012,7 @@ mod tests {
version: FORMAT_VERSION + 1,
installation: "abc".to_owned(),
entries: vec![entry("a", 12.5, 100)],
registers: Registers::new(),
};
assert_eq!(parse(&serde_json::to_vec(&file).unwrap()), None);
assert_eq!(parse(b"not json"), None);