Menu
Public repository / master
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: ¬ify::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)
})
}