use crate::storage::SchemaVersion;
use anyhow::Result;
use log::*;
use serde::{Deserialize, Serialize};
use std::{collections::HashSet, path::Path, sync::Arc};
use tantivy::Index;
use tokio::fs::{create_dir_all, read_dir, write};

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct IncrementalIndexMeta {
    segments: u64,
    valid: u64,
    deleted: u64,
    schema_version: SchemaVersion,

    #[serde(skip)]
    index_name: Option<String>,
    #[serde(skip)]
    start_version: Option<u64>,
    #[serde(skip)]
    end_version: Option<u64>,
    #[serde(skip)]
    preparing_keys: HashSet<String>,
}

impl Default for IncrementalIndexMeta {
    fn default() -> Self {
        Self {
            segments: 0,
            valid: 0,
            deleted: 0,
            schema_version: 1.0.into(),
            index_name: None,
            start_version: None,
            end_version: None,
            preparing_keys: HashSet::new(),
        }
    }
}

impl IncrementalIndexMeta {
    pub fn start_version(&self) -> u64 {
        self.start_version.unwrap_or_default()
    }

    pub fn end_version(&self) -> u64 {
        self.end_version.unwrap_or_default()
    }

    pub fn schema_version(&self) -> &SchemaVersion {
        &self.schema_version
    }

    pub async fn load_from_cloud<C>(&mut self, cloud: &C) -> Result<Self>
    where
        C: cloud::Storage,
    {
        Ok(serde_json::from_str(
            &cloud.read_file(&self.meta_cloud_path()?).await?,
        )?)
    }

    pub fn set_meta(
        &mut self,
        index_name: String,
        start_version: u64,
        end_version: u64,
        schema_version: SchemaVersion,
    ) {
        debug!(
            "IncrementalIndexMeta set_meta: index_name: {}, start_version: {}, end_version: {}, schema_version: {:?}",
            index_name, start_version, end_version, schema_version
        );
        self.index_name = Some(index_name);
        self.start_version = Some(start_version);
        self.end_version = Some(end_version);
        self.schema_version = schema_version;
    }

    pub fn extend_preparing_keys(&mut self, keys: HashSet<String>) {
        self.preparing_keys.extend(keys);
    }

    #[allow(dead_code)] // for test
    fn meta_cloud_path(&self) -> Result<String> {
        match (
            self.index_name.clone(),
            self.start_version,
            self.end_version,
        ) {
            (Some(index_name), Some(start_version), Some(end_version)) => Ok(format!(
                "{}/{}/{}/meta-{}-{}.json",
                crate::storage::INCREMENTAL_META_PATH,
                index_name,
                self.schema_version.big(),
                start_version,
                end_version
            )),
            _ => Err(anyhow::anyhow!(
                "index_name, start_version, end_version must be set with func set_meta()"
            )),
        }
    }

    pub fn index_path(&self, storage_path: &str) -> Result<String> {
        match (
            self.index_name.clone(),
            self.start_version,
            self.end_version,
        ) {
            (Some(index_name), Some(start_version), Some(end_version)) => Ok(format!(
                "{}/{}/{}-{}",
                storage_path, index_name, start_version, end_version
            )),
            _ => Err(anyhow::anyhow!(
                "index_name must be set with func set_meta()"
            )),
        }
    }

    fn index_cloud_path(&self) -> Result<String> {
        match (
            self.index_name.clone(),
            self.start_version,
            self.end_version,
        ) {
            (Some(index_name), Some(start_version), Some(end_version)) => Ok(format!(
                "{}/{}/{}/{}-{}",
                crate::storage::INCREMENTAL_META_PATH,
                index_name,
                self.schema_version.big(),
                start_version,
                end_version
            )),
            _ => Err(anyhow::anyhow!(
                "index_name, start_version, end_version must be set with func set_meta()"
            )),
        }
    }

    async fn meta_path(&self, storage_path: &str) -> Result<String> {
        match (
            self.index_name.clone(),
            self.start_version,
            self.end_version,
        ) {
            (Some(index_name), Some(start_version), Some(end_version)) => {
                let index_path = format!("{}/{}", storage_path, index_name);
                debug!(
                    "Creating meta_path in IncrementalIndexMeta: {:?}",
                    index_path
                );
                create_dir_all(&index_path).await?;
                Ok(format!(
                    "{}/meta-{}-{}.json",
                    index_path, start_version, end_version
                ))
            }
            _ => Err(anyhow::anyhow!(
                "index_name, start_version, end_version must be set with func set_meta()"
            )),
        }
    }

