//! Watch loop event collection and per-debounce-cycle dispatch.
//!
//! `collect_events` filters raw `notify::Event` into the pending-files /
//! pending-notes sets on `WatchState`; `process_file_changes` /
//! `process_note_changes` drain those sets into reindex calls. The
//! coalesce-then-dispatch shape here is what keeps a 100-file save burst
//! from triggering 100 separate embedder runs.

use super::*;

/// Maximum pending files to prevent unbounded memory growth.
/// Override with CQS_WATCH_MAX_PENDING env var.
pub(super) fn max_pending_files() -> usize {
    static CACHE: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *CACHE.get_or_init(|| {
        std::env::var("CQS_WATCH_MAX_PENDING")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(10_000)
    })
}
/// Collect file system events into pending sets, filtering by extension and deduplicating.
pub(super) fn collect_events(event: &notify::Event, cfg: &WatchConfig, state: &mut WatchState) {
    // Per-event entry span so an inotify event that makes it through
    // filtering can be correlated to its later reindex. `trace_span` keeps
    // the per-event noise behind RUST_LOG=trace; the accept-path debug
    // below is the level operators actually reach for.
    let _span = tracing::trace_span!(
        "collect_events",
        event_kind = ?event.kind,
        paths = event.paths.len(),
    )
    .entered();
    for path in &event.paths {
        // Skip canonicalize for deleted files — dunce::canonicalize
        // requires the file to exist (calls std::fs::canonicalize internally).
        let path = if path.exists() {
            dunce::canonicalize(path).unwrap_or_else(|_| path.clone())
        } else {
            path.clone()
        };
        // Skip .cqs directory. Deleted files can't be canonicalized (they
        // don't exist), so compare normalized string forms to handle slash
        // differences on WSL.
        let norm_path = cqs::normalize_path(&path);
        let norm_cqs = cqs::normalize_path(cfg.cqs_dir);
        if norm_path.starts_with(&norm_cqs) {
            tracing::debug!(path = %norm_path, "Skipping .cqs directory event");
            continue;
        }

        // .gitignore-matched paths are skipped. The matcher was built once
        // at cmd_watch startup; when it's None the user either
        // set CQS_WATCH_RESPECT_GITIGNORE=0, passed --no-ignore, or has no
        // .gitignore. The hardcoded `.cqs/` skip above still runs
        // regardless so the system's own files are always excluded.
        //
        // `matched_path_or_any_parents` walks up the path's parents so
        // that a file at `.claude/worktrees/agent-x/src/lib.rs` is
        // ignored when `.claude/` is in .gitignore. The leaf-only
        // `matched()` would miss this.
        if let Ok(matcher_guard) = cfg.gitignore.read() {
            if let Some(matcher) = matcher_guard.as_ref() {
                if matcher
                    .matched_path_or_any_parents(&path, false)
                    .is_ignore()
                {
                    tracing::trace!(
                        path = %norm_path,
                        "Skipping gitignore-matched path (#1002)"
                    );
                    continue;
                }
            }
        }

        // Check if it's notes.toml
        let norm_notes = cqs::normalize_path(cfg.notes_path);
        if norm_path == norm_notes {
            state.pending_notes = true;
            state.last_event = std::time::Instant::now();
            if state.first_pending_event.is_none() {
                state.first_pending_event = Some(state.last_event);
            }
            continue;
        }

        // Skip if not a supported extension
        let ext_raw = path.extension().and_then(|e| e.to_str()).unwrap_or("");
        let ext = ext_raw.to_ascii_lowercase();
        if !cfg.supported_ext.contains(ext.as_str()) {
            tracing::debug!(path = %path.display(), ext = %ext, "Skipping unsupported extension");
            continue;
        }

        // Convert to relative path
        if let Ok(rel) = path.strip_prefix(cfg.root) {
            // mtime-equality skip is gated on the filesystem's actual mtime
            // resolution, not just WSL drvfs.
            //
            // - `mtime < last`: rewind (e.g. `git checkout` restoring a
            //   commit-time mtime). Skip — the inotify path is for real save
            //   events; a rewound mtime without a matching save is treated
            //   as already-indexed.
            // - `mtime == last`: ambiguous on coarse FS (two saves
            //   inside the same resolution window collide on identical
            //   mtimes), unambiguous on fine FS (nanosecond equality
            //   means the same save). Skip iff `resolution.is_zero()`.
            // - `mtime > last`: a real save advanced the mtime → not
            //   stale, regardless of FS resolution.
            //
            // `coarse_fs_resolution` covers HFS+ / NFS / SMB / FAT32 on
            // plain Linux + macOS too, so they don't silently drop rapid
            // re-saves.
            if let Ok(mtime) = std::fs::metadata(&path).and_then(|m| m.modified()) {
                let resolution = cqs::config::coarse_fs_resolution(&path);
                let stale = state.last_indexed_mtime.get(rel).is_some_and(|last| {
                    if mtime < *last {
                        true
                    } else if mtime == *last {
                        resolution.is_zero()
                    } else {
                        false
                    }
                });
                if stale {
                    tracing::trace!(path = %rel.display(), "Skipping unchanged mtime");
                    continue;
                }
            }
            if state.pending_files.len() < max_pending_files() {
                // Keep the queue keyed on slash-normalized paths so a
                // Windows-side edit and a reconcile-side walk can't
                // double-queue the same file under two separators.
                state
                    .pending_files
                    .insert(super::reconcile::normalize_pending_path(rel));
                // Debug the accept-and-queue branch. Operators investigating
                // "why didn't my save get reindexed?" can
                // `RUST_LOG=cqs::cli::watch=debug` and see the full
                // event → queue chain.
                tracing::debug!(
                    path = %rel.display(),
                    event_kind = ?event.kind,
                    "Queued for reindex"
                );
            } else {
                // Log per-event at debug (spammy on bulk drops) and
                // accumulate a counter; the once-per-cycle summary fires in
                // process_file_changes so operators see the total
                // truncation even if the level is info.
                state.dropped_this_cycle = state.dropped_this_cycle.saturating_add(1);
                tracing::debug!(
                    max = max_pending_files(),
                    path = %rel.display(),
                    "Watch pending_files full, dropping file event"
                );
            }
            state.last_event = std::time::Instant::now();
            // Arm the max-latency clock on the first event of a burst.
            // `last_event` restarts the quiet-gap timer on every event;
            // this one is sticky until the flush drains the pending
            // sets, bounding total delay under a never-quiet stream.
            if state.first_pending_event.is_none() {
                state.first_pending_event = Some(state.last_event);
            }
        }
    }
}

