//! Filesystem watcher, polling, and debounce control plane.

use std::panic::AssertUnwindSafe;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

use pdfs_core::control::SyncEngineState;
use tracing::error;

use super::*;

/// How long the watcher must go quiet before a burst counts as finished.
///
/// Measured from the *last* event, not the first: applications do not write a
/// file once. An editor saves by writing a temp file and renaming it over the
/// target; an export or a database dump writes continuously for as long as it
/// takes. Settling a fixed interval after the first event walks the tree while
/// the file is still growing, and uploads a torn snapshot as a real revision —
/// corrected on the next pass, but only after spending the encrypt-and-upload
/// on a file that was never in that state.
const DEBOUNCE: Duration = Duration::from_secs(2);
/// Ceiling on the settle wait, so a directory under sustained change still
/// syncs. Without it, copying a large tree in — events arriving forever —
/// would postpone the reconcile for as long as the copy ran.
const MAX_SETTLE: Duration = Duration::from_secs(30);
/// How often to re-walk remotes so changes made on other clients are pulled in.
const POLL_INTERVAL: Duration = Duration::from_secs(120);
/// How many uploads/downloads/folder-creations a reconcile runs at once. The
/// per-item work is a network round-trip, so a folder with thousands of files is
/// unusable done one-at-a-time; this bounds concurrency like the bulk-upload
/// engine does.
pub(crate) const SYNC_CONCURRENCY: usize = 8;

/// A message to the sync engine's control thread.
pub(crate) enum SyncMsg {
    /// Reconcile one folder by id (a local change, or a targeted `SyncNow`).
    Reconcile(i64),
    /// Reconcile every mirror folder (periodic poll, or startup).
    ReconcileAll,
    /// Re-read the folder set and adjust the filesystem watches (after add/remove).
    Rewatch,
    /// The daemon is tearing down: finish the current pass and end the engine
    /// thread, so the mount can join it (`docs/BUGS.md` B44).
    Stop,
}

/// Start the sync engine: a control thread that owns the filesystem watcher and
/// serialises reconcile passes, plus a periodic poll thread. The engine receives
/// on `rx`; senders live in [`Core::sync_tx`].
pub(crate) fn spawn(core: Core, rx: Receiver<SyncMsg>) -> Option<std::thread::JoinHandle<()>> {
    match std::thread::Builder::new()
        .name("pdfs-sync".into())
        .spawn(move || engine_loop(core, rx))
    {
        Ok(handle) => Some(handle),
        Err(e) => {
            warn!(error = %e, "failed to start sync engine");
            None
        }
    }
}

/// Whether the engine is alive and which pass it is on, for
/// [`Core::diagnostics`] and the supervisor.
///
/// A panic in a pass used to end the engine thread. Every later add, sync-now,
/// file event and poll then went into a channel nobody read, the folder stayed
/// `syncing`, and nothing said so until a restart (`docs/BUGS.md` B259).
#[derive(Default)]
pub(crate) struct EngineHealth {
    running: AtomicBool,
    pass: Mutex<Option<PassClock>>,
    panics: AtomicU64,
}

#[derive(Clone, Copy)]
struct PassClock {
    folder: i64,
    started: Instant,
    moved: Instant,
}

impl EngineHealth {
    /// Note that the running pass reported progress.
    pub(crate) fn moved(&self) {
        if let Some(pass) = self.pass.lock().as_mut() {
            pass.moved = Instant::now();
        }
    }

    /// The engine's state now. `try_lock`, like the rest of the diagnostics:
    /// a report must not wait on the thing it reports about.
    pub(crate) fn report(&self) -> SyncEngineState {
        let pass = self.pass.try_lock().and_then(|pass| *pass);
        SyncEngineState {
            running: self.running.load(Ordering::Relaxed),
            pass_folder: pass.map(|p| p.folder),
            pass_age_secs: pass.map_or(0, |p| p.started.elapsed().as_secs()),
            pass_idle_secs: pass.map_or(0, |p| p.moved.elapsed().as_secs()),
            panics: self.panics.load(Ordering::Relaxed),
        }
    }

    /// Run `pass` over `folder` as the current pass. Returns `false` when it
    /// panicked: the panic ends that pass, not the engine.
    pub(super) fn contain(&self, folder: i64, pass: impl FnOnce()) -> bool {
        let now = Instant::now();
        *self.pass.lock() = Some(PassClock {
            folder,
            started: now,
            moved: now,
        });
        let finished = std::panic::catch_unwind(AssertUnwindSafe(pass)).is_ok();
        *self.pass.lock() = None;
        if !finished {
            self.panics.fetch_add(1, Ordering::Relaxed);
        }
        finished
    }
}

/// Marks the engine running for as long as it lives.
pub(super) struct Running<'a>(&'a EngineHealth);

impl<'a> Running<'a> {
    pub(super) fn start(health: &'a EngineHealth) -> Self {
        health.running.store(true, Ordering::Relaxed);
        Self(health)
    }
}

