basemind 0.29.1

//! Detached background facilities spawned by `serve`: blob GC and the two filesystem watchers.

use std::sync::Arc;
use std::time::Duration;

use super::helpers;
use super::{MapCache, ServerState};

/// Run an in-process blob GC once, logging the outcome and swallowing any error.
///
/// Uses the UNLOCKED `store_gc` primitives (`collect_referenced_hashes` + `gc_blobs`)
/// under a `blocking_read()` store guard — NEVER `store_gc::run_gc`, which re-acquires
/// the `.basemind/.lock` flock that `serve` already holds (that would deadlock). The
/// held read guard blocks the only in-process writer (`scan_and_refresh`) for the
/// mark+sweep; cross-process scans are impossible because serve holds the flock.
pub(super) async fn run_background_gc(state: Arc<ServerState>) {
    if state.shared.store.read().await.blobs_shared {
        tracing::debug!("background blob GC skipped: blob cache is shared across git worktrees");
        return;
    }
    let result = tokio::task::spawn_blocking(move || {
        let store = state.shared.store.blocking_read();
        let referenced = crate::store_gc::collect_referenced_hashes(&store.basemind_dir)?;
        crate::store_gc::gc_blobs(&referenced)
    })
    .await;
    match result {
        Ok(Ok(report)) if report.removed > 0 => tracing::info!(
            removed = report.removed,
            bytes_freed = report.bytes_freed,
            "background blob GC reclaimed orphaned blobs"
        ),
        Ok(Ok(_)) => tracing::debug!("background blob GC: nothing to reclaim"),
        Ok(Err(error)) => tracing::warn!(%error, "background blob GC failed"),
        Err(error) => tracing::warn!(%error, "background blob GC task panicked"),
    }
}

/// Boot-time initial index build for an empty index, spawned once from `BasemindServer::new`.
///
/// Two passes so serve becomes queryable fast without pinning the machine on ONNX embedding:
/// 1. A `Deferred` scan writes the code-map + BM25 keyword lane + content-addressed blobs but NO
///    embeddings — this is what clears `initial_scan_active`, so `status` reports the index ready.
/// 2. A detached `Inline` scan then fills the vectors the fast pass skipped, reusing the fast pass'
///    content-addressed caches (only not-yet-embedded content is embedded, bounded by WS4-A's embed
///    pool). GC runs after it settles so the sweep reaps against the final blob set.
pub(super) fn spawn_initial_scan(state: Arc<ServerState>) {
    tracing::info!("empty index on startup; running initial scan in background");
    #[cfg(all(feature = "comms", any(unix, windows)))]
    if state.shared.daemon_writer {
        tokio::spawn(async move {
            use std::sync::atomic::Ordering;
            state.shared.initial_scan_active.store(true, Ordering::Relaxed);
            let started = std::time::Instant::now();
            match super::daemon_forward::writer_rescan_and_refresh(&state, None, false, false).await {
                Ok(report) => tracing::info!(
                    scanned = report.scanned,
                    updated = report.updated,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "initial scan complete (via daemon writer; embeddings deferred)"
                ),
                Err(error) => tracing::warn!(%error, "initial writer scan failed"),
            }
            state
                .shared
                .initial_scan_ms
                .store(started.elapsed().as_millis() as u64, Ordering::Relaxed);
            state.shared.initial_scan_active.store(false, Ordering::Relaxed);
            let embed_state = Arc::clone(&state);
            tokio::spawn(async move {
                let embed_started = std::time::Instant::now();
                tracing::info!("background embedding pass starting (forwarded to daemon)");
                match super::daemon_forward::writer_rescan_and_refresh(&embed_state, None, false, true).await {
                    Ok(report) => tracing::info!(
                        scanned = report.scanned,
                        updated = report.updated,
                        elapsed_ms = embed_started.elapsed().as_millis() as u64,
                        "background embedding pass complete (via daemon writer)"
                    ),
                    Err(error) => tracing::warn!(%error, "background writer embedding pass failed"),
                }
            });
        });
        return;
    }
    tokio::spawn(async move {
        use std::sync::atomic::Ordering;
        state.shared.initial_scan_active.store(true, Ordering::Relaxed);
        let started = std::time::Instant::now();
        match helpers::scan_and_refresh(Arc::clone(&state), None, crate::scanner::EmbedMode::Deferred).await {
            Ok(report) => tracing::info!(
                scanned = report.stats.scanned,
                updated = report.stats.updated,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "initial background scan complete (code-map + keyword lane; embeddings deferred)"
            ),
            Err(error) => tracing::warn!(%error, "initial background scan failed"),
        }
        state
            .shared
            .initial_scan_ms
            .store(started.elapsed().as_millis() as u64, Ordering::Relaxed);
        state.shared.initial_scan_active.store(false, Ordering::Relaxed);
        let embed_state = Arc::clone(&state);
        tokio::spawn(async move {
            let embed_started = std::time::Instant::now();
            tracing::info!("background embedding pass starting");
            match helpers::scan_and_refresh(Arc::clone(&embed_state), None, crate::scanner::EmbedMode::Inline).await {
                Ok(report) => tracing::info!(
                    scanned = report.stats.scanned,
                    updated = report.stats.updated,
                    elapsed_ms = embed_started.elapsed().as_millis() as u64,
                    "background embedding pass complete"
                ),
                Err(error) => tracing::warn!(%error, "background embedding pass failed"),
            }
            run_background_gc(embed_state).await;
        });
    });
}

