AkurAI-Drive / desktop/src-tauri/src/sync.rs

Latest change 76b668ab797c5342c1cccf4955c8f1943cba1bb7 - OLI-13: pin toolchain 1.98.1 at repo root; desktop: fmt, clippy, plist/quick-xml advisory fix, first unit tests by AkurAI Agent

//! The background sync engine: one long-lived async task that runs a debounced,
//! polled, bidirectional sync (exactly like the CLI daemon, but driving the
//! library's `run_sync` directly), pushes `sync-status` events to the webview,
//! and keeps the tray tooltip/status item in step.

use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use tauri::{AppHandle, Emitter};
use tokio::sync::{Mutex as AsyncMutex, Notify};

use akurai_drive_sync::config::Config;
use akurai_drive_sync::sync::run_sync;

/// How often we poll the remote for changes made elsewhere.
const POLL_INTERVAL: Duration = Duration::from_secs(30);
/// Quiet period after the last filesystem event before we sync.
const DEBOUNCE: Duration = Duration::from_secs(2);

/// Shared, app-managed sync state. Commands and the tray talk to the loop
/// through this.
pub struct SyncManager {
    /// Wake the loop for an immediate reconcile.
    pub trigger: Notify,
    /// Serializes sync passes so a manual `sync_now` and the loop never overlap.
    sync_lock: AsyncMutex<()>,
    syncing: AtomicBool,
    last_summary: Mutex<Option<String>>,
    last_error: Mutex<Option<String>>,
    email: Mutex<Option<String>>,
}

impl SyncManager {
    pub fn new() -> Self {
        Self {
            trigger: Notify::new(),
            sync_lock: AsyncMutex::new(()),
            syncing: AtomicBool::new(false),
            last_summary: Mutex::new(None),
            last_error: Mutex::new(None),
            email: Mutex::new(None),
        }
    }

    pub fn is_syncing(&self) -> bool {
        self.syncing.load(Ordering::SeqCst)
    }
    pub fn last_summary(&self) -> Option<String> {
        self.last_summary.lock().ok().and_then(|g| g.clone())
    }
    pub fn last_error(&self) -> Option<String> {
        self.last_error.lock().ok().and_then(|g| g.clone())
    }
    pub fn email(&self) -> Option<String> {
        self.email.lock().ok().and_then(|g| g.clone())
    }
    pub fn set_email(&self, e: Option<String>) {
        if let Ok(mut g) = self.email.lock() {
            *g = e;
        }
    }
    /// Wake the background loop for an immediate reconcile.
    pub fn trigger_sync(&self) {
        self.trigger.notify_one();
    }
}

/// Emit the current status to the webview and reflect it on the tray.
fn emit_status(app: &AppHandle, manager: &SyncManager, state: &str, detail: Option<String>) {
    let payload = serde_json::json!({
        "state": state,
        "detail": detail,
        "summary": manager.last_summary(),
        "error": manager.last_error(),
    });
    if let Err(e) = app.emit("sync-status", &payload) {
        tracing::error!(error = %e, "failed to emit sync-status event");
    }
    crate::tray::update_status(app, state);
}