impl Drop for Running<'_> {
    fn drop(&mut self) {
        self.0.running.store(false, Ordering::Relaxed);
        if std::thread::panicking() {
            error!("sync engine stopped by a panic; nothing syncs until the daemon restarts");
        }
    }
}

/// One folder's pass, contained: a panic in it sets the folder to `error` and
/// leaves the engine running for the next one.
fn run_pass(core: &Core, folder: &StoredSyncFolder) {
    if core
        .sync_engine
        .contain(folder.id, || core.reconcile_folder(folder))
    {
        return;
    }
    core.progress_end(folder.id);
    let _ = core
        .db
        .sync_folder_set_state(folder.id, "error", folder.last_sync);
    error!(
        id = folder.id,
        path = %folder.local_path,
        "sync: pass panicked; the folder is set to error and tried again on the next poll"
    );
    core.log_activity(
        ActivityKind::Sync,
        format!("sync failed for {}", base_name(&folder.local_path)),
        "an internal error stopped the pass; it is tried again on the next poll",
        false,
    );
}

/// The pass a `Reconcile(id)` asked for. A folder removed since, or one the
/// database cannot read, is said so rather than dropped in silence.
fn run_pass_by_id(core: &Core, id: i64) {
    match core.db.sync_folder_get(id) {
        Ok(Some(folder)) => run_pass(core, &folder),
        Ok(None) => debug!(id, "sync: folder removed before its pass; skipping"),
        Err(e) => warn!(id, error = ?e, "sync: cannot read folder; skipping its pass"),
    }
}

/// The engine control loop. Single-threaded so reconcile passes never overlap.
fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
    let health = core.sync_engine.clone();
    let _running = Running::start(&health);
    // Paths the watcher currently covers, mapped to their folder id, so an event
    // path can be resolved back to the folder it belongs to.
    let watched: Arc<Mutex<Vec<(PathBuf, i64)>>> = Arc::new(Mutex::new(Vec::new()));

    let tx_events = core.sync_tx.clone();
    let watched_events = watched.clone();
    let mut watcher =
        match notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
            let Ok(event) = res else { return };
            // Ignore pure metadata/access noise; content changes are what matter.
            if matches!(event.kind, notify::EventKind::Access(_)) {
                return;
            }
            let msg = event_target(&watched_events.lock(), &event.paths);
            let _ = tx_events.send(msg);
        }) {
            Ok(w) => w,
            Err(e) => {
                warn!(error = %e, "filesystem watcher unavailable; sync is poll-only");
                // Still run the loop so polling and explicit SyncNow work.
                poll_only_loop(&core, rx);
                return;
            }
        };

    // Periodic remote poll.
    {
        let tx = core.sync_tx.clone();
        let shutdown = core.shutdown.clone();
        std::thread::Builder::new()
            .name("pdfs-sync-poll".into())
            .spawn(move || {
                while shutdown.sleep(POLL_INTERVAL) {
                    if tx.send(SyncMsg::ReconcileAll).is_err() {
                        break;
                    }
                }
            })
            .ok();
    }

    rewatch(&core, &mut watcher, &watched);
    reconcile_all(&core);

    while let Ok(msg) = rx.recv() {
        let mut ids: HashSet<i64> = HashSet::new();
        let mut all = false;
        let mut do_rewatch = false;
        if !classify(msg, &mut ids, &mut all, &mut do_rewatch)
            || !settle(&rx, &mut ids, &mut all, &mut do_rewatch)
        {
            info!("sync engine stopping");
            return;
        }

        if do_rewatch {
            rewatch(&core, &mut watcher, &watched);
        }
        if all {
            reconcile_all(&core);
        } else {
            for id in ids {
                if core.sync_paused() {
                    break;
                }
                run_pass_by_id(&core, id);
            }
        }
    }
}

/// Fallback loop when no filesystem watcher could be created: poll + SyncNow only.
fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
    {
        let tx = core.sync_tx.clone();
        let shutdown = core.shutdown.clone();
        std::thread::spawn(move || {
            while shutdown.sleep(POLL_INTERVAL) {
                if tx.send(SyncMsg::ReconcileAll).is_err() {
                    break;
                }
            }
        });
    }
    reconcile_all(core);
    while let Ok(msg) = rx.recv() {
        let mut ids: HashSet<i64> = HashSet::new();
        let mut all = false;
        let mut do_rewatch = false;
        if !classify(msg, &mut ids, &mut all, &mut do_rewatch)
            || !settle(&rx, &mut ids, &mut all, &mut do_rewatch)
        {
            info!("sync engine stopping");
            return;
        }
        if all {
            reconcile_all(core);
        } else {
            for id in ids {
                if core.sync_paused() {
                    break;
                }
                run_pass_by_id(core, id);
            }
        }
    }
}

