use std::{collections::HashSet, sync::Arc};

use crate::{schema::INDEX_SCHEMA, storage::SchemaVersion};
use anyhow::Result;
use log::*;
use serde::{Deserialize, Serialize};
use tantivy::schema::Schema;
use tantivy::IndexSettings;
use tokio::fs::{create_dir_all, write};

#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct IndexMetaTracer {
    segments: Vec<InnerSegmentMeta>,
    #[serde(default)]
    index_settings: IndexSettings,
    schema: Schema,
    opstamp: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    payload: Option<String>,

    #[serde(skip)]
    schema_version: Option<SchemaVersion>,
    #[serde(skip)]
    storage_path: Option<String>,
    #[serde(skip)]
    index_name: Option<String>,
    #[serde(skip)]
    start_version: Option<u64>,
    #[serde(skip)]
    end_version: Option<u64>,
}

impl Default for IndexMetaTracer {
    fn default() -> Self {
        Self {
            segments: vec![],
            index_settings: IndexSettings::default(),
            schema: INDEX_SCHEMA.clone(),
            opstamp: 0,
            payload: None,
            schema_version: None,
            storage_path: None,
            index_name: None,
            start_version: None,
            end_version: None,
        }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
struct DeleteMeta {
    num_deleted_docs: u32,
    opstamp: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct InnerSegmentMeta {
    segment_id: String,
    max_doc: u32,
    deletes: Option<DeleteMeta>,
}

impl IndexMetaTracer {
    pub fn new(
        storage_path: &str,
        index_name: &str,
        start_version: u64,
        end_version: u64,
        schema_version: f32,
    ) -> Self {
        let mut tracer = IndexMetaTracer::default();
        tracer.set_meta(
            schema_version,
            storage_path,
            index_name,
            start_version,
            end_version,
        );

        tracer
    }

    pub fn merge(index_metas: Vec<IndexMetaTracer>) -> IndexMetaTracer {
        assert!(!index_metas.is_empty());

        let mut index_meta = index_metas[0].clone();
        for meta in index_metas.iter().skip(1) {
            index_meta.segments.extend(meta.segments.clone());
        }

        index_meta
    }

    pub fn extend(&mut self, other: &Self) {
        self.segments.extend(other.segments.clone());
        self.opstamp += other.opstamp;
    }

    pub async fn load_from_cloud(&mut self, cloud: Arc<dyn cloud::Storage>) -> Result<()> {
        debug!(
            "Loading from cloud in IndexMetaTracer: {}",
            self.meta_cloud_path()?
        );
        let cloud_tracer: Self =
            serde_json::from_str(&cloud.read_file(&self.meta_cloud_path()?).await?)?;
        self.segments = cloud_tracer.segments;
        self.index_settings = cloud_tracer.index_settings;
        self.schema = cloud_tracer.schema;
        self.opstamp = cloud_tracer.opstamp;
        self.payload = cloud_tracer.payload;

        Ok(())
    }

    pub async fn download_prepare(
        &self,
        cloud: Arc<dyn cloud::Storage>,
    ) -> Result<HashSet<String>> {
        let path = self.index_cloud_path()?;
        let keys = cloud.list_files(&path).await?;
        Ok(HashSet::from_iter(keys.iter().cloned()))
    }

    fn index_cloud_path(&self) -> Result<String> {
        debug!(
            "Meta all option params in Tracer: {:?}",
            (
                self.index_name.clone(),
                self.start_version,
                self.end_version,
                self.schema_version.clone()
            )
        );
        match (
            self.index_name.clone(),
            self.start_version,
            self.end_version,
            self.schema_version.clone(),
        ) {
            (Some(index_name), Some(start_version), Some(end_version), Some(schema_version)) => Ok(format!(
                "{}/{}/{}/{}-{}",
                crate::storage::INCREMENTAL_META_PATH,
                index_name,
                schema_version.big(),
                start_version,
                end_version
            )),
            _ => Err(anyhow::anyhow!(
                "index_name, start_version, end_version, schema_version must be set with func set_meta()"
            )),
        }
    }
    fn meta_cloud_path(&self) -> Result<String> {
        match (
            self.index_name.clone(),
            self.start_version,
            self.end_version,
            self.schema_version.clone(),
        ) {
            (Some(index_name), Some(start_version), Some(end_version), Some(schema_version)) => {
                Ok(format!(
                    "{}/{}/{}/{}-{}/meta.json",
                    crate::storage::INCREMENTAL_META_PATH,
                    index_name,
                    schema_version.big(),
                    start_version,
                    end_version
                ))
            }
            _ => Err(anyhow::anyhow!(
                "index_name, start_version, end_version must be set with func set_meta()"
            )),
        }
    }

    #[allow(dead_code)] // for test
    async fn meta_local_path(&self) -> Result<String> {
        let path = match (
            self.storage_path.clone(),
            self.index_name.clone(),
            self.start_version,
            self.end_version,
        ) {
            (Some(storage_path), Some(index_name), Some(start_version), Some(end_version)) => {
                format!(
                    "{}/{}/{}-{}",
                    storage_path, index_name, start_version, end_version
                )
            }
            _ => {
                return Err(anyhow::anyhow!(
                    "index_name, start_version, end_version must be set with func set_meta()"
                ))
            }
        };

        debug!("Creating meta_local_path in IndexMetaTracer: {:?}", path);
        create_dir_all(&path).await?;

        Ok(format!("{}/meta.json", &path))
    }

    pub fn set_meta(
        &mut self,
        schema_version: f32,
        storage_path: &str,
        index_name: &str,
        start_version: u64,
        end_version: u64,
    ) {
        self.schema_version = Some(schema_version.into());
        self.storage_path = Some(String::from(storage_path));
        self.index_name = Some(String::from(index_name));
        self.start_version = Some(start_version);
        self.end_version = Some(end_version);
    }

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

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

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

#[cfg(test)]
mod tests {
    use super::*;
    use crate::schema::{INDEX_SCHEMA, SCHEMA_VERSION};
    use assert_json_diff::*;
    use cloud::Storage;
    use tantivy::Index;

    #[test]
    fn test_serialize_and_deserialize() {
        // Create an index with schema
        let schema = INDEX_SCHEMA.clone();
        let index = Index::create_in_ram(schema);
        let index_meta = index.load_metas().unwrap();
        assert_eq!(index_meta.segments.len(), 0);

        let index_string = serde_json::to_string(&index_meta).unwrap();
        let index_meta_tracer: IndexMetaTracer = serde_json::from_str(&index_string).unwrap();
        assert_eq!(
            index_meta.schema.num_fields(),
            index_meta_tracer.schema.num_fields()
        );

        // Serialize
        let serialized = serde_json::to_string(&index_meta_tracer).unwrap();

        // compare
        let index_json: serde_json::Value = serde_json::from_str(&index_string).unwrap();
        let excerpted: serde_json::Value = serde_json::from_str(&serialized).unwrap();
        assert_json_include!(actual: index_json, expected: &excerpted);
    }

    #[tokio::test]
    async fn test_save_to_local() {
        let index = Index::create_in_ram(INDEX_SCHEMA.clone());
        let index_meta = index.load_metas().unwrap();
        let mut index_meta_tracer = IndexMetaTracer {
            segments: vec![],
            index_settings: index_meta.index_settings.clone(),
            schema: index_meta.schema.clone(),
            opstamp: index_meta.opstamp,
            ..Default::default()
        };

        let temp_dir = "/tmp/fts-tracer-test";

        // Save to local
        index_meta_tracer.set_meta(SCHEMA_VERSION, temp_dir, "qdrant", 0, 100);
        index_meta_tracer.save().await.unwrap();

        // Verify the file exists
        let meta_path_str = index_meta_tracer.meta_local_path().await.unwrap();
        let metadata = tokio::fs::metadata(&meta_path_str).await;
        assert!(metadata.is_ok());
        let metadata = metadata.unwrap();
        assert!(metadata.is_file());
        assert!(metadata.len() > 0);

        // Clean up
        tokio::fs::remove_file(&meta_path_str).await.unwrap();
    }

    #[tokio::test]
    async fn test_load_from_cloud() {
        let index = Index::create_in_ram(INDEX_SCHEMA.clone());
        let index_meta = index.load_metas().unwrap();
        let mut index_meta_tracer = IndexMetaTracer {
            segments: vec![],
            index_settings: index_meta.index_settings.clone(),
            schema: index_meta.schema.clone(),
            opstamp: index_meta.opstamp,
            ..Default::default()
        };

        let temp_dir = "/tmp/fts-tracer-test-cloud";

        // Save to local
        index_meta_tracer.set_meta(SCHEMA_VERSION, temp_dir, "qdrant", 0, 100);
        index_meta_tracer.save().await.unwrap();
        let meta_path_str = index_meta_tracer.meta_local_path().await.unwrap();
        let metadata = tokio::fs::metadata(&meta_path_str).await;
        assert!(metadata.is_ok());

        // upload to cloud
        let s3_storage = Arc::new(cloud::s3::S3Service::default());
        let cloud_path = index_meta_tracer.meta_cloud_path().unwrap();
        s3_storage
            .upload_file(&cloud_path, &meta_path_str)
            .await
            .unwrap();

        // Load from cloud
        let mut loaded_index_meta_tracer = IndexMetaTracer::new(
            temp_dir,
            "qdrant",
            0,
            100,
            index_meta_tracer.schema_version.clone().unwrap().into(),
        );
        loaded_index_meta_tracer
            .load_from_cloud(s3_storage.clone())
            .await
            .unwrap();
        assert_eq!(
            index_meta_tracer.schema.num_fields(),
            loaded_index_meta_tracer.schema.num_fields()
        );
        assert_eq!(
            index_meta_tracer.index_settings,
            loaded_index_meta_tracer.index_settings
        );
        assert_eq!(index_meta_tracer.opstamp, loaded_index_meta_tracer.opstamp);

        // Clean up
        let meta_path_str = index_meta_tracer.meta_local_path().await.unwrap();
        tokio::fs::remove_file(&meta_path_str).await.unwrap();
        s3_storage.delete_file(&cloud_path).await.unwrap();
    }
}