/// Run exactly one bidirectional sync pass (if the config is complete).
/// Returns the summary line on success.
pub async fn run_one_pass(
    app: AppHandle,
    manager: Arc<SyncManager>,
    reason: &str,
) -> Option<String> {
    let _guard = manager.sync_lock.lock().await;

    let cfg = match Config::load() {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %format!("{e:#}"), "sync: config load failed");
            return None;
        }
    };
    if !(cfg.is_logged_in() && cfg.folder.is_some() && cfg.bucket.is_some()) {
        tracing::debug!(reason, "sync: skipped (not fully configured)");
        return None;
    }

    tracing::info!(reason, "sync pass starting");
    manager.syncing.store(true, Ordering::SeqCst);
    emit_status(&app, &manager, "syncing", Some(reason.to_string()));

    let cfg_for_pass = cfg.clone();
    let res = tokio::task::spawn_blocking(move || run_sync(&cfg_for_pass, true)).await;
    manager.syncing.store(false, Ordering::SeqCst);

    match res {
        Ok(Ok(summary)) => {
            let line = summary.to_line();
            if let Ok(mut g) = manager.last_summary.lock() {
                *g = Some(line.clone());
            }
            if let Ok(mut g) = manager.last_error.lock() {
                *g = None;
            }
            tracing::info!(summary = %line, "sync pass ok");
            emit_status(&app, &manager, "idle", None);
            Some(line)
        }
        Ok(Err(e)) => {
            let msg = format!("{e:#}");
            if let Ok(mut g) = manager.last_error.lock() {
                *g = Some(msg.clone());
            }
            tracing::error!(error = %msg, "sync pass failed");
            emit_status(&app, &manager, "error", Some(msg));
            None
        }
        Err(e) => {
            let msg = format!("sync task panicked: {e}");
            if let Ok(mut g) = manager.last_error.lock() {
                *g = Some(msg.clone());
            }
            tracing::error!(error = %msg, "sync task panicked");
            emit_status(&app, &manager, "error", Some(msg));
            None
        }
    }
}

/// Spawn the long-lived background sync loop. Re-reads config every wake-up so
/// it picks up a freshly-set folder/bucket/token without a restart, and
/// (re)watches the sync folder whenever it changes.
#[allow(unused_assignments)]
pub fn spawn_sync_loop(app: AppHandle, manager: Arc<SyncManager>) {
    tauri::async_runtime::spawn(async move {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<()>();
        // Held only to keep the OS watch alive across iterations (never read).
        let mut _active_watcher: Option<notify::RecommendedWatcher> = None;
        let mut watched: Option<PathBuf> = None;

        // First pass right away.
        run_one_pass(app.clone(), manager.clone(), "startup").await;

        loop {
            // (Re)configure the filesystem watcher when the folder changes.
            let cfg = Config::load().unwrap_or_default();
            let folder = cfg.folder.clone().map(PathBuf::from);
            if folder != watched {
                _active_watcher = None; // drop the old watcher first
                watched = folder.clone();
                if let Some(f) = folder.as_ref() {
                    if f.is_dir() {
                        let txc = tx.clone();
                        match notify::recommended_watcher(
                            move |res: notify::Result<notify::Event>| {
                                if relevant(&res) {
                                    let _ = txc.send(());
                                }
                            },
                        ) {
                            Ok(mut w) => {
                                use notify::Watcher;
                                match w.watch(f, notify::RecursiveMode::Recursive) {
                                    Ok(()) => {
                                        tracing::info!(folder = %f.display(), "watching sync folder");
                                        _active_watcher = Some(w);
                                    }
                                    Err(e) => {
                                        tracing::error!(error = %e, folder = %f.display(), "failed to watch folder")
                                    }
                                }
                            }
                            Err(e) => {
                                tracing::error!(error = %e, "failed to create filesystem watcher")
                            }
                        }
                    }
                }
            }

            // Wait for the next reason to sync.
            let reason = tokio::select! {
                _ = manager.trigger.notified() => "trigger",
                _ = tokio::time::sleep(POLL_INTERVAL) => "poll",
                ev = rx.recv() => {
                    if ev.is_none() {
                        tracing::warn!("watch channel closed; sync loop ending");
                        break;
                    }
                    // Debounce: let the dust settle, then drain queued events.
                    tokio::time::sleep(DEBOUNCE).await;
                    while rx.try_recv().is_ok() {}
                    "local change"
                }
            };
            run_one_pass(app.clone(), manager.clone(), reason).await;
        }
    });
}

/// Ignore pure access events and our own atomic-download temp files.
fn relevant(res: &notify::Result<notify::Event>) -> bool {
    let Ok(event) = res else { return true };
    if matches!(event.kind, notify::EventKind::Access(_)) {
        return false;
    }
    !event.paths.iter().all(|p| {
        p.file_name()
            .map(|n| {
                n.to_string_lossy()
                    .starts_with(akurai_drive_sync::util::TMP_PREFIX)
            })
            .unwrap_or(false)
    })
}