/// Boot-time in-RAM code-map preload, spawned once from `BasemindServer::new_with_options` when the
/// index is already populated (no initial scan needed) on the background `serve` path.
///
/// `serve` boots with an EMPTY [`MapCache`] placeholder so it can answer the MCP `initialize` /
/// `tools/list` handshake immediately, then this task does the heavy `MapCache::build` (a rayon
/// `par_iter` over every L1/L2 blob) on a blocking thread, publishes the full map via `ArcSwap`, and
/// wakes every tool awaiting [`ServerState::cache_ready`]. Without this, that build ran synchronously
/// before `.serve(transport)` and — under rayon-pool contention from other sessions' scans — could take
/// minutes, blowing the client's startup window so the tools never registered.
pub(super) fn spawn_cache_warm(state: Arc<ServerState>) {
    tracing::info!("warming in-RAM code map in background (handshake already served)");
    tokio::spawn(async move {
        use std::sync::atomic::Ordering;
        let started = std::time::Instant::now();
        let build_state = Arc::clone(&state);
        let built = tokio::task::spawn_blocking(move || {
            let store = build_state.shared.store.blocking_read();
            let budget = super::l1_cache::budget_bytes_from(&build_state.shared.config.resources);
            let mut cache = MapCache::build(&store, budget);
            // Load persisted doc↔code links (ADR-0008) on this blocking thread — the LanceStore's own
            // block_on must not nest inside the async reactor.
            super::doc_links_cache::attach(
                &mut cache,
                &store,
                &build_state.shared.config,
                &build_state.shared.scope,
            );
            cache
        })
        .await;
        match built {
            Ok(cache) => {
                let files = cache.len();
                let cache = Arc::new(cache);
                state.shared.cache.store(Arc::clone(&cache));
                state.shared.cache_generation.fetch_add(1, Ordering::Relaxed);
                // Build the `symbols` / `dependents` term index off the query path so the first
                // lookup after boot is as fast as every later one.
                cache.warm_terms();
                state
                    .shared
                    .cache_warm_ms
                    .store(started.elapsed().as_millis() as u64, Ordering::Relaxed);
                state.shared.cache_warming.store(false, Ordering::Relaxed);
                state.shared.cache_ready.notify_waiters();
                tracing::info!(
                    files,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "in-RAM code map warm complete"
                );
            }
            Err(error) => {
                state.shared.cache_warming.store(false, Ordering::Relaxed);
                state.shared.cache_ready.notify_waiters();
                tracing::error!(%error, "in-RAM code map warm task panicked; serving un-warmed cache");
            }
        }
    });
}

/// Run one debounced batch of changed paths through the writer and report `(scanned, updated,
/// removed)`. A `daemon_writer` serve forwards the batch to the daemon (the sole writer) and
/// rebuilds its read-only map; every other serve scans locally under its own write lock. Bridges
/// the watcher's blocking std thread to the async writer via the captured runtime `Handle`.
fn refresh_batch(
    handle: &tokio::runtime::Handle,
    state: &Arc<ServerState>,
    paths: Vec<std::path::PathBuf>,
) -> Result<(usize, usize, usize), String> {
    #[cfg(all(feature = "comms", any(unix, windows)))]
    if state.shared.daemon_writer {
        let report = handle
            .block_on(super::daemon_forward::writer_rescan_and_refresh(
                state,
                Some(paths),
                false,
                true,
            ))
            .map_err(|error| error.to_string())?;
        return Ok((report.scanned, report.updated, report.removed));
    }
    let report = handle
        .block_on(helpers::scan_and_refresh(
            Arc::clone(state),
            Some(paths),
            crate::scanner::EmbedMode::Inline,
        ))
        .map_err(|error| error.to_string())?;
    Ok((report.stats.scanned, report.stats.updated, report.stats.removed))
}

