use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use anyhow::{Context, Result};
use tokio::sync::RwLock;

use crate::db::LanceDb;

/// Default batch interval for reindex processing (30 seconds).
const DEFAULT_BATCH_INTERVAL_SECS: u64 = 30;

/// Default staleness threshold for source code files (5 minutes).
const DEFAULT_STALENESS_THRESHOLD_SECS: u64 = 300;

/// Maximum number of files to process in a single batch run so that one
/// enormous queue does not block the lock for too long.
const MAX_BATCH_SIZE: usize = 500;

/// Manages the freshness of indexed documents by tracking which files need
/// re-indexing and periodically flushing the queue into the vector store.
///
/// Two complementary APIs are provided:
///
/// 1. **Queue-based reindexing** -- callers push file paths via
///    [`mark_for_reindex`] and a background task periodically drains the queue
///    (see [`start_background_sync`]).
///
/// 2. **On-demand freshness checks** -- callers can query [`is_fresh`] or
///    [`get_staleness`] to determine whether a file's vector embedding is
///    up-to-date relative to its on-disk modification time.
pub struct FreshnessManager {
    vector_db: Arc<LanceDb>,
    /// Last-known modification time per file path (in-memory cache).
    known_mtimes: RwLock<HashMap<PathBuf, i64>>,
    /// Files waiting to be re-indexed.
    reindex_queue: Arc<RwLock<Vec<PathBuf>>>,
    batch_interval: Duration,
}

impl FreshnessManager {
    /// Create a new `FreshnessManager` backed by the given vector database.
    pub fn new(vector_db: Arc<LanceDb>) -> Self {
        Self {
            vector_db,
            known_mtimes: RwLock::new(HashMap::new()),
            reindex_queue: Arc::new(RwLock::new(Vec::new())),
            batch_interval: Duration::from_secs(DEFAULT_BATCH_INTERVAL_SECS),
        }
    }

    /// Create a new `FreshnessManager` with a custom batch interval.
    pub fn with_batch_interval(vector_db: Arc<LanceDb>, interval: Duration) -> Self {
        Self {
            vector_db,
            known_mtimes: RwLock::new(HashMap::new()),
            reindex_queue: Arc::new(RwLock::new(Vec::new())),
            batch_interval: interval,
        }
    }

    // -----------------------------------------------------------------------
    // Background sync
    // -----------------------------------------------------------------------

