DarkRoom/ui/dr-ui/src/derived_sync.rs

//! TRACES: FR-CAT-3 | FR-CAT-7 | FR-NC-7
//! Pushing derived state to Nextcloud: thumbnail shards and the catalog.
//!
//! # What travels, and why only this
//!
//! Sidecars are handled elsewhere ([`crate::library::spawn_sidecar_writes`])
//! and are the *authoritative* store — they are the reason a catalog can be
//! deleted and rebuilt (ARCH §6.12). What moves here is derived state that is
//! merely expensive:
//!
//! - **Thumbnail shards.** A thumbnail costs a range fetch plus a decode, and
//!   is byte-identical for every client looking at the same file. A second
//!   device that downloads the shards gets a full grid without touching a
//!   single RAW — hours of indexing against a few hundred MB of transfer.
//! - **The catalog**, for its collections. Every other thing the catalog holds
//!   has authoritative backing in a sidecar; a manually assembled collection
//!   does not, so without this it exists on one machine only.
//!
//! # Why sealed shards make this cheap
//!
//! A shard stops being written once it reaches its cap, and is never rewritten
//! after — deleting a thumbnail tombstones it in the index rather than editing
//! the sealed blob. So a client that has downloaded a sealed shard never needs
//! to ask about it again, and an up-to-date client transfers only the index and
//! whichever shard is currently open. That is the whole reason for sharding at
//! 25 MB rather than keeping one growing file.
//!
//! # Where it lives
//!
//! Under the library root, in a dotted folder beside the trash. The root is the
//! only place the user granted access to, and writing outside it may cross a
//! share boundary the account cannot write to. The scanner excludes it by the
//! same mechanism that excludes the trash.

use std::path::{Path, PathBuf};

use dr_sync::{RemoteBackend, RemoteId, RemotePath};
use dr_sync_nextcloud::{AppCredentials, NextcloudBackend};
use dr_thumbs::ThumbStore;

/// Folder under the library root holding derived state.
///
/// Defined by the scanner, which must exclude it: a walk that indexed this
/// folder would pay a listing for it on every sync of every device.
pub use dr_sync::scan::DERIVED_DIR;

/// What a sync pass did, for logging and for telling the user.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct SyncReport {
    pub shards_uploaded: usize,
    pub shards_downloaded: usize,
    pub thumbnails_adopted: usize,
    pub catalog_uploaded: bool,
    pub catalog_merged: bool,
    pub collections_gained: usize,
}

impl SyncReport {
    pub fn did_anything(&self) -> bool {
        self.shards_uploaded > 0
            || self.shards_downloaded > 0
            || self.catalog_uploaded
            || self.catalog_merged
    }
}

/// Progress from the sync worker.
#[derive(Debug)]
pub enum SyncMessage {
    Status(String),
    Finished(Box<SyncReport>),
    Failed(String),
}

/// Push shards and the catalog, and take anything newer from the server.
///
/// Runs on its own thread with its own runtime, like every other network path
/// here — the Slint loop must never block (NFR-P9).
pub fn spawn_sync(
    creds: AppCredentials,
    user_id: String,
    root: String,
    thumbs_dir: PathBuf,
    catalog_path: PathBuf,
    scratch: PathBuf,
) -> std::sync::mpsc::Receiver<SyncMessage> {
    let (tx, rx) = std::sync::mpsc::channel();

    std::thread::spawn(move || {
        // A current-thread runtime here is what made the library scan itself
        // rather than adopt the shards the server already had; see net_runtime.
        let rt = match crate::net_runtime::build() {
            Ok(rt) => rt,
            Err(e) => {
                let _ = tx.send(SyncMessage::Failed(e.to_string()));
                return;
            }
        };

        rt.block_on(async {
            let backend = match NextcloudBackend::new(&creds, &user_id) {
                Ok(b) => b,
                Err(e) => {
                    let _ = tx.send(SyncMessage::Failed(e.to_string()));
                    return;
                }
            };

            match run(&backend, &root, &thumbs_dir, &catalog_path, &scratch, &tx).await {
                Ok(report) => {
                    let _ = tx.send(SyncMessage::Finished(Box::new(report)));
                }
                Err(e) => {
                    let _ = tx.send(SyncMessage::Failed(e));
                }
            }
        });
    });

    rx
}