const VIEW_WATCHER_DEBOUNCE: Duration = Duration::from_millis(150);
const VIEW_WATCHER_SHUTDOWN_POLL: Duration = Duration::from_millis(200);

/// Bound on the view watcher's debouncer→consumer queue, in whole debounced batches.
///
/// A MapCache rebuild is synchronous and takes as long as reopening the index plus rebuilding the
/// map, so a writer rewriting the index in a loop can outrun this consumer; unbounded, every one of
/// those rewrites would sit in the queue holding its event paths. 8 slots is ~1.2 s of backlog at
/// [`VIEW_WATCHER_DEBOUNCE`] and the consumer collapses whatever it finds into a single rebuild
/// anyway, so a deeper queue would only buy duplicate work. Overflow sets `pending_touch` instead of
/// blocking the debouncer's emit thread or dropping the notification — a dropped index rewrite would
/// leave the served MapCache permanently stale.
const VIEW_WATCHER_QUEUE_CAPACITY: usize = 8;

/// Does a debounced batch touch the view's index file? That one path is the whole of the view
/// watcher's event filtering, so it runs both in the consumer and — on queue overflow — in the
/// debouncer callback.
fn touches_index(events: &[notify_debouncer_full::DebouncedEvent], target: &std::path::Path) -> bool {
    events
        .iter()
        .any(|de| de.event.paths.iter().any(|p| p.as_path() == target))
}

/// Drop guard that requests shutdown of a filesystem watcher.
pub(super) struct WatcherGuard {
    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
}

impl Drop for WatcherGuard {
    fn drop(&mut self) {
        if let Some(shutdown) = self.shutdown.take() {
            let _ = shutdown.send(());
        }
    }
}

/// Spawn a working-tree watcher whose lifetime is controlled by the returned guard.
///
/// Unlike [`spawn_view_watcher`], this watches source paths and bridges each debounced refresh from
/// a blocking OS thread back through the current Tokio runtime. Dropping the guard requests
/// shutdown and releases the watcher-owned server state.
pub(super) fn spawn_serve_watcher(state: Arc<ServerState>) -> WatcherGuard {
    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    spawn_serve_watcher_thread(state, shutdown_rx);
    WatcherGuard {
        shutdown: Some(shutdown_tx),
    }
}

fn spawn_serve_watcher_thread(state: Arc<ServerState>, shutdown_rx: tokio::sync::oneshot::Receiver<()>) {
    let root = state.shared.root.clone();
    let config = Arc::clone(&state.shared.config);
    let handle = tokio::runtime::Handle::current();
    std::thread::Builder::new()
        .name("basemind-mcp-serve-watcher".to_string())
        .spawn(move || {
            tracing::info!(root = %root.display(), "serve watcher armed (live incremental rescan)");
            let result = crate::watcher::watch_paths(&root, &config, shutdown_rx, |paths, _kind| {
                use std::sync::atomic::Ordering;
                let refresh_state = Arc::clone(&state);
                refresh_state.shared.rescan_active.store(true, Ordering::Relaxed);
                let outcome = refresh_batch(&handle, &refresh_state, paths);
                refresh_state.shared.rescan_active.store(false, Ordering::Relaxed);
                match outcome {
                    Ok((scanned, updated, removed)) => {
                        tracing::debug!(scanned, updated, removed, "serve watcher: incremental rescan complete")
                    }
                    Err(error) => tracing::warn!(
                        %error,
                        "serve watcher: incremental rescan failed (watcher continues)"
                    ),
                }
            });
            if let Err(error) = result {
                tracing::warn!(%error, "serve watcher exited with error");
            }
            tracing::info!("serve watcher: exiting");
        })
        .ok();
}

/// Reopen the working store read-only for a MapCache rebuild. A `daemon_writer` serve opens
/// blobs-only (never the fjall index) so it can't steal the exclusive index lock from its own
/// daemon (the sole writer); every other serve opens the index normally.
fn reopen_read_only(state: &ServerState, view: &str) -> Result<crate::store::Store, crate::store::StoreError> {
    #[cfg(all(feature = "comms", any(unix, windows)))]
    if state.shared.daemon_writer {
        return crate::store::Store::open_read_only_no_index(state.shared.root.as_path(), view);
    }
    crate::store::Store::open_read_only(state.shared.root.as_path(), view)
}

