AkurAI-Drive / client/src/sync.rs

Latest change 2d36cd6c72fc6a63a7edb87bb54a4020e50f2f8a - desktop: Tauri v2 tray + Google-Drive-style file browser app by olibuijr

//! Sync orchestration: gather the three snapshots, run the pure decision
//! engine, then execute the resulting actions against the backend and disk.

use anyhow::{bail, Context, Result};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use crate::api::Api;
use crate::config::{state_db_path, Config};
use crate::engine::{self, Action, LocalEntry, RemoteEntry, StateEntry};
use crate::state::{Record, State};
use crate::util::{self, TMP_PREFIX};

/// A local file as walked from disk.
struct LocalFile {
    sha256: String,
    size: u64,
    mtime: i64,
}

#[derive(Debug, Default, Clone)]
pub struct Summary {
    pub uploaded: u32,
    pub downloaded: u32,
    pub deleted_remote: u32,
    pub deleted_local: u32,
    pub adopted: u32,
    pub conflicts: u32,
    pub errors: u32,
}

impl Summary {
    pub fn to_line(&self) -> String {
        format!(
            "uploaded={} downloaded={} deleted_remote={} deleted_local={} adopted={} conflicts={} errors={}",
            self.uploaded,
            self.downloaded,
            self.deleted_remote,
            self.deleted_local,
            self.adopted,
            self.conflicts,
            self.errors
        )
    }
}

/// Run one full bidirectional sync. When `quiet`, suppress the per-file and
/// summary prints (used for the daemon's frequent re-syncs).
pub fn run_sync(cfg: &Config, quiet: bool) -> Result<Summary> {
    let folder = require_folder(cfg)?;
    let bucket = cfg
        .bucket
        .clone()
        .context("no bucket configured — run `akurai-drive-sync init <folder> --bucket <name>`")?;
    let token = cfg
        .token
        .clone()
        .filter(|t| !t.is_empty())
        .context("not logged in — run `akurai-drive-sync login`")?;

    let api = Api::new(cfg.resolved_api(), token)?;
    let state = State::open(&state_db_path()?)?;
    let st_map = state.load()?;

    // ── Snapshot: remote ───────────────────────────────────────────────────
    let remote_files = api.files(&bucket, "")?;
    let remote: BTreeMap<String, RemoteEntry> = remote_files
        .into_iter()
        .map(|f| {
            (
                f.key,
                RemoteEntry {
                    etag: f.etag,
                    size: f.size,
                },
            )
        })
        .collect();

    // ── Snapshot: local (hashing fast-path reuses unchanged state) ─────────
    let local = gather_local(&folder, &st_map)?;

    // ── Build engine inputs and compute the plan ──────────────────────────
    let state_engine: BTreeMap<String, StateEntry> = st_map
        .iter()
        .map(|(k, r)| {
            (
                k.clone(),
                StateEntry {
                    sha256: r.sha256.clone(),
                    etag: r.etag.clone(),
                    size: r.size,
                },
            )
        })
        .collect();
    let local_engine: BTreeMap<String, LocalEntry> = local
        .iter()
        .map(|(k, f)| {
            (
                k.clone(),
                LocalEntry {
                    sha256: f.sha256.clone(),
                    size: f.size,
                },
            )
        })
        .collect();

    let plan = engine::plan_actions(&state_engine, &remote, &local_engine);
    tracing::debug!(
        bucket = %bucket,
        remote = remote.len(),
        local = local.len(),
        actions = plan.len(),
        "sync plan computed"
    );

    // ── Execute ────────────────────────────────────────────────────────────
    let exec = Executor {
        api: &api,
        state: &state,
        folder: &folder,
        bucket: &bucket,
        remote: &remote,
        local: &local,
        quiet,
    };
    let mut summary = Summary::default();
    for action in plan {
        if let Err(e) = exec.run(&action, &mut summary) {
            summary.errors += 1;
            let detail = format!("{e:#}");
            tracing::error!(key = %action.key(), error = %detail, "sync action failed");
            eprintln!("  ! {} failed: {detail}", action.key());
        }
    }

    state.meta_set("last_sync", &summary.to_line())?;
    state.meta_set("last_sync_at", &now_rfc3339())?;
    tracing::info!(summary = %summary.to_line(), "sync pass complete");

    if !quiet {
        print_summary(&summary);
    }
    Ok(summary)
}