async fn run(
    backend: &NextcloudBackend,
    root: &str,
    thumbs_dir: &Path,
    catalog_path: &Path,
    scratch: &Path,
    tx: &std::sync::mpsc::Sender<SyncMessage>,
) -> Result<SyncReport, String> {
    let mut report = SyncReport::default();
    let base = derived_path(root);

    // The folder may not exist on a first sync. Creating it unconditionally is
    // cheaper than probing, and an existing folder is not an error.
    let _ = backend.create_dir(&base).await;

    let _ = tx.send(SyncMessage::Status("checking thumbnails…".into()));
    sync_shards(backend, &base, thumbs_dir, scratch, &mut report).await?;

    let _ = tx.send(SyncMessage::Status("checking collections…".into()));
    sync_catalog(backend, &base, catalog_path, scratch, &mut report).await?;

    Ok(report)
}

/// The derived folder for a library root.
fn derived_path(root: &str) -> RemotePath {
    if root.is_empty() {
        RemotePath::new(DERIVED_DIR)
    } else {
        RemotePath::new(format!("{root}/{DERIVED_DIR}"))
    }
}

/// Exchange thumbnail shards with the server.
///
/// Upload what the server lacks, download what we lack. Sealed shards are
/// immutable, so a name match is a content match and nothing needs comparing
/// beyond existence — which is what keeps a steady-state sync to one listing.
async fn sync_shards(
    backend: &NextcloudBackend,
    base: &RemotePath,
    thumbs_dir: &Path,
    scratch: &Path,
    report: &mut SyncReport,
) -> Result<(), String> {
    let store = match ThumbStore::open(thumbs_dir) {
        Ok(s) => s,
        Err(e) => {
            // No local store is not a failure: a fresh device has nothing to
            // upload and everything to gain from downloading.
            log::debug!("thumbnail store unavailable: {e}");
            return Ok(());
        }
    };

    let remote: std::collections::HashMap<String, u64> = backend
        .list(base, None)
        .await
        .map(|entries| {
            entries
                .into_iter()
                .filter(|e| e.kind == dr_sync::EntryKind::File)
                .map(|e| (e.path.name().to_string(), e.size))
                .collect()
        })
        // A missing folder lists as an error on some servers; treat it as empty
        // rather than aborting a first sync.
        .unwrap_or_default();

    let local = store.shards().map_err(|e| e.to_string())?;

    // ---- upload ----------------------------------------------------------
    for shard in &local {
        let path = store.shard_path(shard.id);
        let Ok(bytes) = std::fs::read(&path) else {
            continue;
        };
        let name = shard_name(shard.id);

        // A sealed shard the server already has is byte-identical by
        // construction, so its presence is proof enough. The open shard is
        // re-uploaded whenever its size differs, which is the only way it
        // changes.
        let skip = match remote.get(&name) {
            Some(_) if shard.sealed => true,
            Some(size) => *size == bytes.len() as u64,
            None => false,
        };
        if skip {
            continue;
        }

        let target = RemotePath::new(format!("{}/{name}", base.as_str()));
        match backend.put(&target, bytes, None).await {
            Ok(_) => report.shards_uploaded += 1,
            // One shard failing must not abort the rest: they are independent
            // and the next pass retries.
            Err(e) => log::warn!("uploading {name}: {e}"),
        }
    }

    // ---- download --------------------------------------------------------
    let have: std::collections::HashSet<u32> = local.iter().map(|s| s.id).collect();
    let mut store = store;

    for name in remote.keys() {
        let Some(id) = shard_id(name) else { continue };
        if have.contains(&id) {
            continue;
        }

        let source = RemotePath::new(format!("{}/{name}", base.as_str()));
        let bytes = match backend.get(&RemoteId::Path(source), None).await {
            Ok(b) => b,
            Err(e) => {
                log::warn!("downloading {name}: {e}");
                continue;
            }
        };

        // Written to scratch and merged, rather than dropped into the store
        // directory: a downloaded shard's *id* is the other device's numbering,
        // and two devices independently fill shard 0.
        let tmp = scratch.join(name);
        if std::fs::write(&tmp, &bytes).is_err() {
            continue;
        }
        match store.merge_shard(&tmp) {
            Ok(n) => {
                report.shards_downloaded += 1;
                report.thumbnails_adopted += n;
            }
            Err(e) => log::warn!("merging {name}: {e}"),
        }
        let _ = std::fs::remove_file(&tmp);
    }

    Ok(())
}

