use crate::storage::incremental::*;
use crate::storage::index_tracer::IndexMetaTracer;
use crate::storage::SchemaVersion;
use crate::storage::*;
use anyhow::Result;
use core::panic;
use lazy_static::lazy_static;
use log::*;
use regex::Regex;
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, fs::create_dir_all, sync::Arc};
use tokio::fs::{read_dir, read_to_string};

const META_VERSION: &str = "meta_version";
const CLOUD_VERSION_FILE: &str = ".active_version";

lazy_static! {
    static ref STORAGE_META_FILE_PATTERN: Regex =
        Regex::new(r#"^meta-(?<meta_version>\d+).json$"#).unwrap();
    static ref STORAGE_META_KEY_PATTERN: Regex = Regex::new(&format!(
        r#"^{}/meta-(?<meta_version>\d+).json$"#,
        STORAGE_META_PATH
    ))
    .unwrap();
    static ref CLOUD_VERSION_KEY: String = format!("{}/{}", STORAGE_META_PATH, CLOUD_VERSION_FILE);
}

#[derive(Clone, Default, Debug, Serialize, Deserialize)]
pub struct StorageMeta {
    pub version: u64,
    pub indexes: HashMap<String, Index>,
}

impl PartialEq for StorageMeta {
    fn eq(&self, other: &Self) -> bool {
        self.version == other.version
    }
}

impl StorageMeta {
    pub async fn load_from_cloud_by_version(
        cloud: Arc<dyn cloud::Storage>,
        meta_version: u64,
    ) -> Result<Self> {
        let full_key = format!("{}/meta-{}.json", STORAGE_META_PATH, meta_version);
        debug!("Loading storage meta from cloud: {}", full_key);
        let meta_str = match cloud.read_file(&full_key).await {
            Ok(v) => v,
            Err(e) => {
                error!("Failed to read cloud meta: {:?}", e);
                return Err(e);
            }
        };

        let meta = match serde_json::from_str::<Self>(&meta_str) {
            Ok(v) => v,
            Err(e) => {
                error!("Failed to parse cloud meta: {:?}", e);
                return Err(e.into());
            }
        };

        Ok(meta)
    }

    pub async fn load_latest_cloud(cloud: Arc<dyn cloud::Storage>) -> Result<Option<Self>> {
        let versions = Self::get_cloud_versions(cloud.clone()).await?;
        if versions.is_empty() {
            return Ok(None);
        }

        let latest_version = versions.first().unwrap_or(&0);

        Ok(Some(
            Self::load_from_cloud_by_version(cloud.clone(), *latest_version).await?,
        ))
    }

    pub async fn get_cloud_version(cloud: Arc<dyn cloud::Storage>) -> Result<u64> {
        let version = match cloud.read_file(&CLOUD_VERSION_KEY).await {
            Ok(v) => v,
            Err(e) => {
                warn!("Failed to read cloud version: {:?}", e);
                return Err(e);
            }
        };
        let version = match version.trim().parse::<u64>() {
            Ok(v) => v,
            Err(e) => {
                warn!("Failed to parse cloud version: {:?}", e);
                return Err(e.into());
            }
        };

        Ok(version)
    }

    pub async fn activate(cloud: Arc<dyn cloud::Storage>, version: u64) -> Result<()> {
        cloud
            .upload_string(&CLOUD_VERSION_KEY, &version.to_string())
            .await?;
        debug!("Activated cloud version: {}", version);
        Ok(())
    }

    pub async fn get_cloud_versions(cloud: Arc<dyn cloud::Storage>) -> Result<Vec<u64>> {
        let keys = match cloud.list_files(STORAGE_META_PATH).await {
            Ok(keys) => keys,
            Err(e) => panic!("Failed to list files: {:?}", e),
        };

        let mut meta_versions: Vec<u64> = vec![];
        for key in keys {
            if let Some(captures) = STORAGE_META_KEY_PATTERN.captures(&key) {
                let version = captures
                    .name(META_VERSION)
                    .unwrap()
                    .as_str()
                    .parse::<u64>()?;
                meta_versions.push(version);
            }
        }
        meta_versions.sort_by(|a, b| b.cmp(a)); // sort in descending order
        Ok(meta_versions)
    }

    fn full_key(&self) -> String {
        format!("{}/meta-{}.json", STORAGE_META_PATH, self.version)
    }

    pub async fn upload(&self, cloud: Arc<dyn cloud::Storage>) -> Result<()> {
        info!("Uploading storage meta version: {}", self.version);
        cloud
            .upload_string(&self.full_key(), &serde_json::to_string(self)?)
            .await?;
        Ok(())
    }

    pub async fn delete_from_cloud(&self, cloud: Arc<dyn cloud::Storage>) -> Result<()> {
        info!("Deleting storage meta version: {}", self.version);
        cloud.delete_file(&self.full_key()).await?;
        Ok(())
    }

    pub fn first_start_version(&self, index_name: &str) -> u64 {
        if let Some(index) = self.indexes.get(index_name) {
            if let Some(first) = index.incrementals.first() {
                return first.start_version;
            }
        }

        0
    }

    pub fn latest_end_version(&self, index_name: &str) -> u64 {
        let mut last = 0;
        if let Some(index) = self.indexes.get(index_name) {
            for i in index.incrementals.iter() {
                last = std::cmp::max(last, i.end_version);
            }
        }

        last
    }

    pub fn save(&self, storage_path: &str) -> Result<()> {
        create_dir_all(storage_path)?;
        let full_path = format!("{}/meta-{}.json", storage_path, self.version);
        if std::path::Path::new(&full_path).exists() {
            return Err(anyhow::anyhow!("File already exists: {}", full_path));
        }
        std::fs::write(full_path, serde_json::to_string(self)?)?;
        Ok(())
    }

    pub fn sum_incrementals(&self) -> u64 {
        self.indexes
            .values()
            .map(|index| index.incrementals.len() as u64)
            .sum()
    }

    pub async fn load_latest_local(storage_path: &str) -> Result<Self> {
        let mut files = read_dir(storage_path).await?;
        let mut latest_version = 0;
        while let Some(file) = files.next_entry().await? {
            if file.file_type().await?.is_dir() {
                continue;
            }

            if let Some(captures) =
                STORAGE_META_FILE_PATTERN.captures(&file.file_name().to_string_lossy())
            {
                let version = captures
                    .name(META_VERSION)
                    .unwrap()
                    .as_str()
                    .parse::<u64>()?;
                latest_version = std::cmp::max(latest_version, version);
            }
        }

        let full_path = format!("{}/meta-{}.json", storage_path, latest_version);
        Ok(serde_json::from_str(&read_to_string(full_path).await?)?)
    }

    pub async fn load_local_by_version(storage_path: &str, version: u64) -> Result<Self> {
        let full_path = format!("{}/meta-{}.json", storage_path, version);
        Ok(serde_json::from_str(&read_to_string(full_path).await?)?)
    }

    pub async fn get_local_versions(storage_path: &str) -> Result<Vec<u64>> {
        let mut files = if let Ok(fs) = read_dir(&storage_path).await {
            fs
        } else {
            return Ok(vec![]);
        };
        let mut meta_versions: Vec<u64> = vec![];
        while let Some(file) = files.next_entry().await? {
            if file.file_type().await?.is_dir() {
                continue;
            }
            if let Some(file_name) = file.file_name().to_str() {
                if let Some(caps) = STORAGE_META_FILE_PATTERN.captures(file_name) {
                    if let Some(version) = caps.name(META_VERSION) {
                        meta_versions.push(version.as_str().parse::<u64>()?);
                    }
                }
            }
        }
        meta_versions.sort_by(|a, b| b.cmp(a)); // sort in descending order
        Ok(meta_versions)
    }

    pub async fn local_gc(&self, storage_path: &str) -> Result<()> {
        let mut failed: Vec<String> = vec![];
        for (index_name, index) in &self.indexes {
            let mut inc_meta = IncrementalIndexMeta::default();
            let (start_version, end_version) = (
                self.first_start_version(index_name),
                self.latest_end_version(index_name),
            );
            inc_meta.set_meta(
                index_name.into(),
                start_version,
                end_version,
                index.schema_version.clone(),
            );
            match inc_meta.delete_local_files(storage_path).await {
                Ok(_) => debug!(
                    "Deleted local inclurmental {}-{} for index: {}",
                    start_version, end_version, index_name
                ),
                Err(e) => {
                    warn!(
                        "Failed to delete local inclurmental {}-{} for index: {} : {:?}",
                        start_version, end_version, index_name, e
                    );
                    failed.push(index_name.clone());
                }
            };
        }

        // Keep this meta file if any dir survived — it is the only pointer to those dirs, so
        // dropping it leaks them. The next GC round retries.
        if !failed.is_empty() {
            return Err(anyhow::anyhow!(
                "kept storage meta {}: index dirs still on disk for {:?}",
                self.version,
                failed
            ));
        }

        match tokio::fs::remove_file(format!("{}/meta-{}.json", storage_path, self.version)).await {
            Ok(_) => debug!("Deleted meta file: {}", self.version),
            Err(e) => warn!("Failed to delete meta file: {:?}", e),
        };

        Ok(())
    }

    pub async fn download_all_incrementals_and_merge(
        &self,
        cloud: Arc<dyn cloud::Storage>,
        storage_path: &str,
    ) -> Result<()> {
        for (index_name, index) in self.indexes.iter() {
            let mut merging_index_tracer = IndexMetaTracer::new(
                storage_path,
                index_name,
                self.first_start_version(index_name),
                self.latest_end_version(index_name),
                index.schema_version.clone().into(),
            );
            let mut incremental_meta = IncrementalIndexMeta::default();
            incremental_meta.set_meta(
                index_name.clone(),
                self.first_start_version(index_name),
                self.latest_end_version(index_name),
                index.schema_version.clone(),
            );

            for incremental in index.incrementals.iter() {
                let mut index_meta_tracer = IndexMetaTracer::new(
                    storage_path,
                    index_name,
                    incremental.start_version,
                    incremental.end_version,
                    index.schema_version.clone().into(),
                );
                index_meta_tracer.load_from_cloud(cloud.clone()).await?;

                let keys = index_meta_tracer.download_prepare(cloud.clone()).await?;
                incremental_meta.extend_preparing_keys(keys);
                merging_index_tracer.extend(&index_meta_tracer);
            }
            debug!("Merging index-tracer: {:?}", &merging_index_tracer);
            merging_index_tracer.save().await?;
            incremental_meta
                .download_prepared_files(cloud.clone(), storage_path)
                .await?;
            incremental_meta.save(storage_path).await?;
        }

        self.save(storage_path)?;
        Ok(())
    }

    pub async fn lock_cloud_version(
        &self,
        cloud: Arc<dyn cloud::Storage>,
        lock: bool,
    ) -> Result<()> {
        if lock {
            let key = format!("{}/storage-meta-{}.0.lock", STORAGE_META_PATH, self.version);
            cloud.upload_string(&key, "").await?;
        } else {
            let key = format!("{}/storage-meta-{}.0.lock", STORAGE_META_PATH, self.version);
            cloud.delete_file(&key).await?;
        }

        Ok(())
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Index {
    pub schema_version: SchemaVersion,
    pub incrementals: Vec<Incremental>,
}

impl Index {
    pub fn new(schema_version: SchemaVersion, incrementals: Vec<Incremental>) -> Self {
        Self {
            schema_version,
            incrementals,
        }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Incremental {
    pub start_version: u64,
    pub end_version: u64,
}

impl Incremental {
    pub fn new(start_version: u64, end_version: u64) -> Self {
        Self {
            start_version,
            end_version,
        }
    }
}

impl From<&IncrementalIndexMeta> for Incremental {
    fn from(meta: &IncrementalIndexMeta) -> Self {
        Self {
            start_version: meta.start_version(),
            end_version: meta.end_version(),
        }
    }
}

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

    #[tokio::test]
    async fn test_load_latest_from_cloud() {
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let storage_meta = StorageMeta::load_latest_cloud(s3_storage).await;
        assert!(storage_meta.is_ok());
    }

    #[tokio::test]
    async fn test_upload() {
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let mut storage_meta = StorageMeta::load_latest_cloud(s3_storage.clone())
            .await
            .unwrap()
            .unwrap_or_default();
        storage_meta.version = storage_meta.version + 1;

        // update
        let result = storage_meta.upload(s3_storage.clone()).await;
        assert!(result.is_ok());

        //clean up
        let result = storage_meta.delete_from_cloud(s3_storage).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_load_from_cloud_by_version() {
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let version = StorageMeta::load_latest_cloud(s3_storage.clone())
            .await
            .unwrap()
            .unwrap_or_default()
            .version; // Assuming version 1 exists in the cloud storage
        let storage_meta = StorageMeta::load_from_cloud_by_version(s3_storage, version).await;
        assert!(storage_meta.is_ok());
        assert_eq!(storage_meta.unwrap().version, version);
    }

    #[tokio::test]
    async fn test_get_cloud_versions() {
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let versions = StorageMeta::get_cloud_versions(s3_storage).await;
        assert!(versions.is_ok());
        assert!(!versions.unwrap().is_empty());
    }

    #[tokio::test]
    async fn test_save_and_load_local() {
        let storage_path = "/tmp/test_storage_meta";
        let storage_meta =
            StorageMeta::load_latest_cloud(Arc::new(cloud::s3::S3Service::default()))
                .await
                .unwrap()
                .unwrap();

        // Save to local
        let save_result = storage_meta.save(storage_path);
        assert!(save_result.is_ok());

        // Load from local
        let load_result =
            StorageMeta::load_local_by_version(storage_path, storage_meta.version).await;
        assert!(load_result.is_ok());
        assert_eq!(load_result.unwrap().version, storage_meta.version);

        // Clean up
        std::fs::remove_file(format!(
            "{}/meta-{}.json",
            storage_path, storage_meta.version
        ))
        .unwrap();
    }

    #[tokio::test]
    async fn test_lock_and_unlock_cloud_version() {
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let storage_meta = StorageMeta {
            version: 1,
            indexes: HashMap::new(),
        };

        // Lock version
        let lock_result = storage_meta
            .lock_cloud_version(s3_storage.clone(), true)
            .await;
        assert!(lock_result.is_ok());

        // Unlock version
        let unlock_result = storage_meta.lock_cloud_version(s3_storage, false).await;
        assert!(unlock_result.is_ok());
    }

    #[tokio::test]
    async fn test_download_all_incrementals_and_merge() {
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let storage_path = "/tmp/test_storage_meta_download";
        let storage_meta = StorageMeta::load_latest_cloud(s3_storage.clone())
            .await
            .unwrap()
            .unwrap();

        let result = storage_meta
            .download_all_incrementals_and_merge(s3_storage, storage_path)
            .await;
        assert!(result.is_ok());

        // Clean up
        std::fs::remove_dir_all(storage_path).unwrap();
    }

    #[tokio::test]
    async fn test_local_gc_keeps_meta_when_index_dir_survives() {
        let storage_path = "/tmp/fts-storage-meta-gc-fail-test";
        let _ = tokio::fs::remove_dir_all(storage_path).await;
        std::fs::create_dir_all(format!("{}/idx", storage_path)).unwrap();

        let storage_meta = StorageMeta {
            version: 7,
            indexes: HashMap::from([(
                "idx".to_string(),
                Index::new(1.0f32.into(), vec![Incremental::new(0, 100)]),
            )]),
        };
        storage_meta.save(storage_path).unwrap();

        // regular file where the index dir belongs -> remove_dir_all fails with ENOTDIR
        std::fs::write(format!("{}/idx/0-100", storage_path), "not a dir").unwrap();
        std::fs::write(format!("{}/idx/meta-0-100.json", storage_path), "{}").unwrap();

        assert!(storage_meta.local_gc(storage_path).await.is_err());
        assert!(
            std::path::Path::new(&format!("{}/meta-7.json", storage_path)).exists(),
            "root storage meta must survive so the next GC retries this version"
        );
        assert!(
            std::path::Path::new(&format!("{}/idx/meta-0-100.json", storage_path)).exists(),
            "index meta must survive alongside the dir it points at"
        );

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

    #[tokio::test]
    async fn test_local_gc_removes_dir_and_metas() {
        let storage_path = "/tmp/fts-storage-meta-gc-ok-test";
        let _ = tokio::fs::remove_dir_all(storage_path).await;
        std::fs::create_dir_all(format!("{}/idx/0-100", storage_path)).unwrap();
        std::fs::write(format!("{}/idx/0-100/seg.idx", storage_path), "x").unwrap();
        std::fs::write(format!("{}/idx/meta-0-100.json", storage_path), "{}").unwrap();

        let storage_meta = StorageMeta {
            version: 7,
            indexes: HashMap::from([(
                "idx".to_string(),
                Index::new(1.0f32.into(), vec![Incremental::new(0, 100)]),
            )]),
        };
        storage_meta.save(storage_path).unwrap();

        assert!(storage_meta.local_gc(storage_path).await.is_ok());
        assert!(!std::path::Path::new(&format!("{}/idx/0-100", storage_path)).exists());
        assert!(!std::path::Path::new(&format!("{}/idx/meta-0-100.json", storage_path)).exists());
        assert!(!std::path::Path::new(&format!("{}/meta-7.json", storage_path)).exists());

        // a second round must not error just because everything is already gone
        assert!(storage_meta.local_gc(storage_path).await.is_ok());

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