/// Walk the sync folder into a key→LocalFile map, skipping temp files. Reuses a
/// stored SHA-256 when size and mtime are unchanged to avoid re-hashing.
fn gather_local(
    folder: &Path,
    state: &BTreeMap<String, Record>,
) -> Result<BTreeMap<String, LocalFile>> {
    let mut map = BTreeMap::new();
    for entry in walkdir::WalkDir::new(folder)
        .into_iter()
        .filter_map(|e| e.ok())
    {
        if !entry.file_type().is_file() {
            continue;
        }
        let name = entry.file_name().to_string_lossy();
        if name.starts_with(TMP_PREFIX) {
            continue;
        }
        let path = entry.path();
        let rel = match path.strip_prefix(folder) {
            Ok(r) => r,
            Err(_) => continue,
        };
        let key = rel
            .to_string_lossy()
            .replace(std::path::MAIN_SEPARATOR, "/");
        if key.is_empty() {
            continue;
        }

        let meta = entry.metadata()?;
        let size = meta.len();
        let mtime = util::mtime_millis(path)?;

        let sha256 = match state.get(&key) {
            Some(rec) if rec.size == size && rec.mtime == mtime => rec.sha256.clone(),
            _ => util::sha256_file(path)?,
        };

        map.insert(
            key,
            LocalFile {
                sha256,
                size,
                mtime,
            },
        );
    }
    Ok(map)
}

/// Borrowed bundle of everything an action needs, so executing each action is a
/// single method call rather than a many-argument function.
struct Executor<'a> {
    api: &'a Api,
    state: &'a State,
    folder: &'a Path,
    bucket: &'a str,
    remote: &'a BTreeMap<String, RemoteEntry>,
    local: &'a BTreeMap<String, LocalFile>,
    quiet: bool,
}

impl Executor<'_> {
    fn run(&self, action: &Action, summary: &mut Summary) -> Result<()> {
        match action {
            Action::Upload(key) => self.upload(key, summary),
            Action::Download(key) => self.download(key, summary),
            Action::DeleteRemote(key) => self.delete_remote(key, summary),
            Action::DeleteLocal(key) => self.delete_local(key, summary),
            Action::Adopt(key) => self.adopt(key, summary),
            Action::DropState(key) => {
                self.state.remove(key)?;
                Ok(())
            }
            Action::Conflict(key) => self.conflict(key, summary),
        }
    }

    fn upload(&self, key: &str, summary: &mut Summary) -> Result<()> {
        let lf = self
            .local
            .get(key)
            .context("missing local file for upload")?;
        let ct = util::content_type_for(key);
        let presign = self.api.presign(self.bucket, key, "put", Some(ct))?;
        let abspath = self.folder.join(key);
        let etag = self
            .api
            .upload_file(&presign, &abspath, Some(ct))?
            .unwrap_or_default();
        self.state.upsert(
            key,
            &Record {
                sha256: lf.sha256.clone(),
                etag,
                size: lf.size,
                mtime: lf.mtime,
            },
        )?;
        summary.uploaded += 1;
        self.log(format!("↑ {key}"));
        Ok(())
    }

    fn download(&self, key: &str, summary: &mut Summary) -> Result<()> {
        let re = self
            .remote
            .get(key)
            .context("missing remote entry for download")?;
        let abspath = self.folder.join(key);
        self.download_atomic(key, &abspath)?;
        self.record_from_disk(key, &abspath, &re.etag)?;
        summary.downloaded += 1;
        self.log(format!("↓ {key}"));
        Ok(())
    }

    fn delete_remote(&self, key: &str, summary: &mut Summary) -> Result<()> {
        self.api.delete(self.bucket, key)?;
        self.state.remove(key)?;
        summary.deleted_remote += 1;
        self.log(format!("⊘ remote {key}"));
        Ok(())
    }

    fn delete_local(&self, key: &str, summary: &mut Summary) -> Result<()> {
        let abspath = self.folder.join(key);
        match std::fs::remove_file(&abspath) {
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e).with_context(|| format!("removing {}", abspath.display())),
        }
        prune_empty_dirs(self.folder, &abspath);
        self.state.remove(key)?;
        summary.deleted_local += 1;
        self.log(format!("⊘ local {key}"));
        Ok(())
    }

    fn adopt(&self, key: &str, summary: &mut Summary) -> Result<()> {
        let lf = self
            .local
            .get(key)
            .context("missing local file for adopt")?;
        let re = self
            .remote
            .get(key)
            .context("missing remote entry for adopt")?;
        self.state.upsert(
            key,
            &Record {
                sha256: lf.sha256.clone(),
                etag: re.etag.clone(),
                size: lf.size,
                mtime: lf.mtime,
            },
        )?;
        summary.adopted += 1;
        self.log(format!("= {key} (already in sync)"));
        Ok(())
    }

    /// Keep-both conflict resolution: move the local copy aside under a
    /// `(conflict <id>)` name and upload it, then download the remote version
    /// to the original path.
    fn conflict(&self, key: &str, summary: &mut Summary) -> Result<()> {
        let lf = self
            .local
            .get(key)
            .context("missing local file for conflict")?;
        let re = self
            .remote
            .get(key)
            .context("missing remote entry for conflict")?;
        let sid = util::short_id(&format!("{key}:{}", re.etag));
        let conflict_key = engine::conflict_name(key, &sid);

        let orig_abs = self.folder.join(key);
        let conflict_abs = self.folder.join(&conflict_key);

        // 1. Move the local edit aside.
        std::fs::rename(&orig_abs, &conflict_abs).with_context(|| {
            format!(
                "renaming {} -> {}",
                orig_abs.display(),
                conflict_abs.display()
            )
        })?;

        // 2. Download the remote version to the original path.
        self.download_atomic(key, &orig_abs)?;
        self.record_from_disk(key, &orig_abs, &re.etag)?;

        // 3. Upload the kept-aside local copy so it propagates everywhere.
        let ct = util::content_type_for(&conflict_key);
        let presign = self
            .api
            .presign(self.bucket, &conflict_key, "put", Some(ct))?;
        let cetag = self
            .api
            .upload_file(&presign, &conflict_abs, Some(ct))?
            .unwrap_or_default();
        self.state.upsert(
            &conflict_key,
            &Record {
                sha256: lf.sha256.clone(),
                etag: cetag,
                size: lf.size,
                mtime: util::mtime_millis(&conflict_abs)?,
            },
        )?;

        summary.conflicts += 1;
        // Conflicts are always reported, even in quiet mode.
        tracing::warn!(%key, %conflict_key, "conflict: kept remote, saved local copy aside");
        eprintln!("  ⚠ conflict on {key}: kept remote; local copy saved as {conflict_key}");
        Ok(())
    }

    /// Download a presigned GET to a temp sibling, then atomically rename.
    fn download_atomic(&self, key: &str, dest: &Path) -> Result<()> {
        if let Some(parent) = dest.parent() {
            std::fs::create_dir_all(parent)
                .with_context(|| format!("creating {}", parent.display()))?;
        }
        let presign = self.api.presign(self.bucket, key, "get", None)?;
        let tmp = temp_sibling(dest);
        self.api.download_to(&presign, &tmp)?;
        std::fs::rename(&tmp, dest)
            .with_context(|| format!("finalizing download of {}", dest.display()))?;
        Ok(())
    }

    /// Record state for a key from the file currently on disk plus a remote etag.
    fn record_from_disk(&self, key: &str, abspath: &Path, etag: &str) -> Result<()> {
        let sha256 = util::sha256_file(abspath)?;
        let size = std::fs::metadata(abspath)?.len();
        let mtime = util::mtime_millis(abspath)?;
        self.state.upsert(
            key,
            &Record {
                sha256,
                etag: etag.to_string(),
                size,
                mtime,
            },
        )?;
        Ok(())
    }

    fn log(&self, msg: String) {
        // Always record the action in the structured log (key + arrow), even in
        // quiet mode — the desktop app drives quiet syncs and diagnoses from logs.
        tracing::info!(action = %msg, "sync action");
        if !self.quiet {
            println!("  {msg}");
        }
    }
}