    pub async fn download_prepared_files(
        &self,
        cloud: Arc<dyn cloud::Storage>,
        storage_path: &str,
    ) -> Result<()> {
        let mut managed = HashSet::new();
        for key in &self.preparing_keys {
            let file_name = if let Some(name) = key.split('/').last() {
                name.to_string()
            } else {
                warn!("Invalid key in download_preared_files: {}", key);
                continue;
            };
            if file_name == ".managed.json" || file_name == "meta.json" {
                continue;
            }
            let path = format!("{}/{}", self.index_path(storage_path).unwrap(), &file_name);
            debug!("Downloading file: {} to {}", key, path);
            cloud.download_file(key, &path).await?;
            if file_name == ".tantivy-writer.lock" || file_name == ".tantivy-meta.lock" {
                continue;
            }
            managed.insert(file_name);
        }

        // write managed to index_path/.managed.json
        let managed_path = format!("{}/.managed.json", self.index_path(storage_path)?);
        debug!(
            "Writing {:?} managed files: {:?}",
            &self.index_name, managed
        );
        write(managed_path, serde_json::to_string(&managed)?).await?;

        Ok(())
    }

    pub async fn save(&mut self, storage_path: &str) -> Result<()> {
        let meta_path = self.meta_path(storage_path).await?;
        debug!("IncrementalIndexMeta save to local path: {:?}", meta_path);

        self.segments = 0;
        self.valid = 0;
        self.deleted = 0;

        let index = Index::open_in_dir(self.index_path(storage_path)?)?;
        for m in index.load_metas()?.segments {
            debug!("Saving segment: {:?}", m);
            self.segments += 1;
            self.valid += m.num_docs() as u64;
            self.deleted += m.num_deleted_docs() as u64;
        }

        // if the file already exists
        if tokio::fs::metadata(&meta_path).await.is_ok() {
            warn!("'{}' write conflict, need check why it exists", meta_path);
        }

        write(meta_path, serde_json::to_string(self)?).await?;
        Ok(())
    }

    pub async fn upload(&self, storage_path: &str, cloud: Arc<dyn cloud::Storage>) -> Result<()> {
        let index_path = self.index_path(storage_path)?;
        let mut files = read_dir(Path::new(&index_path)).await?;
        while let Some(entry) = files.next_entry().await? {
            if entry.file_type().await?.is_dir() {
                continue;
            }
            if let Some(file_name) = entry.file_name().to_str() {
                let file_path = format!("{}/{}", index_path, file_name);
                let key = format!("{}/{}", self.index_cloud_path()?, file_name);
                info!("Uploading file: {} to {}", file_path, key);
                // if entry.metadata().await?.len() < cloud::MIN_CHUNK_SIZE as u64
                //     || file_name.starts_with(".")
                // {
                cloud.upload_file(&key, &file_path).await?;
                //     continue;
                // }
                // cloud.upload_large_file(&key, &file_path).await?;
            }
        }

        let path = self.meta_path(storage_path).await?;
        debug!("Uploading file: {} to {}", path, self.meta_cloud_path()?);
        cloud.upload_file(&self.meta_cloud_path()?, &path).await?;
        Ok(())
    }

    /// Dir before meta, never concurrently: the meta file is the only handle `local_gc` walks,
    /// so dropping it while the dir survives orphans the dir forever.
    pub(crate) async fn delete_local_files(&self, storage_path: &str) -> Result<()> {
        let index_path = self.index_path(storage_path)?;
        let meta_path = self.meta_path(storage_path).await?;
        debug!("Deleting local files: {} and {}", index_path, meta_path);

        if let Err(e) = tokio::fs::remove_dir_all(&index_path).await {
            if e.kind() != std::io::ErrorKind::NotFound {
                return Err(anyhow::anyhow!("remove index dir '{}': {}", index_path, e));
            }
        }

        crate::index::remove_file_ignore_error(Path::new(&meta_path)).await;
        Ok(())
    }

    pub async fn delete(&self, cloud: Arc<dyn cloud::Storage>) -> Result<()> {
        let meta_key = self.meta_cloud_path()?;
        let index_key = self.index_cloud_path()?;
        info!("Deleting files: {} and {}", meta_key, index_key);

        cloud.delete_file(&meta_key).await?;
        cloud.delete_file(&index_key).await?;

        Ok(())
    }
}