/// Spawn a passive index watcher whose lifetime is controlled by the returned guard.
pub(super) fn spawn_view_watcher(state: Arc<ServerState>) -> WatcherGuard {
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();
    let guard = WatcherGuard {
        shutdown: Some(shutdown_tx),
    };
    let (basemind_dir, view) = {
        let store = match state.shared.store.try_read() {
            Ok(g) => g,
            Err(_) => return guard,
        };
        (store.basemind_dir.clone(), store.view.clone())
    };
    let view_dir = basemind_dir.join(crate::store::VIEWS_DIR).join(&view);
    let target = view_dir.join(crate::store::INDEX_FILE);

    std::thread::Builder::new()
        .name("basemind-mcp-view-watcher".to_string())
        .spawn(move || {
            use notify::RecommendedWatcher;
            use notify_debouncer_full::{DebounceEventResult, NoCache, new_debouncer_opt};
            use std::sync::atomic::{AtomicBool, Ordering};

            let (tx, rx) = std::sync::mpsc::sync_channel::<DebounceEventResult>(VIEW_WATCHER_QUEUE_CAPACITY);
            // Overflow state for a full queue: one bit, because every batch that reaches `target`
            // asks for the same single rebuild. Set from the debouncer's emit thread, which must
            // never block on this queue.
            let pending_touch = Arc::new(AtomicBool::new(false));
            let touch_producer = Arc::clone(&pending_touch);
            let target_producer = target.clone();
            let handler = move |result: DebounceEventResult| {
                if let Err(std::sync::mpsc::TrySendError::Full(Ok(events))) = tx.try_send(result)
                    && touches_index(&events, &target_producer)
                {
                    touch_producer.store(true, Ordering::Relaxed);
                }
            };
            // ~keep NoCache, not the default FileIdMap — see src/watcher.rs (issue #43). We only
            // ~keep compare event paths to `target`, so the FileId rename cache is dead weight here.
            let mut debouncer = match new_debouncer_opt::<_, RecommendedWatcher, NoCache>(
                VIEW_WATCHER_DEBOUNCE,
                None,
                handler,
                NoCache::new(),
                notify::Config::default(),
            ) {
                Ok(d) => d,
                Err(e) => {
                    tracing::warn!(error = %e, "view watcher: failed to start debouncer");
                    return;
                }
            };
            if let Err(e) = debouncer.watch(&view_dir, notify::RecursiveMode::NonRecursive) {
                tracing::warn!(error = %e, dir = %view_dir.display(), "view watcher: failed to watch");
                return;
            }
            tracing::info!(target = %target.display(), "view watcher armed");

            loop {
                if !matches!(
                    shutdown_rx.try_recv(),
                    Err(tokio::sync::oneshot::error::TryRecvError::Empty)
                ) {
                    break;
                }
                let mut rebuild = false;
                match rx.recv_timeout(VIEW_WATCHER_SHUTDOWN_POLL) {
                    Ok(Ok(events)) => rebuild |= touches_index(&events, &target),
                    Ok(Err(_)) => {}
                    Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
                    Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
                }
                // Collapse the rest of the queue into this iteration: N index rewrites that piled up
                // during the last rebuild still need exactly one rebuild.
                for _ in 1..VIEW_WATCHER_QUEUE_CAPACITY {
                    match rx.try_recv() {
                        Ok(Ok(events)) => rebuild |= touches_index(&events, &target),
                        Ok(Err(_)) => {}
                        Err(_) => break,
                    }
                }
                // Whatever overflowed while the last rebuild ran. Cleared as it is read, and read on
                // every iteration (the idle timeout included) so it can never be stranded.
                rebuild |= pending_touch.swap(false, Ordering::Relaxed);
                if !rebuild {
                    continue;
                }
                let view = state
                    .shared
                    .store
                    .try_read()
                    .map(|g| g.view.clone())
                    .unwrap_or_default();
                let new_store = match reopen_read_only(&state, &view) {
                    Ok(s) => s,
                    Err(e) => {
                        tracing::warn!(error = %e, "view watcher: store reopen failed");
                        continue;
                    }
                };
                let fingerprint = super::map_fingerprint::index_fingerprint(&new_store);
                if fingerprint == state.shared.cache.load().fingerprint {
                    tracing::debug!("view watcher: index rewritten but unchanged; keeping the current MapCache");
                    continue;
                }
                let budget = super::l1_cache::budget_bytes_from(&state.shared.config.resources);
                let mut rebuilt = MapCache::build(&new_store, budget);
                // Reload doc↔code links (ADR-0008) after a refreshed-index rebuild; safe here because
                // this runs on a plain std thread with no tokio runtime entered.
                super::doc_links_cache::attach(&mut rebuilt, &new_store, &state.shared.config, &state.shared.scope);
                let new_cache = Arc::new(rebuilt);
                tracing::info!(
                    files = new_cache.len(),
                    "view watcher: rebuilt MapCache from refreshed index"
                );
                state.shared.cache.store(new_cache);
                state
                    .shared
                    .cache_generation
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            }
            tracing::info!("view watcher: channel closed; exiting");
        })
        .ok();
    guard
}