use notify::{recommended_watcher, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::sync::RwLock;
use tokio::time::{Duration, Instant};

use crate::embed::EmbedConfig;
use crate::index::format::{index_dir, load_meta, save_index, save_meta, IndexMeta};
use crate::index::SymbolIndex;
use crate::indexer::is_supported_extension;
use crate::indexer::{registry, Indexer};
use crate::tools::index_project::current_source_snapshot;

const DEBOUNCE_WINDOW: Duration = Duration::from_millis(500);
const CHANNEL_CAPACITY: usize = 1024;

pub struct ProjectWatcher {
    _watcher: RecommendedWatcher,
}

impl ProjectWatcher {
    pub fn start(
        project_path: PathBuf,
        index: Arc<RwLock<SymbolIndex>>,
        index_path: PathBuf,
        meta_path: PathBuf,
        embed_config: Option<Arc<EmbedConfig>>,
    ) -> anyhow::Result<Self> {
        let project_path_clone = project_path.clone();

        let parsers = registry::build_default_registry();
        let indexer = Arc::new(Indexer::new(parsers));

        let (tx, rx) = mpsc::channel::<PathBuf>(CHANNEL_CAPACITY);
        let overflowed = Arc::new(AtomicBool::new(false));

        tokio::spawn(run_debounce_loop(
            rx,
            project_path_clone,
            indexer,
            index,
            DEBOUNCE_WINDOW,
            index_path,
            meta_path,
            embed_config,
            Arc::clone(&overflowed),
        ));

        let handler = move |result: notify::Result<Event>| {
            if let Ok(event) = result {
                match event.kind {
                    EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => {
                        for path in event.paths {
                            let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
                            if is_supported_extension(ext) {
                                // Non-blocking send; drop the event if the channel is full rather
                                // than blocking the notify callback thread. If we do drop one,
                                // mark the project dirty so the debounce loop does a safe full
                                // resync instead of silently diverging.
                                if tx.try_send(path).is_err() {
                                    overflowed.store(true, Ordering::Release);
                                }
                            }
                        }
                    }
                    _ => {}
                }
            }
        };

        let mut watcher = recommended_watcher(handler)?;
        watcher.watch(&project_path, RecursiveMode::Recursive)?;

        Ok(Self { _watcher: watcher })
    }
}

/// Collects file paths from `rx` into a `HashSet` (deduplication) and flushes
/// the batch through a single write-lock acquisition once the debounce window
/// expires without a new event, or when the channel is closed.
#[allow(clippy::too_many_arguments)]
async fn run_debounce_loop(
    mut rx: mpsc::Receiver<PathBuf>,
    root: PathBuf,
    indexer: Arc<Indexer>,
    index: Arc<RwLock<SymbolIndex>>,
    debounce_window: Duration,
    index_path: PathBuf,
    meta_path: PathBuf,
    embed_config: Option<Arc<EmbedConfig>>,
    overflowed: Arc<AtomicBool>,
) {
    let mut pending: HashSet<PathBuf> = HashSet::new();
    let mut deadline: Option<Instant> = None;

    loop {
        let timeout = deadline
            .map(|d| d.saturating_duration_since(Instant::now()))
            .unwrap_or(Duration::MAX);

        tokio::select! {
            maybe_path = rx.recv() => {
                match maybe_path {
                    Some(path) => {
                        pending.insert(path);
                        deadline = Some(Instant::now() + debounce_window);
                    }
                    // Channel closed — flush whatever is pending and exit.
                    None => {
                        flush_pending(
                            &mut pending,
                            &root,
                            &indexer,
                            &index,
                            &index_path,
                            &meta_path,
                            embed_config.as_ref().map(Arc::clone),
                            &overflowed,
                        )
                        .await;
                        return;
                    }
                }
            }

            // Debounce window expired — flush the batch.
            _ = tokio::time::sleep(timeout), if deadline.is_some() => {
                flush_pending(
                    &mut pending,
                    &root,
                    &indexer,
                    &index,
                    &index_path,
                    &meta_path,
                    embed_config.as_ref().map(Arc::clone),
                    &overflowed,
                )
                .await;
                deadline = None;
            }
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn flush_pending(
    pending: &mut HashSet<PathBuf>,
    root: &Path,
    indexer: &Arc<Indexer>,
    index: &Arc<RwLock<SymbolIndex>>,
    index_path: &Path,
    meta_path: &Path,
    embed_config: Option<Arc<EmbedConfig>>,
    overflowed: &Arc<AtomicBool>,
) {
    if overflowed.swap(false, Ordering::AcqRel) {
        full_resync(root, indexer, index, index_path, meta_path, embed_config).await;
        pending.clear();
        return;
    }

    if pending.is_empty() {
        return;
    }

    reindex_batch(
        pending,
        root,
        indexer,
        index,
        index_path,
        meta_path,
        embed_config,
    )
    .await;
    pending.clear();
}

async fn full_resync(
    root: &Path,
    indexer: &Arc<Indexer>,
    index: &Arc<RwLock<SymbolIndex>>,
    index_path: &Path,
    meta_path: &Path,
    embed_config: Option<Arc<EmbedConfig>>,
) {
    let old_removed_ids = {
        let idx = index.read().await;
        idx.symbols.keys().cloned().collect::<Vec<_>>()
    };

    let root_buf = root.to_path_buf();
    let indexer = Arc::clone(indexer);
    let rebuilt = tokio::task::spawn_blocking(move || indexer.index_project(&root_buf, &[])).await;

    let (rebuilt_index, source_file_count) = match rebuilt {
        Ok(Ok(result)) => result,
        Ok(Err(err)) => {
            eprintln!("pitlane-mcp: full watcher resync failed: {}", err);
            return;
        }
        Err(err) => {
            eprintln!("pitlane-mcp: full watcher resync task panicked: {}", err);
            return;
        }
    };

    let changed_files: HashSet<PathBuf> = rebuilt_index.by_file.keys().cloned().collect();

    {
        let mut idx = index.write().await;
        *idx = rebuilt_index;

        if let Err(e) = save_index(&idx, index_path) {
            eprintln!("pitlane-mcp: failed to flush index to disk: {}", e);
        }

        crate::cache::invalidate(root);

        if let Ok(tantivy_dir) = index_dir(root).map(|d| d.join("tantivy")) {
            let _ = crate::index::bm25::mark_stale(&tantivy_dir);
        }
        crate::index::bm25::invalidate(root);
    }

    if let Some(cfg) = &embed_config {
        let store_path = match index_dir(root) {
            Ok(d) => d.join("embeddings.bin"),
            Err(e) => {
                tracing::warn!("embed: cannot resolve store path: {e}");
                return;
            }
        };
        crate::embed::update_embeddings_for_files(
            &*index.read().await,
            &changed_files,
            &old_removed_ids,
            cfg,
            &store_path,
        )
        .await;
    }

    let mut meta = IndexMeta::new(root);
    if let Ok(snapshot) = current_source_snapshot(root, &[]) {
        meta.file_mtimes = snapshot.file_mtimes;
        meta.dir_mtimes = snapshot.dir_mtimes;
    }
    meta.source_file_count = source_file_count;
    if let Err(e) = save_meta(&meta, meta_path) {
        eprintln!("pitlane-mcp: failed to flush meta to disk: {}", e);
    }
}

async fn reindex_batch(
    paths: &HashSet<PathBuf>,
    root: &Path,
    indexer: &Arc<Indexer>,
    index: &Arc<RwLock<SymbolIndex>>,
    index_path: &Path,
    meta_path: &Path,
    embed_config: Option<Arc<EmbedConfig>>,
) {
    let removed_ids: Vec<crate::indexer::language::SymbolId>;
    {
        let mut idx = index.write().await;
        // Capture symbol IDs for changed files BEFORE reindex_file mutates by_file
        removed_ids = paths
            .iter()
            .flat_map(|p| idx.by_file.get(p).cloned().unwrap_or_default())
            .collect();
        for path in paths {
            if let Err(e) = indexer.reindex_file(path, root, &mut idx) {
                eprintln!("Error re-indexing {:?}: {}", path, e);
            }
        }

        // Flush updated index to disk while holding the write lock for consistency.
        if let Err(e) = save_index(&idx, index_path) {
            eprintln!("pitlane-mcp: failed to flush index to disk: {}", e);
        }

        // Invalidate the in-memory cache so the next query reloads the fresh
        // snapshot from disk rather than serving stale data.
        crate::cache::invalidate(root);

        // Mark the BM25 index stale so the next BM25 search rebuilds it from
        // the updated symbol set. We remove the ready sentinel rather than rebuilding here
        // to avoid holding the write lock during tantivy I/O.
        if let Ok(tantivy_dir) = index_dir(root).map(|d| d.join("tantivy")) {
            let _ = crate::index::bm25::mark_stale(&tantivy_dir);
        }
        crate::index::bm25::invalidate(root);
    }
    // write lock released — safe to re-acquire index.read() below

    if let Some(cfg) = &embed_config {
        let store_path = match index_dir(root) {
            Ok(d) => d.join("embeddings.bin"),
            Err(e) => {
                tracing::warn!("embed: cannot resolve store path: {e}");
                return;
            }
        };
        crate::embed::update_embeddings_for_files(
            &*index.read().await,
            paths,
            &removed_ids,
            cfg,
            &store_path,
        )
        .await;
    }

    // Update file_mtimes in meta for the changed paths so is_index_up_to_date
    // returns true on the next server start without a forced re-index.
    let mut meta = load_meta(meta_path).unwrap_or_else(|_| IndexMeta::new(root));
    for path in paths {
        let key = path.display().to_string();
        match std::fs::metadata(path) {
            Ok(fs_meta) => {
                if let Ok(modified) = fs_meta.modified() {
                    if let Ok(dur) = modified.duration_since(std::time::UNIX_EPOCH) {
                        meta.file_mtimes.insert(key, dur.as_nanos() as u64);
                    }
                }
            }
            Err(_) => {
                // File was deleted — remove its mtime so it isn't treated as fresh.
                meta.file_mtimes.remove(&key);
            }
        }
        if let Some(parent) = path.parent() {
            if let Ok(fs_meta) = std::fs::metadata(parent) {
                if let Ok(modified) = fs_meta.modified() {
                    if let Ok(dur) = modified.duration_since(std::time::UNIX_EPOCH) {
                        meta.dir_mtimes
                            .insert(parent.display().to_string(), dur.as_nanos() as u64);
                    }
                }
            }
        }
    }
    meta.source_file_count = meta.file_mtimes.len();
    if let Err(e) = save_meta(&meta, meta_path) {
        eprintln!("pitlane-mcp: failed to flush meta to disk: {}", e);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::indexer::registry;
    use std::sync::atomic::AtomicBool;
    use tempfile::TempDir;

    const TEST_DEBOUNCE: Duration = Duration::from_millis(50);

    fn setup(dir: &TempDir) -> (Arc<RwLock<SymbolIndex>>, Arc<Indexer>) {
        let indexer = Arc::new(Indexer::new(registry::build_default_registry()));
        let (index, _) = indexer.index_project(dir.path(), &[]).unwrap();
        (Arc::new(RwLock::new(index)), indexer)
    }

    fn spawn_loop(
        rx: mpsc::Receiver<PathBuf>,
        dir: &TempDir,
        indexer: Arc<Indexer>,
        index: Arc<RwLock<SymbolIndex>>,
    ) -> tokio::task::JoinHandle<()> {
        spawn_loop_with_overflow(rx, dir, indexer, index, Arc::new(AtomicBool::new(false)))
    }

    fn spawn_loop_with_overflow(
        rx: mpsc::Receiver<PathBuf>,
        dir: &TempDir,
        indexer: Arc<Indexer>,
        index: Arc<RwLock<SymbolIndex>>,
        overflowed: Arc<AtomicBool>,
    ) -> tokio::task::JoinHandle<()> {
        let root = dir.path().to_path_buf();
        let index_path = dir.path().join("index.bin");
        let meta_path = dir.path().join("meta.json");
        tokio::spawn(run_debounce_loop(
            rx,
            root,
            indexer,
            index,
            TEST_DEBOUNCE,
            index_path,
            meta_path,
            None,
            overflowed,
        ))
    }

    /// A single modified file is reindexed after the debounce window expires.
    #[tokio::test]
    async fn test_single_path_reindexed_after_window() {
        let dir = TempDir::new().unwrap();
        let file = dir.path().join("lib.rs");
        std::fs::write(&file, b"fn original() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let (tx, rx) = mpsc::channel(16);
        let handle = spawn_loop(rx, &dir, indexer, index.clone());

        std::fs::write(&file, b"fn updated() {}").unwrap();
        tx.send(file).await.unwrap();
        drop(tx);
        handle.await.unwrap();

        let idx = index.read().await;
        let names: Vec<_> = idx.symbols.values().map(|s| s.name.as_str()).collect();
        assert!(names.contains(&"updated"), "expected 'updated' in index");
        assert!(!names.contains(&"original"), "expected 'original' removed");
    }

    #[tokio::test]
    async fn test_reindex_marks_bm25_stale() {
        let dir = TempDir::new().unwrap();
        let file = dir.path().join("lib.rs");
        std::fs::write(&file, b"fn original() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let tantivy_dir = index_dir(dir.path()).unwrap().join("tantivy");
        {
            let idx = index.read().await;
            crate::index::bm25::build(&idx.symbols, &tantivy_dir).unwrap();
        }

        let original_hits =
            crate::index::bm25::search("original", dir.path(), &tantivy_dir, None, None, 10)
                .unwrap();
        assert!(
            original_hits.iter().any(|id| id.contains("original")),
            "precondition: BM25 should find the original symbol"
        );

        let (tx, rx) = mpsc::channel(16);
        let handle = spawn_loop(rx, &dir, indexer, index.clone());

        std::fs::write(&file, b"fn updated() {}").unwrap();
        tx.send(file).await.unwrap();
        drop(tx);
        handle.await.unwrap();

        {
            let idx = index.read().await;
            crate::index::bm25::ensure(&idx.symbols, &tantivy_dir).unwrap();
        }

        let updated_hits =
            crate::index::bm25::search("updated", dir.path(), &tantivy_dir, None, None, 10)
                .unwrap();
        assert!(
            updated_hits.iter().any(|id| id.contains("updated")),
            "BM25 should rebuild after watcher reindex and find the updated symbol"
        );
    }

    /// Multiple distinct paths sent in a burst are all batched into one flush.
    #[tokio::test]
    async fn test_burst_of_paths_all_reindexed() {
        let dir = TempDir::new().unwrap();
        let a = dir.path().join("a.rs");
        let b = dir.path().join("b.rs");
        let c = dir.path().join("c.rs");
        std::fs::write(&a, b"fn a_old() {}").unwrap();
        std::fs::write(&b, b"fn b_old() {}").unwrap();
        std::fs::write(&c, b"fn c_old() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let (tx, rx) = mpsc::channel(16);
        let handle = spawn_loop(rx, &dir, indexer, index.clone());

        std::fs::write(&a, b"fn a_new() {}").unwrap();
        std::fs::write(&b, b"fn b_new() {}").unwrap();
        std::fs::write(&c, b"fn c_new() {}").unwrap();

        // Send all three paths without any delay between them.
        tx.send(a).await.unwrap();
        tx.send(b).await.unwrap();
        tx.send(c).await.unwrap();
        drop(tx);
        handle.await.unwrap();

        let idx = index.read().await;
        let names: Vec<_> = idx.symbols.values().map(|s| s.name.as_str()).collect();
        assert!(names.contains(&"a_new"));
        assert!(names.contains(&"b_new"));
        assert!(names.contains(&"c_new"));
        assert!(!names.contains(&"a_old"));
        assert!(!names.contains(&"b_old"));
        assert!(!names.contains(&"c_old"));
    }

    /// Sending the same path multiple times within one window reindexes it only once
    /// (deduplication via HashSet), so the symbol count stays correct.
    #[tokio::test]
    async fn test_duplicate_paths_deduplicated() {
        let dir = TempDir::new().unwrap();
        let file = dir.path().join("lib.rs");
        std::fs::write(&file, b"fn foo() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let (tx, rx) = mpsc::channel(16);
        let handle = spawn_loop(rx, &dir, indexer, index.clone());

        std::fs::write(&file, b"fn bar() {}").unwrap();

        for _ in 0..10 {
            tx.send(file.clone()).await.unwrap();
        }
        drop(tx);
        handle.await.unwrap();

        let idx = index.read().await;
        // Exactly one symbol should be present despite 10 events for the same file.
        assert_eq!(idx.symbol_count(), 1);
        let names: Vec<_> = idx.symbols.values().map(|s| s.name.as_str()).collect();
        assert!(names.contains(&"bar"));
    }

    /// Closing the channel (sender dropped) flushes pending paths immediately
    /// without waiting for the debounce window to expire.
    #[tokio::test]
    async fn test_channel_close_flushes_pending() {
        let dir = TempDir::new().unwrap();
        let file = dir.path().join("lib.rs");
        std::fs::write(&file, b"fn before() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let (tx, rx) = mpsc::channel(16);
        let handle = spawn_loop(rx, &dir, indexer, index.clone());

        std::fs::write(&file, b"fn after() {}").unwrap();
        tx.send(file).await.unwrap();

        // Drop the sender immediately — loop should flush without sleeping the full window.
        let start = std::time::Instant::now();
        drop(tx);
        handle.await.unwrap();
        let elapsed = start.elapsed();

        // Should complete well before the debounce window.
        assert!(
            elapsed < TEST_DEBOUNCE,
            "flush took {elapsed:?}, expected faster than {TEST_DEBOUNCE:?}"
        );

        let idx = index.read().await;
        let names: Vec<_> = idx.symbols.values().map(|s| s.name.as_str()).collect();
        assert!(names.contains(&"after"));
        assert!(!names.contains(&"before"));
    }

    /// Events for files with unsupported extensions (.txt, .json) are not sent
    /// through the channel, so the index must not change.
    #[tokio::test]
    async fn test_unsupported_extensions_ignored() {
        let dir = TempDir::new().unwrap();
        let rs_file = dir.path().join("lib.rs");
        std::fs::write(&rs_file, b"fn keep() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let (tx, rx) = mpsc::channel(16);
        let handle = spawn_loop(rx, &dir, indexer, index.clone());

        // Only send paths that the handler would filter out — the debounce loop
        // itself does not filter, but we verify the contract expected by the handler.
        let txt = dir.path().join("notes.txt");
        let json = dir.path().join("config.json");
        std::fs::write(&txt, b"hello").unwrap();
        std::fs::write(&json, b"{}").unwrap();

        // Simulate the handler's extension filter: neither path should be sent.
        for path in [&txt, &json] {
            let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
            if ext == "rs" || ext == "py" {
                tx.send(path.clone()).await.unwrap();
            }
        }
        drop(tx);
        handle.await.unwrap();

        // Index must be unchanged — still only the original `keep` symbol.
        let idx = index.read().await;
        assert_eq!(idx.symbol_count(), 1);
        let names: Vec<_> = idx.symbols.values().map(|s| s.name.as_str()).collect();
        assert!(names.contains(&"keep"));
    }

    /// If the notify queue overflows, the debounce loop falls back to a full
    /// resync so dropped paths do not leave the index stale.
    #[tokio::test]
    async fn test_overflow_triggers_full_resync() {
        let dir = TempDir::new().unwrap();
        let a = dir.path().join("a.rs");
        let b = dir.path().join("b.rs");
        std::fs::write(&a, b"fn a_old() {}").unwrap();

        let (index, indexer) = setup(&dir);
        let (tx, rx) = mpsc::channel(16);
        let overflowed = Arc::new(AtomicBool::new(false));
        let handle = spawn_loop_with_overflow(rx, &dir, indexer, index.clone(), overflowed.clone());

        std::fs::write(&a, b"fn a_new() {}").unwrap();
        std::fs::write(&b, b"fn b_new() {}").unwrap();

        tx.send(a).await.unwrap();
        overflowed.store(true, Ordering::Release);

        drop(tx);
        handle.await.unwrap();

        let idx = index.read().await;
        let names: Vec<_> = idx.symbols.values().map(|s| s.name.as_str()).collect();
        assert!(names.contains(&"a_new"));
        assert!(names.contains(&"b_new"));
        assert!(!names.contains(&"a_old"));
    }
}