    /// Start a background task that periodically drains the reindex queue.
    ///
    /// The spawned task runs indefinitely, sleeping for `batch_interval`
    /// between processing rounds.  Errors during processing are logged but
    /// do not terminate the loop.
    pub async fn start_background_sync(self: Arc<Self>) {
        let interval = self.batch_interval;
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(interval).await;
                if let Err(e) = self.process_reindex_queue().await {
                    tracing::error!("Reindex batch failed: {}", e);
                }
            }
        });
    }

    // -----------------------------------------------------------------------
    // Reindex queue management
    // -----------------------------------------------------------------------

    /// Mark a file for reindexing.  Duplicate entries are allowed; the
    /// processing step is idempotent so re-indexing the same file twice is
    /// harmless (just slightly wasteful).
    pub async fn mark_for_reindex(&self, file_path: &str) {
        let mut queue = self.reindex_queue.write().await;
        queue.push(PathBuf::from(file_path));
    }

    /// Mark multiple files for reindexing in one lock acquisition.
    pub async fn mark_batch_for_reindex(&self, file_paths: &[&str]) {
        let mut queue = self.reindex_queue.write().await;
        for path in file_paths {
            queue.push(PathBuf::from(path));
        }
    }

    /// Return the number of files currently pending reindexing.
    pub async fn pending_count(&self) -> usize {
        let queue = self.reindex_queue.read().await;
        queue.len()
    }

    /// Process up to `MAX_BATCH_SIZE` files from the reindex queue.
    ///
    /// Files that fail to reindex are logged but not re-queued -- callers
    /// (e.g. file watchers) will re-detect them on the next change event.
    async fn process_reindex_queue(&self) -> Result<()> {
        let batch: Vec<PathBuf> = {
            let mut queue = self.reindex_queue.write().await;
            if queue.is_empty() {
                return Ok(());
            }
            let drain_count = queue.len().min(MAX_BATCH_SIZE);
            queue.drain(..drain_count).collect()
        };

        tracing::info!("Processing reindex queue: {} files", batch.len());

        let mut success_count: usize = 0;
        let mut fail_count: usize = 0;

        for file_path in &batch {
            match self.reindex_file(file_path).await {
                Ok(()) => success_count += 1,
                Err(e) => {
                    tracing::error!("Failed to reindex {:?}: {}", file_path, e);
                    fail_count += 1;
                }
            }
        }

        tracing::info!(
            "Reindex batch complete: {} succeeded, {} failed",
            success_count,
            fail_count,
        );

        Ok(())
    }

    /// Read a file from disk and upsert it into the vector store.
    async fn reindex_file(&self, file_path: &PathBuf) -> Result<()> {
        let content = tokio::fs::read_to_string(file_path)
            .await
            .with_context(|| format!("Reading file for reindex: {:?}", file_path))?;

        let embedding = super::embedder::embed_text(&content);

        let path_str = file_path.to_string_lossy();
        self.vector_db
            .upsert(path_str.as_ref(), &content, embedding)
            .await
            .with_context(|| format!("Upserting into vector DB: {:?}", file_path))?;

        // Update the in-memory freshness cache.
        let now = chrono::Utc::now().timestamp();
        self.mark_fresh(file_path, now).await;

        tracing::debug!("Reindexed {:?}", file_path);
        Ok(())
    }

    // -----------------------------------------------------------------------
    // Freshness queries
    // -----------------------------------------------------------------------

    /// Check whether a file's vector embedding is fresh (i.e., the on-disk
    /// modification time is not newer than the indexed time in LanceDb).
    pub async fn is_fresh(&self, file_path: &Path) -> Result<bool> {
        let disk_mtime = match Self::get_mtime(file_path) {
            Ok(t) => t,
            Err(e) => {
                tracing::trace!(
                    path = %file_path.display(),
                    error = %e,
                    "Cannot stat file, treating as stale"
                );
                return Ok(false);
            }
        };

        // Check the in-memory cache first.
        {
            let cache = self.known_mtimes.read().await;
            if let Some(&cached_mtime) = cache.get(file_path) {
                if cached_mtime >= disk_mtime {
                    return Ok(true);
                }
            }
        }

        // Fall back to checking the vector DB indexed time.
        let path_str = file_path.to_string_lossy().to_string();
        let indexed_at = self.vector_db.get_indexed_time(&path_str).await?;
        if indexed_at >= disk_mtime {
            let mut cache = self.known_mtimes.write().await;
            cache.insert(file_path.to_path_buf(), indexed_at);
            return Ok(true);
        }

        Ok(false)
    }

    /// Compute how stale a file's index entry is relative to its modification
    /// time on disk.
    ///
    /// Returns `Duration::ZERO` if the file was indexed at or after its last
    /// modification.
    pub async fn get_staleness(&self, file_path: &str) -> Result<Duration> {
        let indexed_time = self.vector_db.get_indexed_time(file_path).await?;

        let modified_time = tokio::fs::metadata(file_path)
            .await
            .with_context(|| format!("Checking metadata for staleness: {}", file_path))?
            .modified()
            .context("File system does not support modification times")?;

        let indexed_system_time =
            SystemTime::UNIX_EPOCH + Duration::from_secs(indexed_time as u64);

        Ok(modified_time
            .duration_since(indexed_system_time)
            .unwrap_or(Duration::ZERO))
    }

    /// Check whether a file's index entry is stale beyond the threshold for
    /// its file type.
    pub async fn is_stale(&self, file_path: &str) -> Result<bool> {
        let staleness = self.get_staleness(file_path).await?;
        let ext = std::path::Path::new(file_path)
            .extension()
            .and_then(|e| e.to_str())
            .unwrap_or("");
        let threshold = self.get_max_staleness_for_type(ext).await?;
        Ok(staleness > threshold)
    }

    /// Return the maximum acceptable staleness for a given file extension.
    ///
    /// - Source code (`.rs`, `.py`, `.ts`, `.js`, `.go`, `.java`, `.c`, `.cpp`,
    ///   `.h`): 5 minutes
    /// - Config / lock files (`.toml`, `.json`, `.yaml`, `.yml`, `.lock`):
    ///   10 minutes
    /// - Everything else: 5 minutes (the default)
    pub async fn get_max_staleness_for_type(&self, file_type: &str) -> Result<Duration> {
        let secs = match file_type {
            "rs" | "py" | "ts" | "js" | "go" | "java" | "c" | "cpp" | "h" => {
                DEFAULT_STALENESS_THRESHOLD_SECS
            }
            "toml" | "json" | "yaml" | "yml" | "lock" => 600,
            _ => DEFAULT_STALENESS_THRESHOLD_SECS,
        };
        Ok(Duration::from_secs(secs))
    }

    /// Mark a file as freshly indexed with the given timestamp.
    pub async fn mark_fresh(&self, file_path: &Path, indexed_at: i64) {
        let mut cache = self.known_mtimes.write().await;
        cache.insert(file_path.to_path_buf(), indexed_at);
    }

    /// Mark a file as stale, forcing re-indexing on next access.
    pub async fn mark_stale(&self, file_path: &Path) {
        let mut cache = self.known_mtimes.write().await;
        cache.remove(file_path);
        tracing::debug!(path = %file_path.display(), "Marked stale");
    }

    /// Scan a directory tree and return all file paths that are stale
    /// relative to their vector embeddings.
    pub async fn find_stale_files(&self, dir: &Path) -> Result<Vec<PathBuf>> {
        let mut stale = Vec::new();

        let walker = walkdir::WalkDir::new(dir)
            .follow_links(false)
            .into_iter()
            .filter_map(|e| e.ok())
            .filter(|e| e.file_type().is_file());

        for entry in walker {
            let path = entry.path().to_path_buf();
            match self.is_fresh(&path).await {
                Ok(true) => {}
                Ok(false) => stale.push(path),
                Err(e) => {
                    tracing::warn!(
                        path = %path.display(),
                        error = %e,
                        "Error checking freshness"
                    );
                }
            }
        }

        tracing::debug!(count = stale.len(), dir = %dir.display(), "Stale files found");
        Ok(stale)
    }

    /// Clear all cached freshness data.
    pub async fn clear(&self) {
        let mut cache = self.known_mtimes.write().await;
        cache.clear();
        tracing::debug!("Freshness cache cleared");
    }

    // -----------------------------------------------------------------------
    // Internal helpers
    // -----------------------------------------------------------------------

    /// Get a file's modification time as a Unix timestamp (seconds).
    fn get_mtime(path: &Path) -> Result<i64> {
        let metadata = std::fs::metadata(path)?;
        let mtime = metadata
            .modified()?
            .duration_since(SystemTime::UNIX_EPOCH)
            .map(|d| d.as_secs() as i64)
            .unwrap_or(0);
        Ok(mtime)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_staleness_thresholds() {
        let tmp = std::env::temp_dir().join("furnace_freshness_test");
        let _ = std::fs::remove_dir_all(&tmp);
        let db = Arc::new(LanceDb::new(&tmp).await.expect("create test LanceDb"));
        let fm = FreshnessManager::new(db);

        let rs_dur = fm.get_max_staleness_for_type("rs").await.unwrap();
        assert_eq!(rs_dur, Duration::from_secs(300));

        let toml_dur = fm.get_max_staleness_for_type("toml").await.unwrap();
        assert_eq!(toml_dur, Duration::from_secs(600));

        let unknown_dur = fm.get_max_staleness_for_type("xyz").await.unwrap();
        assert_eq!(unknown_dur, Duration::from_secs(300));

        let _ = std::fs::remove_dir_all(&tmp);
    }

    #[tokio::test]
    async fn test_mark_and_pending_count() {
        let tmp = std::env::temp_dir().join("furnace_freshness_test_pending");
        let _ = std::fs::remove_dir_all(&tmp);
        let db = Arc::new(LanceDb::new(&tmp).await.expect("create test LanceDb"));
        let fm = FreshnessManager::new(db);

        assert_eq!(fm.pending_count().await, 0);

        fm.mark_for_reindex("/some/file.rs").await;
        fm.mark_for_reindex("/another/file.py").await;
        assert_eq!(fm.pending_count().await, 2);

        fm.mark_batch_for_reindex(&["/a.rs", "/b.rs", "/c.rs"]).await;
        assert_eq!(fm.pending_count().await, 5);

        let _ = std::fs::remove_dir_all(&tmp);
    }

    #[tokio::test]
    async fn test_mark_fresh_and_stale() {
        let tmp = std::env::temp_dir().join("furnace_freshness_test_marks");
        let _ = std::fs::remove_dir_all(&tmp);
        let db = Arc::new(LanceDb::new(&tmp).await.expect("create test LanceDb"));
        let fm = FreshnessManager::new(db);

        let path = Path::new("/fake/path.rs");

        // Initially no cache entry.
        {
            let cache = fm.known_mtimes.read().await;
            assert!(!cache.contains_key(path));
        }

        // Mark fresh.
        fm.mark_fresh(path, 12345).await;
        {
            let cache = fm.known_mtimes.read().await;
            assert_eq!(cache.get(path), Some(&12345));
        }

        // Mark stale removes it.
        fm.mark_stale(path).await;
        {
            let cache = fm.known_mtimes.read().await;
            assert!(!cache.contains_key(path));
        }

        let _ = std::fs::remove_dir_all(&tmp);
    }
}