/// Absorb the rest of an event burst, returning once the watcher has been quiet
/// for `quiet` or `cap` has elapsed since the first event — whichever comes
/// first. The caller has already classified the event that opened the burst.
///
/// Split out with its timings as parameters so the settling behaviour can be
/// tested in milliseconds rather than in the tens of seconds the real constants
/// describe.
pub(super) fn settle_with(
    rx: &Receiver<SyncMsg>,
    ids: &mut HashSet<i64>,
    all: &mut bool,
    do_rewatch: &mut bool,
    quiet: Duration,
    cap: Duration,
) -> bool {
    let deadline = Instant::now() + cap;
    loop {
        let now = Instant::now();
        if now >= deadline {
            break;
        }
        // Never wait past the cap, however quiet the watcher goes.
        let wait = quiet.min(deadline - now);
        match rx.recv_timeout(wait) {
            // Another event: the burst is still going, so the quiet window
            // starts again from here.
            // A stop lands mid-burst: give up the rest of the burst rather
            // than settling for it, and let the caller end the loop.
            Ok(m) => {
                if !classify(m, ids, all, do_rewatch) {
                    return false;
                }
            }
            // Quiet for a full window: whatever was being written has stopped.
            Err(RecvTimeoutError::Timeout) => break,
            // Every sender is gone; the caller's loop is about to end anyway.
            Err(RecvTimeoutError::Disconnected) => break,
        }
    }
    true
}

/// [`settle_with`] at the real timings.
fn settle(
    rx: &Receiver<SyncMsg>,
    ids: &mut HashSet<i64>,
    all: &mut bool,
    do_rewatch: &mut bool,
) -> bool {
    settle_with(rx, ids, all, do_rewatch, DEBOUNCE, MAX_SETTLE)
}

/// Fold one message into the pending work of the current burst. Returns `false`
/// for [`SyncMsg::Stop`], which is not work but an instruction to end the loop.
pub(super) fn classify(
    msg: SyncMsg,
    ids: &mut HashSet<i64>,
    all: &mut bool,
    rewatch: &mut bool,
) -> bool {
    match msg {
        SyncMsg::Reconcile(id) => {
            ids.insert(id);
        }
        SyncMsg::ReconcileAll => *all = true,
        SyncMsg::Rewatch => *rewatch = true,
        SyncMsg::Stop => return false,
    }
    true
}

/// Reconcile every mirror folder in turn. Skipped while sync is paused: the
/// resume sends a fresh [`SyncMsg::ReconcileAll`], which walks everything the
/// watcher reported meanwhile.
fn reconcile_all(core: &Core) {
    if core.sync_paused() {
        debug!("sync: paused, skipping reconcile");
        return;
    }
    let folders = match core.db.sync_folder_list() {
        Ok(f) => f,
        Err(e) => {
            warn!(error = ?e, "sync: cannot list folders");
            return;
        }
    };
    for folder in folders {
        run_pass(core, &folder);
    }
}

/// The reconcile a watcher event asks for: the one mirror folder all its paths
/// lie in, so an edit in one folder does not walk every other one too. An event
/// that spans folders, or that no watched folder claims (a watch list a moment
/// out of date), falls back to every folder.
pub(super) fn event_target(watched: &[(PathBuf, i64)], paths: &[PathBuf]) -> SyncMsg {
    let mut ids = paths.iter().map(|path| {
        watched
            .iter()
            .filter(|(root, _)| path.starts_with(root))
            // The deepest root, should one mirror folder sit inside another.
            .max_by_key(|(root, _)| root.components().count())
            .map(|(_, id)| *id)
    });
    match ids.next() {
        Some(Some(first)) if ids.all(|id| id == Some(first)) => SyncMsg::Reconcile(first),
        _ => SyncMsg::ReconcileAll,
    }
}

/// Bring the filesystem watches in line with the current mirror-folder set:
/// watch newly-added folders, drop removed ones.
///
/// The watch calls run on a copy, outside the lock: the watcher's event
/// callback takes the same lock, and a backend that hands a watch request to
/// its event thread and waits for the answer would deadlock against it.
fn rewatch(
    core: &Core,
    watcher: &mut notify::RecommendedWatcher,
    watched: &Mutex<Vec<(PathBuf, i64)>>,
) {
    let want: Vec<(PathBuf, i64)> = match core.db.sync_folder_list() {
        Ok(f) => f
            .into_iter()
            .filter(|f| f.mode == "mirror")
            .map(|f| (PathBuf::from(f.local_path), f.id))
            .collect(),
        Err(e) => {
            warn!(error = ?e, "sync: cannot list folders to watch");
            return;
        }
    };
    let mut have = watched.lock().clone();
    // Drop watches no longer wanted.
    have.retain(|(path, _)| {
        if want.iter().any(|(p, _)| p == path) {
            true
        } else {
            let _ = watcher.unwatch(path);
            false
        }
    });
    // Add newly wanted watches.
    for (path, id) in &want {
        if have.iter().any(|(p, _)| p == path) {
            continue;
        }
        match watcher.watch(path, RecursiveMode::Recursive) {
            Ok(()) => have.push((path.clone(), *id)),
            Err(e) => warn!(path = %path.display(), error = %e, "sync: watch failed"),
        }
    }
    *watched.lock() = have;
}

// ---- reconcile ------------------------------------------------------------