/// Temp file alongside the destination (same dir ⇒ same filesystem ⇒ atomic
/// rename). Named with the skip-prefix so the walker ignores it.
fn temp_sibling(dest: &Path) -> PathBuf {
    let parent = dest.parent().unwrap_or_else(|| Path::new("."));
    let id = util::short_id(&dest.to_string_lossy());
    parent.join(format!("{TMP_PREFIX}{id}"))
}

/// Remove now-empty parent directories of a deleted file, up to (not including)
/// the sync-folder root. Best-effort; failures are ignored.
fn prune_empty_dirs(root: &Path, removed: &Path) {
    let mut dir = removed.parent();
    while let Some(d) = dir {
        if d == root || !d.starts_with(root) {
            break;
        }
        if std::fs::remove_dir(d).is_err() {
            break; // not empty (or gone) — stop climbing
        }
        dir = d.parent();
    }
}

fn require_folder(cfg: &Config) -> Result<PathBuf> {
    let folder = cfg.folder.clone().context(
        "no sync folder configured — run `akurai-drive-sync init <folder> --bucket <name>`",
    )?;
    let path = PathBuf::from(folder);
    if !path.is_dir() {
        bail!("sync folder {} does not exist", path.display());
    }
    Ok(path)
}

fn print_summary(s: &Summary) {
    println!(
        "\nSync complete: {} uploaded, {} downloaded, {} deleted remote, {} deleted local, {} conflicts{}",
        s.uploaded,
        s.downloaded,
        s.deleted_remote,
        s.deleted_local,
        s.conflicts,
        if s.errors > 0 {
            format!(", {} errors", s.errors)
        } else {
            String::new()
        }
    );
}

/// Minimal RFC3339-ish UTC timestamp without pulling in chrono.
fn now_rfc3339() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let secs = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    // Civil date from Unix seconds (Howard Hinnant's algorithm).
    let days = (secs / 86_400) as i64;
    let rem = secs % 86_400;
    let (h, mi, se) = (rem / 3600, (rem % 3600) / 60, rem % 60);
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!("{y:04}-{m:02}-{d:02}T{h:02}:{mi:02}:{se:02}Z")
}