/// Removes `{start}-{end}` dirs under `{storage_path}/{index_name}` with no
/// `meta-{start}-{end}.json` beside them, and returns how many it removed. `local_gc` walks
/// meta files, so a dir that lost its meta is invisible to it; this scans the dir instead.
///
/// `keep` is the range being served — kept even without a meta file. Never run this while a
/// download is in flight: `download_all_incrementals_and_merge` creates the dir before its
/// meta file, so a sweep in that window would delete the download.
pub async fn gc_orphan_index_dirs(
    storage_path: &str,
    index_name: &str,
    keep: (u64, u64),
) -> Result<usize> {
    let index_dir = format!("{}/{}", storage_path, index_name);
    let mut entries = match read_dir(&index_dir).await {
        Ok(e) => e,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(e.into()),
    };

    let mut dirs: Vec<(u64, u64)> = vec![];
    let mut metas: HashSet<(u64, u64)> = HashSet::new();
    while let Some(entry) = entries.next_entry().await? {
        let name = entry.file_name().to_string_lossy().to_string();
        if entry.file_type().await?.is_dir() {
            if let Some((start, end)) = name.split_once('-') {
                if let (Ok(start), Ok(end)) = (start.parse(), end.parse()) {
                    dirs.push((start, end));
                }
            }
            continue;
        }
        if let Some(caps) = crate::index::INDEX_VERSION_META_FILE_PATTERN.captures(&name) {
            metas.insert((caps["start"].parse::<u64>()?, caps["end"].parse::<u64>()?));
        }
    }

    let mut removed = 0;
    for (start, end) in dirs {
        if (start, end) == keep || metas.contains(&(start, end)) {
            continue;
        }
        let path = format!("{}/{}-{}", index_dir, start, end);
        match tokio::fs::remove_dir_all(&path).await {
            Ok(_) => {
                info!("Removed orphan index dir: {}", path);
                removed += 1;
            }
            Err(e) => warn!("Failed to remove orphan index dir '{}': {}", path, e),
        }
    }

    Ok(removed)
}

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

    #[tokio::test]
    async fn test_set_meta() {
        let mut meta = IncrementalIndexMeta::default();
        meta.set_meta("test_index".to_string(), 1, 2, SchemaVersion::from(1.0));

        assert_eq!(meta.index_name, Some("test_index".to_string()));
        assert_eq!(meta.start_version, Some(1));
        assert_eq!(meta.end_version, Some(2));
        assert_eq!(meta.schema_version, SchemaVersion::from(1.0));
    }

    #[tokio::test]
    async fn test_extend_preparing_keys() {
        let mut meta = IncrementalIndexMeta::default();
        let mut keys = HashSet::new();
        keys.insert("key1".to_string());
        keys.insert("key2".to_string());

        meta.extend_preparing_keys(keys.clone());

        assert_eq!(meta.preparing_keys, keys);
    }

    // A regular file where the index directory belongs makes `remove_dir_all` fail with
    // ENOTDIR — a portable stand-in for the real production failure (a PVC that refuses
    // to remove a directory whose files are still mapped).
    #[tokio::test]
    async fn test_delete_local_files_keeps_meta_when_dir_removal_fails() {
        let storage_path = "/tmp/fts-delete-local-files-fail-test";
        let _ = tokio::fs::remove_dir_all(storage_path).await;

        let mut meta = IncrementalIndexMeta::default();
        meta.set_meta("idx".to_string(), 0, 100, SchemaVersion::from(1.0));

        // meta_path() creates {storage_path}/idx for us
        let meta_path = meta.meta_path(storage_path).await.unwrap();
        write(&meta_path, "{}").await.unwrap();

        let index_path = meta.index_path(storage_path).unwrap();
        write(&index_path, "not a dir").await.unwrap();

        assert!(meta.delete_local_files(storage_path).await.is_err());
        assert!(
            Path::new(&meta_path).exists(),
            "meta file must survive so the dir stays reachable by a later GC"
        );

        let _ = tokio::fs::remove_dir_all(storage_path).await;
    }

    #[tokio::test]
    async fn test_gc_orphan_index_dirs() {
        let storage_path = "/tmp/fts-gc-orphan-test";
        let _ = tokio::fs::remove_dir_all(storage_path).await;
        let index_dir = format!("{}/idx", storage_path);

        // orphan: a range dir with no meta file next to it
        create_dir_all(format!("{}/0-100", index_dir))
            .await
            .unwrap();
        write(format!("{}/0-100/seg.idx", index_dir), "x")
            .await
            .unwrap();
        // reachable: a range dir that still has its meta file
        create_dir_all(format!("{}/0-200", index_dir))
            .await
            .unwrap();
        write(format!("{}/meta-0-200.json", index_dir), "{}")
            .await
            .unwrap();
        // currently serving: no meta file yet, must still be kept
        create_dir_all(format!("{}/0-300", index_dir))
            .await
            .unwrap();
        // unrelated name: never a sweep target
        create_dir_all(format!("{}/compacting", index_dir))
            .await
            .unwrap();

        let removed = gc_orphan_index_dirs(storage_path, "idx", (0, 300))
            .await
            .unwrap();

        assert_eq!(removed, 1);
        assert!(!Path::new(&format!("{}/0-100", index_dir)).exists());
        assert!(Path::new(&format!("{}/0-200", index_dir)).exists());
        assert!(Path::new(&format!("{}/0-300", index_dir)).exists());
        assert!(Path::new(&format!("{}/compacting", index_dir)).exists());

        let _ = tokio::fs::remove_dir_all(storage_path).await;

        // a missing index dir is not an error
        let removed = gc_orphan_index_dirs(storage_path, "idx", (0, 1))
            .await
            .unwrap();
        assert_eq!(removed, 0);
    }
}