//! Native notifications are bounded hints; periodic reconciliation is authoritative.
use notify::{Event, EventKind, RecursiveMode, Watcher};
use serde::Serialize;
use std::{
    collections::{BTreeMap, BTreeSet},
    path::{Path, PathBuf},
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc, Condvar, Mutex,
    },
    thread::{self, JoinHandle},
    time::{Duration, Instant},
};

const MAX_ROOTS: usize = 64;
const MAX_PATHS: usize = 128;

pub struct ScanRequest {
    pub root: PathBuf,
    pub paths: Option<Vec<PathBuf>>,
    pub cancelled: Arc<AtomicBool>,
    pub media_hint: bool,
}
pub enum ScanOutcome {
    Completed,
    Busy,
    Failed(String),
}
#[derive(Clone, Serialize)]
pub struct WatchStatus {
    pub root: PathBuf,
    pub attached: bool,
    pub pending: bool,
    pub error: Option<String>,
    pub watch_error: Option<String>,
}
#[derive(Clone, Copy)]
struct Timing {
    debounce: Duration,
    max_delay: Duration,
    periodic: Duration,
    retry: Duration,
}
impl Default for Timing {
    fn default() -> Self {
        Self {
            debounce: Duration::from_millis(750),
            max_delay: Duration::from_secs(5),
            periodic: Duration::from_secs(300),
            retry: Duration::from_secs(5),
        }
    }
}
struct Root {
    paths: BTreeSet<PathBuf>,
    full: bool,
    media_hint: bool,
    ever_attached: bool,
    first: Option<Instant>,
    last: Instant,
    next_scan: Instant,
    retry_at: Instant,
    attach_retry: Instant,
    failures: u32,
    attached: bool,
    error: Option<String>,
    watch_error: Option<String>,
}
impl Root {
    fn new(now: Instant) -> Self {
        Self {
            paths: BTreeSet::new(),
            full: true,
            media_hint: false,
            ever_attached: false,
            first: Some(now),
            last: now,
            next_scan: now,
            retry_at: now,
            attach_retry: now,
            failures: 0,
            attached: false,
            error: None,
            watch_error: None,
        }
    }
    fn dirty(&mut self, paths: Option<&[PathBuf]>, now: Instant, media_hint: bool) {
        self.media_hint |= media_hint;
        self.first.get_or_insert(now);
        self.last = now;
        match paths {
            Some(paths) if !self.full => {
                self.paths.extend(paths.iter().cloned());
                if self.paths.len() > MAX_PATHS {
                    self.paths.clear();
                    self.full = true;
                }
            }
            None => {
                self.paths.clear();
                self.full = true;
            }
            _ => {}
        }
    }
}
struct State {
    roots: BTreeMap<PathBuf, Root>,
    stop: bool,
}
struct Shared {
    state: Mutex<State>,
    changed: Condvar,
    ignored: Vec<PathBuf>,
    timing: Timing,
    cancelled: Arc<AtomicBool>,
}
/// Own this service for the application lifetime. Dropping joins its worker after
/// the currently executing reconciliation callback finishes.
pub struct WatchService {
    shared: Arc<Shared>,
    worker: Mutex<Option<JoinHandle<()>>>,
}
impl WatchService {
    pub fn start(
        roots: Vec<PathBuf>,
        ignored: Vec<PathBuf>,
        callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static,
    ) -> Result<Self, String> {
        Self::start_timed(roots, ignored, callback, Timing::default())
    }
    fn start_timed(
        roots: Vec<PathBuf>,
        ignored: Vec<PathBuf>,
        callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static,
        timing: Timing,
    ) -> Result<Self, String> {
        validate(&roots)?;
        if ignored.len() > 128 || ignored.iter().any(|p| !p.is_absolute()) {
            return Err("Watcher exclusions must be bounded absolute paths.".into());
        }
        let shared = Arc::new(Shared {
            state: Mutex::new(State {
                roots: BTreeMap::new(),
                stop: false,
            }),
            changed: Condvar::new(),
            ignored,
            timing,
            cancelled: Arc::new(AtomicBool::new(false)),
        });
        let target = shared.clone();
        let worker = thread::Builder::new()
            .name("searchi-reconcile".into())
            .spawn(move || run(target, callback))
            .map_err(|e| e.to_string())?;
        let service = Self {
            shared,
            worker: Mutex::new(Some(worker)),
        };
        service.replace_roots(roots)?;
        Ok(service)
    }
    /// Stop and join only this owned worker. Call on application exit to break
    /// callback ownership of the desktop emitter. Cancellation waits for the
    /// current bounded parser operation to reach its next cooperative check.
    pub fn stop(&self) {
        self.shared.cancelled.store(true, Ordering::Release);
        if let Ok(mut state) = self.shared.state.lock() {
            state.stop = true;
            self.shared.changed.notify_all();
        }
        let worker = self.worker.lock().unwrap_or_else(|e| e.into_inner()).take();
        if let Some(worker) = worker {
            if worker.thread().id() != thread::current().id() {
                let _ = worker.join();
            }
        }
    }

