AkurAI-Drive / client/src/daemon.rs

Latest change feb7ac86b5e60d55b429eacb7146cf0dc32c6628 - Add native Linux sync client (akurai-drive-sync) by olibuijr

//! Long-running daemon: sync once, then react to local file changes (debounced)
//! and poll the remote on a fixed interval. Exits cleanly on SIGINT.

use anyhow::{Context, Result};
use notify::{RecursiveMode, Watcher};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{channel, RecvTimeoutError};
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::config::Config;
use crate::sync::run_sync;

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

pub fn run(cfg: &Config) -> Result<()> {
    let folder = cfg
        .folder
        .clone()
        .map(PathBuf::from)
        .context("no sync folder configured — run `init` first")?;

    // SIGINT / SIGTERM → flip the running flag and let the loop unwind.
    let running = Arc::new(AtomicBool::new(true));
    {
        let r = running.clone();
        ctrlc::set_handler(move || r.store(false, Ordering::SeqCst))
            .context("installing signal handler")?;
    }

    println!("Initial sync...");
    if let Err(e) = run_sync(cfg, false) {
        eprintln!("initial sync failed: {e:#}");
    }

    // Watch the folder recursively; events arrive on a channel.
    let (tx, rx) = channel();
    let mut watcher = notify::recommended_watcher(move |res| {
        let _ = tx.send(res);
    })
    .context("creating filesystem watcher")?;
    watcher
        .watch(&folder, RecursiveMode::Recursive)
        .with_context(|| format!("watching {}", folder.display()))?;

    println!(
        "Watching {} — syncing on change (debounced {}s) and polling every {}s. Ctrl-C to stop.",
        folder.display(),
        DEBOUNCE.as_secs(),
        POLL_INTERVAL.as_secs()
    );

    let mut last_change: Option<Instant> = None;
    let mut last_poll = Instant::now();

    while running.load(Ordering::SeqCst) {
        match rx.recv_timeout(TICK) {
            Ok(event) => {
                if is_relevant(&event) {
                    last_change = Some(Instant::now());
                }
            }
            Err(RecvTimeoutError::Timeout) => {}
            Err(RecvTimeoutError::Disconnected) => break,
        }

        // Debounced local change → sync.
        if let Some(changed_at) = last_change {
            if changed_at.elapsed() >= DEBOUNCE {
                last_change = None;
                sync_now(cfg, "local change");
                last_poll = Instant::now();
            }
        }

        // Periodic remote poll → sync.
        if last_poll.elapsed() >= POLL_INTERVAL {
            last_poll = Instant::now();
            sync_now(cfg, "remote poll");
        }
    }

    println!("\nShutting down cleanly.");
    Ok(())
}

fn sync_now(cfg: &Config, reason: &str) {
    match run_sync(cfg, true) {
        Ok(s) => {
            let touched = s.uploaded
                + s.downloaded
                + s.deleted_remote
                + s.deleted_local
                + s.conflicts
                + s.adopted;
            if touched > 0 || s.errors > 0 {
                println!("[{reason}] {}", s.to_line());
            }
        }
        Err(e) => eprintln!("[{reason}] sync failed: {e:#}"),
    }
}

/// Ignore pure access events and our own temp files.
fn is_relevant(event: &notify::Result<notify::Event>) -> bool {
    let Ok(event) = event 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(crate::util::TMP_PREFIX))
            .unwrap_or(false)
    })
}