/// Exchange the catalog, for its collections.
///
/// Only collections merge — see [`dr_catalog::sync`]. The rest of a catalog
/// describes local state (folder ETags, cache paths, job rows) and importing
/// another device's version would be actively wrong.
async fn sync_catalog(
    backend: &NextcloudBackend,
    base: &RemotePath,
    catalog_path: &Path,
    scratch: &Path,
    report: &mut SyncReport,
) -> Result<(), String> {
    let remote_name = "catalog.sqlite";
    let target = RemotePath::new(format!("{}/{remote_name}", base.as_str()));

    // ---- take theirs first -----------------------------------------------
    //
    // Merging before uploading means our upload carries the union rather than
    // only our own half, so a third device syncing next gets everything in one
    // fetch.
    if let Ok(bytes) = backend.get(&RemoteId::Path(target.clone()), None).await {
        let downloaded = scratch.join("catalog-remote.sqlite");
        if std::fs::write(&downloaded, &bytes).is_ok() {
            match dr_catalog::Catalog::open(catalog_path) {
                Ok(catalog) => match catalog.merge_remote_catalog(&downloaded) {
                    Ok(merge) => {
                        report.catalog_merged = true;
                        report.collections_gained = merge.inserted + merge.updated;
                    }
                    Err(e) => log::warn!("merging remote catalog: {e}"),
                },
                Err(e) => log::warn!("opening catalog to merge: {e}"),
            }
            let _ = std::fs::remove_file(&downloaded);
        }
    }

    // ---- then push ours --------------------------------------------------
    //
    // Never the live file: committed transactions can sit in the `-wal` with
    // the main file lagging, so copying it uploads a torn snapshot. The backup
    // API serialises against writers instead of racing them.
    let snapshot = scratch.join("catalog-upload.sqlite");
    let catalog = dr_catalog::Catalog::open(catalog_path).map_err(|e| e.to_string())?;
    catalog
        .snapshot_for_upload(&snapshot)
        .map_err(|e| e.to_string())?;

    let bytes = std::fs::read(&snapshot).map_err(|e| e.to_string())?;
    match backend.put(&target, bytes, None).await {
        Ok(_) => report.catalog_uploaded = true,
        Err(e) => log::warn!("uploading catalog: {e}"),
    }
    let _ = std::fs::remove_file(&snapshot);

    Ok(())
}

fn shard_name(id: u32) -> String {
    format!("shard-{id:04}.sqlite")
}

/// The shard id in a filename, or `None` if it is not a shard.
///
/// Guards the download loop against adopting the catalog, a stray file, or
/// anything else the folder happens to contain.
fn shard_id(name: &str) -> Option<u32> {
    name.strip_prefix("shard-")?
        .strip_suffix(".sqlite")?
        .parse()
        .ok()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn derived_folder_sits_under_the_library_root() {
        // Outside the root the account may not have write access — the root is
        // the only thing the user granted.
        assert_eq!(
            derived_path("PhotosRaw").as_str(),
            "PhotosRaw/.darkroom-derived"
        );
        // A library at the account root still gets a relative path.
        assert_eq!(derived_path("").as_str(), ".darkroom-derived");
    }

    #[test]
    fn shard_names_round_trip() {
        assert_eq!(shard_name(0), "shard-0000.sqlite");
        assert_eq!(shard_name(42), "shard-0042.sqlite");
        assert_eq!(shard_id("shard-0042.sqlite"), Some(42));
        assert_eq!(shard_id(&shard_name(7)), Some(7));
    }

    #[test]
    fn non_shard_files_are_not_adopted() {
        // The folder also holds the catalog; downloading it as a shard would
        // hand a catalog to the thumbnail merger.
        assert_eq!(shard_id("catalog.sqlite"), None);
        assert_eq!(shard_id("shard-0000.sqlite-wal"), None);
        assert_eq!(shard_id("notes.txt"), None);
        assert_eq!(shard_id("shard-abc.sqlite"), None);
    }

    #[test]
    fn a_report_that_did_nothing_says_so() {
        assert!(!SyncReport::default().did_anything());
        assert!(SyncReport {
            shards_uploaded: 1,
            ..Default::default()
        }
        .did_anything());
        // Adopting thumbnails without moving a shard cannot happen, but the
        // report must not claim work on collections alone either.
        assert!(SyncReport {
            catalog_merged: true,
            ..Default::default()
        }
        .did_anything());
    }
}