    pub fn add_root(&self, root: PathBuf) -> Result<(), String> {
        let mut s = self
            .shared
            .state
            .lock()
            .map_err(|_| "Watcher is unavailable.")?;
        if s.roots.contains_key(&root) {
            return Ok(());
        }
        if s.roots.len() >= MAX_ROOTS || !root.is_absolute() {
            return Err("Watcher folders must be absolute and limited to 64.".into());
        }
        s.roots.insert(root, Root::new(Instant::now()));
        self.shared.changed.notify_all();
        Ok(())
    }
    pub fn replace_roots(&self, roots: Vec<PathBuf>) -> Result<(), String> {
        validate(&roots)?;
        let mut s = self
            .shared
            .state
            .lock()
            .map_err(|_| "Watcher is unavailable.")?;
        s.roots.retain(|p, _| roots.contains(p));
        for p in roots {
            s.roots
                .entry(p)
                .or_insert_with(|| Root::new(Instant::now()));
        }
        self.shared.changed.notify_all();
        Ok(())
    }
    pub fn status(&self) -> Vec<WatchStatus> {
        self.shared
            .state
            .lock()
            .map(|s| {
                s.roots
                    .iter()
                    .map(|(p, r)| WatchStatus {
                        root: p.clone(),
                        attached: r.attached,
                        pending: r.first.is_some(),
                        error: r.error.clone().or_else(|| r.watch_error.clone()),
                        watch_error: r.watch_error.clone(),
                    })
                    .collect()
            })
            .unwrap_or_default()
    }
}
impl Drop for WatchService {
    fn drop(&mut self) {
        self.stop();
    }
}
fn validate(roots: &[PathBuf]) -> Result<(), String> {
    if roots.len() > MAX_ROOTS || roots.iter().any(|p| !p.is_absolute()) {
        return Err("Watcher folders must be absolute and limited to 64.".into());
    }
    Ok(())
}
fn excluded(path: &Path, shared: &Shared) -> bool {
    shared.ignored.iter().any(|p| path.starts_with(p))
        || path.components().any(|c| {
            matches!(
                c.as_os_str().to_str(),
                Some(".git" | "node_modules" | "target")
            )
        })
}
fn hint(shared: &Shared, event: notify::Result<Event>) {
    let mut s = match shared.state.lock() {
        Ok(s) => s,
        Err(_) => return,
    };
    let now = Instant::now();
    match event {
        Ok(e)
            if !e.need_rescan()
                && matches!(
                    e.kind,
                    EventKind::Access(_)
                        | EventKind::Modify(notify::event::ModifyKind::Metadata(
                            notify::event::MetadataKind::AccessTime
                        ))
                ) => {}
        Ok(e) if !e.need_rescan() && !e.paths.is_empty() => {
            // Events can contain unbounded path vectors; consume a bounded prefix
            // and invalidate all roots if its representation overflowed.
            if e.paths.len() > MAX_PATHS {
                for r in s.roots.values_mut() {
                    r.dirty(None, now, true);
                }
            } else {
                for (root, r) in &mut s.roots {
                    let paths: Vec<_> = e
                        .paths
                        .iter()
                        .filter(|p| p.starts_with(root.as_path()) && !excluded(p, shared))
                        .cloned()
                        .collect();
                    if !paths.is_empty() {
                        r.dirty(Some(&paths), now, true);
                    }
                }
            }
        }
        _ => {
            for r in s.roots.values_mut() {
                r.dirty(None, now, true);
            }
        }
    }
    shared.changed.notify_all();
}
fn run(shared: Arc<Shared>, callback: impl Fn(ScanRequest) -> ScanOutcome) {
    let events = shared.clone();
    let mut watcher = notify::recommended_watcher(move |event| hint(&events, event)).ok();
    let mut watched = BTreeSet::new();
    loop {
        let roots = {
            let s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
            if s.stop {
                break;
            }
            s.roots.keys().cloned().collect::<Vec<_>>()
        };
        for removed in watched
            .iter()
            .filter(|p| !roots.contains(p))
            .cloned()
            .collect::<Vec<_>>()
        {
            if let Some(w) = &mut watcher {
                let _ = w.unwatch(&removed);
            }
            watched.remove(&removed);
        }
        let now = Instant::now();
        for root in &roots {
            if watched.contains(root) && root.is_dir() {
                continue;
            }
            if watched.remove(root) {
                if let Some(r) = shared
                    .state
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .roots
                    .get_mut(root)
                {
                    r.attached = false;
                }
                if let Some(w) = &mut watcher {
                    let _ = w.unwatch(root);
                }
            }
            let due = shared
                .state
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .roots
                .get(root)
                .is_some_and(|r| now >= r.attach_retry);
            if !due {
                continue;
            }
            if watcher.is_none() {
                let events = shared.clone();
                watcher = notify::recommended_watcher(move |e| hint(&events, e)).ok();
            }
            let result = match &mut watcher {
                Some(w) => w
                    .watch(root, RecursiveMode::Recursive)
                    .map_err(|e| e.to_string()),
                None => Err(
                    "Native watcher is unavailable; periodic reconciliation remains enabled."
                        .into(),
                ),
            };
            let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(r) = s.roots.get_mut(root) {
                r.attached = result.is_ok();
                r.attach_retry = if result.is_ok() {
                    now
                } else {
                    now + shared.timing.retry
                };
                if result.is_ok() {
                    r.watch_error = None;
                    r.dirty(None, now, r.ever_attached);
                    r.ever_attached = true;
                    watched.insert(root.clone());
                } else {
                    r.watch_error = result.err().map(|e| e.chars().take(512).collect());
                }
            }
        }
        let request = {
            let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
            if s.stop {
                break;
            }
            let now = Instant::now();
            for r in s.roots.values_mut() {
                if now >= r.next_scan {
                    r.dirty(None, now, false);
                    r.next_scan = now + shared.timing.periodic;
                }
            }
            s.roots.iter_mut().find_map(|(root, r)| {
                let due = r.first.is_some_and(|first| {
                    now >= r.retry_at
                        && (now.duration_since(r.last) >= shared.timing.debounce
                            || now.duration_since(first) >= shared.timing.max_delay)
                });
                if !due {
                    return None;
                }
                let paths = if r.full {
                    None
                } else {
                    Some(std::mem::take(&mut r.paths).into_iter().collect())
                };
                r.full = false;
                let media_hint = std::mem::take(&mut r.media_hint);
                r.first = None;
                Some(ScanRequest {
                    root: root.clone(),
                    paths,
                    cancelled: shared.cancelled.clone(),
                    media_hint,
                })
            })
        };
        if let Some(request) = request {
            let root = request.root.clone();
            let paths = request.paths.clone();
            let media_hint = request.media_hint;
            let result =
                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| callback(request)))
                    .unwrap_or_else(|_| {
                        ScanOutcome::Failed(
                            "Folder reconciliation failed unexpectedly; retrying.".into(),
                        )
                    });
            let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(r) = s.roots.get_mut(&root) {
                let now = Instant::now();
                match result {
                    ScanOutcome::Completed => {
                        r.failures = 0;
                        r.error = None;
                        r.retry_at = now;
                    }
                    ScanOutcome::Busy => {
                        r.dirty(paths.as_deref(), now, media_hint);
                        r.retry_at = now + Duration::from_millis(250);
                    }
                    ScanOutcome::Failed(e) => {
                        r.dirty(paths.as_deref(), now, media_hint);
                        r.failures = r.failures.saturating_add(1);
                        r.error = Some(e.chars().take(512).collect());
                        r.retry_at = now
                            + shared
                                .timing
                                .retry
                                .saturating_mul(1u32 << r.failures.saturating_sub(1).min(4))
                                .min(Duration::from_secs(60));
                    }
                }
            }
        } else {
            let s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
            if s.stop {
                break;
            }
            let _ = shared.changed.wait_timeout(s, Duration::from_millis(100));
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::mpsc;
    fn fast() -> Timing {
        Timing {
            debounce: Duration::from_millis(40),
            max_delay: Duration::from_millis(120),
            periodic: Duration::from_secs(30),
            retry: Duration::from_millis(20),
        }
    }
    fn receive(rx: &mpsc::Receiver<ScanRequest>) -> ScanRequest {
        rx.recv_timeout(Duration::from_secs(10))
            .expect("watch reconciliation")
    }
    #[test]
    fn native_windows_or_linux_create_update_rename_delete() {
        let root = tempfile::tempdir().unwrap();
        let (tx, rx) = mpsc::channel();
        let service = WatchService::start_timed(
            vec![root.path().to_owned()],
            vec![],
            move |r| {
                tx.send(r).unwrap();
                ScanOutcome::Completed
            },
            fast(),
        )
        .unwrap();
        assert!(receive(&rx).paths.is_none());
        let first = root.path().join("first.txt");
        let second = root.path().join("second.txt");
        std::fs::write(&first, "one").unwrap();
        assert!(receive(&rx).paths.unwrap().contains(&first));
        std::fs::write(&first, "two").unwrap();
        assert!(receive(&rx).paths.unwrap().contains(&first));
        std::fs::rename(&first, &second).unwrap();
        let rename = receive(&rx).paths.unwrap();
        assert!(rename.contains(&first) || rename.contains(&second));
        std::fs::remove_file(&second).unwrap();
        assert!(receive(&rx).paths.unwrap().contains(&second));
        assert!(service.status()[0].attached);
    }
    #[test]
    fn overflow_rescan_and_events_during_scan_are_retained() {
        let root = tempfile::tempdir().unwrap();
        let (tx, rx) = mpsc::channel();
        let service = WatchService::start_timed(
            vec![root.path().to_owned()],
            vec![],
            move |r| {
                tx.send(r).unwrap();
                ScanOutcome::Completed
            },
            fast(),
        )
        .unwrap();
        let _ = receive(&rx);
        let paths = (0..129)
            .map(|i| root.path().join(format!("{i}.txt")))
            .collect();
        hint(
            &service.shared,
            Ok(Event::new(EventKind::Any).add_path(root.path().join("a"))),
        );
        let mut e = Event::new(EventKind::Any);
        e.paths = paths;
        hint(&service.shared, Ok(e));
        let overflow = receive(&rx);
        assert!(overflow.paths.is_none());
        assert!(overflow.media_hint);
    }
    #[test]
    fn events_arriving_during_reconciliation_survive_completion() {
        let root = tempfile::tempdir().unwrap();
        let (tx, rx) = mpsc::channel();
        let (release_tx, release_rx) = mpsc::channel();
        let release_rx = Mutex::new(release_rx);
        let service = WatchService::start_timed(
            vec![root.path().to_owned()],
            vec![],
            move |r| {
                tx.send(r).unwrap();
                release_rx.lock().unwrap().recv().unwrap();
                ScanOutcome::Completed
            },
            fast(),
        )
        .unwrap();
        let _ = receive(&rx);
        let path = root.path().join("later.txt");
        hint(
            &service.shared,
            Ok(Event::new(EventKind::Any).add_path(path.clone())),
        );
        release_tx.send(()).unwrap();
        assert!(receive(&rx).paths.unwrap().contains(&path));
        release_tx.send(()).unwrap();
        drop(service);
    }

    #[test]
    fn callback_panic_retries_and_recovers() {
        let root = tempfile::tempdir().unwrap();
        let (tx, rx) = mpsc::channel();
        let attempts = AtomicBool::new(false);
        let service = WatchService::start_timed(
            vec![root.path().to_owned()],
            vec![],
            move |r| {
                if !attempts.swap(true, Ordering::AcqRel) {
                    panic!("owned callback failure");
                }
                tx.send(r).unwrap();
                ScanOutcome::Completed
            },
            fast(),
        )
        .unwrap();
        assert!(receive(&rx).paths.is_none());
        service.stop();
        assert!(!service.status()[0].pending);
    }

    #[test]
    fn busy_retry_preserves_native_media_hint() {
        let root = tempfile::tempdir().unwrap();
        let (tx, rx) = mpsc::channel();
        let service = WatchService::start_timed(
            vec![root.path().to_owned()],
            vec![],
            move |r| {
                if r.media_hint {
                    tx.send(r).unwrap();
                    ScanOutcome::Busy
                } else {
                    tx.send(r).unwrap();
                    ScanOutcome::Completed
                }
            },
            fast(),
        )
        .unwrap();
        assert!(!receive(&rx).media_hint);
        let mut event = Event::new(EventKind::Any);
        event.attrs.set_flag(notify::event::Flag::Rescan);
        hint(&service.shared, Ok(event));
        let first = receive(&rx);
        assert!(first.paths.is_none());
        assert!(first.media_hint);
        assert!(receive(&rx).media_hint);
        service.stop();
    }

    #[test]
    fn periodic_reconciliation_and_shutdown_signal() {
        let root = tempfile::tempdir().unwrap();
        let (tx, rx) = mpsc::channel();
        let mut timing = fast();
        timing.periodic = Duration::from_millis(150);
        let service = WatchService::start_timed(
            vec![root.path().to_owned()],
            vec![],
            move |r| {
                tx.send(r).unwrap();
                ScanOutcome::Completed
            },
            timing,
        )
        .unwrap();
        let initial = receive(&rx);
        assert!(initial.paths.is_none());
        assert!(!initial.media_hint);
        let periodic = receive(&rx);
        assert!(periodic.paths.is_none());
        assert!(!periodic.media_hint);
        drop(service);
        assert!(initial.cancelled.load(Ordering::Acquire));
    }
}