/// Process pending file changes: parse, embed, store atomically, then update HNSW.
///
/// Uses incremental HNSW insertion when an Owned index is available in memory.
/// Falls back to full rebuild on first run or after `hnsw_rebuild_threshold()` incremental inserts.
///
/// On a successful drain the changed-file set is propagated into
/// `siblings`' per-slot delta queues — the active reindex has just
/// written the fresh embeddings back to the global cache, so same-model
/// sibling drains that follow are pure cache hits.
pub(super) fn process_file_changes(
    cfg: &WatchConfig,
    store: &Store,
    state: &mut WatchState,
    siblings: &mut super::siblings::SiblingSet,
) {
    let files: Vec<PathBuf> = state.pending_files.drain().collect();
    let _span = info_span!("process_file_changes", file_count = files.len()).entered();
    state.pending_files.shrink_to(64);

    // Warn at the top so operators see the count, but DO NOT reset the
    // counter here — the outer loop's `publish_watch_snapshot` runs after
    // this function returns, and `WatchSnapshot::compute` uses
    // `dropped_this_cycle > 0` as a Stale signal. If we zero it before the
    // embedder check below (which may early-return on init failure), the
    // snapshot reports `Fresh` even though events were dropped and never
    // reindexed — defeating `cqs eval --require-fresh`. Reset only after a
    // successful drain so the next cycle's snapshot reflects the truthful
    // state.
    if state.dropped_this_cycle > 0 {
        tracing::warn!(
            dropped = state.dropped_this_cycle,
            cap = max_pending_files(),
            "Watch event queue full this cycle; dropping events. Run `cqs index` to catch up"
        );
    }
    // Structured tracing instead of stdout println. The daemon has no
    // terminal — stdout goes to journald via the systemd unit which writes
    // unstructured. Tracing routes through the configured subscriber
    // (journald JSON or stderr text) and honours filter levels.
    tracing::info!(
        file_count = files.len(),
        files = ?files,
        "watch: reindexing changed files",
    );

    let emb = match try_init_embedder(cfg.embedder, &mut state.embedder_backoff, cfg.model_config) {
        Some(e) => e,
        None => return,
    };

    // Capture mtimes BEFORE reindexing to avoid race condition
    let pre_mtimes: HashMap<PathBuf, SystemTime> = files
        .iter()
        .filter_map(|f| {
            std::fs::metadata(cfg.root.join(f))
                .and_then(|m| m.modified())
                .ok()
                .map(|t| (f.clone(), t))
        })
        .collect();

    // Note: concurrent searches during this window may see partial
    // results. Per-file transactions are atomic but the batch is not —
    // files indexed so far are visible, remaining are stale. Self-heals
    // after HNSW rebuild. Acceptable for a dev tool.
    //
    // Mark both HNSW kinds dirty before writing chunks. The base index
    // derives from the same chunks as enriched, so a crash mid-write can
    // leave either graph stale.
    if let Err(e) = store.set_hnsw_dirty(cqs::HnswKind::Enriched, true) {
        tracing::warn!(error = %e, "Cannot set enriched HNSW dirty flag — skipping reindex to prevent stale index on crash");
        return;
    }
    if let Err(e) = store.set_hnsw_dirty(cqs::HnswKind::Base, true) {
        tracing::warn!(error = %e, "Cannot set base HNSW dirty flag — skipping reindex to prevent stale index on crash");
        return;
    }
    // Wall-clock the reindex pass so `cqs status --watch` can report
    // last-reindex latency — previously only visible as a
    // tracing span duration in journalctl.
    let reindex_started = std::time::Instant::now();
    match reindex_files(
        cfg.root,
        store,
        &files,
        cfg.parser,
        emb,
        cfg.global_cache,
        cfg.quiet,
    ) {
        Ok((count, content_hashes)) => {
            // Propagate the drained delta to sibling slots. Strictly
            // after the active reindex succeeded: the global cache now
            // holds the fresh embeddings, so a same-model sibling drain
            // resolves the whole set via `resolve_reuse` without
            // touching the embedder. O(files × slots) set inserts —
            // does not affect active-slot freshness latency.
            siblings.enqueue(&files);
            // Publish-side stats for `cqs status --watch`: a
            // timestamp + duration pair the snapshot publisher reads
            // every tick.
            state.last_reindex = Some(cqs::watch_status::ReindexLatency {
                at_unix_secs: cqs::unix_secs_i64().unwrap_or(0),
                duration_ms: u64::try_from(reindex_started.elapsed().as_millis())
                    .unwrap_or(u64::MAX),
                files: u64::try_from(files.len()).unwrap_or(u64::MAX),
            });
            // Record mtimes to skip duplicate events
            for (file, mtime) in pre_mtimes {
                state.last_indexed_mtime.insert(file, mtime);
            }
            // Reset only after a successful drain. The dropped events
            // surfaced in the warn above are also queued for reconcile
            // (Layer 2) on the next idle pass, so the count stays meaningful
            // exactly until the reconcile refills `pending_files` with the
            // same paths.
            state.dropped_this_cycle = 0;
            // Recency prune for the mtime map, using the map's `SystemTime`
            // values to age out stale entries in-memory. Avoids a
            // `Path::exists()` per entry, which on WSL 9P mounts issued up to
            // 5000 serial `stat()` syscalls on the watch thread. Re-adding a
            // surviving file on its next event is a trivial insert.
            let pruned = prune_last_indexed_mtime(&mut state.last_indexed_mtime);
            if pruned > 0 {
                tracing::debug!(
                    pruned,
                    remaining = state.last_indexed_mtime.len(),
                    "Pruned stale last_indexed_mtime entries"
                );
            }
            if !cfg.quiet {
                println!("Indexed {} chunk(s)", count);
            }

            // Incremental SPLADE encoding. Encoder is held in WatchConfig
            // and stays resident for the daemon's lifetime. We encode every
            // chunk in the files that were reindexed — upsert_sparse_vectors
            // is idempotent, so re-encoding an unchanged chunk is correct
            // just slightly wasteful.
            if count > 0 {
                match cfg.splade_encoder {
                    Some(encoder_mu) => {
                        // Build the list of files that actually had chunks
                        // reindexed (excluding deleted ones, which are
                        // handled by the FK CASCADE on DELETE FROM chunks).
                        // We re-use the original `files` snapshot — the
                        // ones that survived parsing are still tracked.
                        encode_splade_for_changed_files(encoder_mu, store, &files);
                    }
                    None if cqs::splade::resolve_splade_model_dir().is_some() => {
                        tracing::debug!(
                            new_chunks = count,
                            "SPLADE model present but encoder disabled this daemon — \
                             sparse coverage will drift until manual 'cqs index' \
                             (CQS_WATCH_INCREMENTAL_SPLADE=0 or load failed)"
                        );
                    }
                    None => {
                        // No SPLADE model configured — nothing to do.
                    }
                }
            }

            // Refresh the observed store stamp now that this cycle's writes
            // (chunk reindex + SPLADE encode) are committed. Every HNSW save
            // below passes it as the snapshot: a live stamp that has moved
            // past it at save time means a writer this loop never observed,
            // and the save gets discarded instead of overwriting newer
            // sidecars with contents that lack that writer's chunks.
            match cqs::hnsw::StoreStamp::read(store) {
                Ok(s) => state.observed_stamp = Some(s),
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "Failed to read store stamp after reindex — the next HNSW save \
                         falls back to the live stamp (no foreign-writer detection)"
                    );
                    state.observed_stamp = None;
                }
            }

            // === HNSW maintenance ===
            //
            // Rebuilds run in a background thread (`spawn_hnsw_rebuild`).
            // The watch loop's responsibilities each cycle are:
            //
            //   1. Fetch this cycle's committed (id, embedding, hash) triples
            //      and — if a rebuild is in flight — merge them into its
            //      delta BEFORE the drain. Ordering invariant: the drain's
            //      sidecar save is stamped with `observed_stamp`, which was
            //      refreshed above and already covers this cycle's writes; a
            //      drain that saved (and cleared the dirty flag) without
            //      these pairs would persist an index missing chunks its
            //      stamp claims to contain, and a crash before the follow-up
            //      incremental save would let the dirty-path self-heal trust
            //      it.
            //   2. Drain a completed rebuild — replay the delta (now
            //      including this cycle) into the new index, save, and
            //      atomically swap into `state.hnsw_index`.
            //   3. Decide whether to start a *new* rebuild (Owned-needed, or
            //      threshold reached).
            //   4. If the drain did NOT consume this cycle's pairs (none was
            //      pending, or pending was cleared without a swap), queue
            //      them into the freshly spawned rebuild's delta or take the
            //      fast incremental path on the in-memory Owned index.
            //
            // The result: incremental_insert never blocks on a full rebuild,
            // editor saves don't pause for 10-30s of CUDA work, and search
            // keeps using the prior index until the new one is ready.

            // 1. Fetch this cycle's pairs; pre-drain merge into the delta.
            let pairs: Vec<(String, Embedding, String)> = if content_hashes.is_empty() {
                Vec::new()
            } else {
                let hash_refs: Vec<&str> = content_hashes.iter().map(|s| s.as_str()).collect();
                match store.get_chunk_ids_and_embeddings_by_hashes(&hash_refs) {
                    Ok(p) => p,
                    Err(e) => {
                        warn!(error = %e, "Failed to fetch embeddings for HNSW update");
                        Vec::new()
                    }
                }
            };
            if !pairs.is_empty() {
                if let Some(ref mut pending) = state.pending_rebuild {
                    // Clone: if the drain clears pending without a swap
                    // (error / saturated / stale-discard), the delta dies
                    // with it and the locals below re-handle the pairs.
                    // Bounded by one cycle's file count — small next to the
                    // rebuild itself.
                    pending.queue_delta(pairs.clone(), store.dim(), cfg.quiet);
                }
            }

            // 2. Drain a completed rebuild, if any.
            let drain_outcome = drain_pending_rebuild(cfg, store, state);
            // Pairs are consumed when the drain replayed them (Swapped) or
            // the delta retaining them is still pending (StillRunning).
            let pairs_consumed = matches!(
                drain_outcome,
                super::rebuild::DrainOutcome::Swapped | super::rebuild::DrainOutcome::StillRunning
            );

            let rebuild_in_flight = state.pending_rebuild.is_some();
            // The periodic threshold rebuild exists to clean orphaned HNSW
            // vectors and absorb accumulated deltas. When the tiered backend
            // serves queries, that compaction happens inside cuVS (the
            // brute-force tier absorbs adds; the CAGRA tier rebuilds
            // internally), so the expensive periodic full HNSW rebuild is
            // redundant — skip it. The first-build case (`hnsw_index.is_none()`)
            // still proceeds so the cheap CPU-side fallback / base-index router
            // stays fed. `tiered_index_active()` is always false without the
            // `tiered-index` feature, so default and CAGRA builds keep the
            // original threshold behavior unchanged.
            let threshold_rebuild_due =
                !tiered_index_active() && state.incremental_count >= hnsw_rebuild_threshold();
            let needs_owned = state.hnsw_index.is_none() || threshold_rebuild_due;

            // 3. Start a new rebuild, if appropriate.
            if needs_owned && !rebuild_in_flight {
                let context = if state.hnsw_index.is_none() {
                    "rebuild_from_empty"
                } else {
                    "threshold_rebuild"
                };
                let pending = spawn_hnsw_rebuild(
                    cfg.cqs_dir.to_path_buf(),
                    cfg.cqs_dir.join(cqs::INDEX_DB_FILENAME),
                    store.dim(),
                    context,
                );
                info!(context, "Spawned background HNSW rebuild");
                if !cfg.quiet {
                    println!(
                        "  HNSW index: rebuild started in background ({}, search keeps using current index)",
                        context
                    );
                }
                state.pending_rebuild = Some(pending);
            }

            // 4. Re-handle pairs the drain didn't consume: queue into the
            //    just-spawned rebuild's delta, or run the fast incremental
            //    path. (`StillRunning` implies the spawn block above didn't
            //    fire, so a Some(pending) here is always the fresh spawn.)
            if !pairs.is_empty() && !pairs_consumed {
                if let Some(ref mut pending) = state.pending_rebuild {
                    // A rebuild just spawned this cycle. Its SQLite
                    // snapshot may not include these chunks — capture
                    // them so `drain_pending_rebuild` can replay them
                    // after the swap (the fingerprint dedup skips any
                    // the snapshot did consume).
                    pending.queue_delta(pairs, store.dim(), cfg.quiet);
                } else if let Some(ref mut index) = state.hnsw_index {
                    // Fast incremental path — Owned in memory, no rebuild pending.
                    // Modified chunks get new IDs; old vectors become orphans
                    // in the HNSW graph (hnsw_rs has no deletion). Orphans are
                    // harmless: search post-filters against live SQLite chunk
                    // IDs. They're cleaned on the next threshold rebuild.
                    //
                    // `pairs` carries content_hash as the third tuple
                    // slot for the rebuild-window path; the
                    // incremental insert only needs (id, embedding).
                    let items: Vec<(String, &[f32])> = pairs
                        .iter()
                        .map(|(id, emb, _hash)| (id.clone(), emb.as_slice()))
                        .collect();
                    match index.insert_batch(&items) {
                        Ok(n) => {
                            state.incremental_count += n;
                            // Stamped save: the in-memory index now
                            // reflects everything this loop has
                            // observed (= observed_stamp, refreshed
                            // above). A live-stamp mismatch under
                            // the save lock means a foreign writer's
                            // chunks are missing from this index —
                            // discard the save and leave the dirty
                            // flag set rather than overwrite the
                            // newer on-disk sidecars.
                            let snapshot = match state.observed_stamp {
                                Some(s) => Ok(s),
                                None => cqs::hnsw::StoreStamp::read(store).map_err(|e| {
                                    cqs::hnsw::HnswError::Internal(format!(
                                        "Failed to read store stamp for incremental save: {e}"
                                    ))
                                }),
                            };
                            match snapshot.and_then(|snap| {
                                index.save_stamped(cfg.cqs_dir, "index", store, snap)
                            }) {
                                Ok(cqs::hnsw::SaveOutcome::Saved) => {
                                    clear_hnsw_dirty_with_retry(
                                        store,
                                        cqs::HnswKind::Enriched,
                                        "incremental_insert",
                                    );
                                }
                                Ok(cqs::hnsw::SaveOutcome::DiscardedStale) => {
                                    warn!(
                                        "HNSW incremental save discarded: store moved \
                                                 past the watch loop's observed state \
                                                 (concurrent indexer); dirty flag stays set"
                                    );
                                }
                                Err(e) => {
                                    warn!(error = %e, "Failed to save HNSW after incremental insert");
                                }
                            }
                            info!(
                                inserted = n,
                                total = index.len(),
                                incremental_count = state.incremental_count,
                                "HNSW incremental insert"
                            );
                            if !cfg.quiet {
                                println!(
                                    "  HNSW index: +{} vectors (incremental, {} total)",
                                    n,
                                    index.len()
                                );
                            }
                        }
                        Err(e) => {
                            // Insert failed. Queue a background rebuild
                            // rather than blocking on a synchronous one
                            // — search keeps serving from the current
                            // index meanwhile.
                            warn!(
                                error = %e,
                                "HNSW incremental insert failed; spawning background rebuild"
                            );
                            let pending = spawn_hnsw_rebuild(
                                cfg.cqs_dir.to_path_buf(),
                                cfg.cqs_dir.join(cqs::INDEX_DB_FILENAME),
                                store.dim(),
                                "incremental_insert_failure",
                            );
                            // Carry these new chunks over into the new
                            // rebuild's delta so they survive the swap
                            // (cap-aware, same as every delta append).
                            let mut p = pending;
                            p.queue_delta(pairs, store.dim(), cfg.quiet);
                            state.pending_rebuild = Some(p);
                        }
                    }
                }
                // No pending and no in-memory index → first save with
                // empty store. The needs_owned branch above already
                // spawned a rebuild this cycle; pairs were captured
                // there. Nothing to do here.
            }
        }
        Err(e) => {
            warn!(error = %e, "Reindex error");
            // Surface the failure to `cqs status --watch` so
            // operators don't need journalctl to see why the index is
            // behind. Sticky — see `WatchErrorInfo` docs.
            state.last_error = Some(cqs::watch_status::WatchErrorInfo {
                at_unix_secs: cqs::unix_secs_i64().unwrap_or(0),
                message: format!("reindex failed: {e}"),
            });
            // The dirty flag was set before the reindex attempt and the
            // success path's `clear_hnsw_dirty_with_retry` is unreachable
            // from this arm. Surface that the HNSW stays marked dirty on
            // disk until a successful reindex cycle clears it — operators
            // correlate this with the prior `Reindex error` to diagnose
            // persistent dirty state (SQLite busy / OOM / etc.); search may
            // serve stale results in the meantime.
            tracing::warn!(
                hnsw_kinds = "enriched,base",
                "HNSW dirty flag remains set after reindex failure; \
                 search may serve stale results until next successful \
                 reindex cycle clears it"
            );
        }
    }
}

/// Process notes.toml changes: parse and store notes (no embedding needed).
///
/// Takes `&mut WatchState` so a notes-reindex failure lands in
/// `state.last_error` for `cqs status --watch`, same as the
/// file-reindex path.
pub(super) fn process_note_changes(
    root: &Path,
    store: &Store,
    quiet: bool,
    state: &mut WatchState,
) {
    if !quiet {
        println!("\nNotes changed, reindexing...");
    }
    match reindex_notes(root, store, quiet) {
        Ok(count) => {
            if !quiet {
                println!("Indexed {} note(s)", count);
            }
        }
        Err(e) => {
            warn!(error = %e, "Notes reindex error");
            state.last_error = Some(cqs::watch_status::WatchErrorInfo {
                at_unix_secs: cqs::unix_secs_i64().unwrap_or(0),
                message: format!("notes reindex failed: {e}"),
            });
        }
    }
}
