use anyhow::Result;
use colored::Colorize;
use fts_index::{
    index::{apply_deleteds, IndexMeta},
    storage::{
        deletes::Deletes,
        meta::{Incremental, StorageMeta},
    },
};
use lazy_static::lazy_static;
use log::*;
use prettytable::{row, Table};
use std::sync::Arc;

lazy_static! {
    static ref DEFAULT_FTS_SERVING_SERVICE_URL: String = "https://localhost:8443".to_string();
}

pub(crate) async fn search(
    base_url: Option<String>,
    tenant_token: String,
    collection_id: u64,
    index_name: String,
    query: String,
) -> Result<()> {
    println!("Index name: {}", index_name.bold().green());
    println!("Query: {}", query.bold().green());

    let base_url = base_url.unwrap_or(DEFAULT_FTS_SERVING_SERVICE_URL.clone());
    let client = fts_index::client::Client::new(base_url)?;
    let response = client
        .search(
            &index_name,
            collection_id,
            &tenant_token,
            query.into(),
            Some(10),
            None,
        )
        .await?;
    println!(
        "Index version: {}",
        format!("{}", response.index_version).bold().green()
    );
    let mut table = Table::new();
    table.set_titles(row![bFg -> "doc_id", bFgr -> "version", bFg -> "score"]);
    if let Some(hits) = response.hits {
        for hit in hits {
            table.add_row(row![bFg -> hit.doc_id, bFgr -> hit.version, bFg -> hit.score]);
        }
    }
    table.printstd();
    Ok(())
}

pub(crate) async fn reload_index(base_url: Option<String>) -> Result<()> {
    let base_url = base_url.unwrap_or(DEFAULT_FTS_SERVING_SERVICE_URL.clone());
    let client = fts_index::client::Client::new(base_url)?;
    let response = client.reload_index().await?;
    println!(
        "Activated storage version {}",
        format!("{}", response).bold().green()
    );
    Ok(())
}

pub(crate) async fn update_index(base_url: Option<String>, version: u64) -> Result<()> {
    let base_url = base_url.unwrap_or(DEFAULT_FTS_SERVING_SERVICE_URL.clone());
    let client = fts_index::client::Client::new(base_url)?;
    let response = client.update_index(version).await?;
    println!(
        "Activated storage version {}",
        format!("{}", response).bold().green()
    );
    Ok(())
}

pub(crate) async fn indexes(base_url: Option<String>) -> Result<()> {
    let base_url = base_url.unwrap_or(DEFAULT_FTS_SERVING_SERVICE_URL.clone());
    let client = fts_index::client::Client::new(base_url)?;
    let response = client.get_indexes().await?;

    let mut table = Table::new();
    table.set_titles(row![bFg -> "", "version", "index", "start", "end"]);

    for desc in response {
        for index in desc.indexes {
            let current = if desc.current {
                "*".to_string()
            } else {
                "".to_string()
            };
            table.add_row(row![bFg -> current, desc.version, index.0, index.1.0, index.1.1]);
        }
    }
    table.printstd();
    Ok(())
}

pub(crate) async fn versions(base_url: Option<String>, index_name: String) -> Result<()> {
    let base_url = base_url.unwrap_or(DEFAULT_FTS_SERVING_SERVICE_URL.clone());
    let client = fts_index::client::Client::new(base_url)?;
    let response = client.get_versions(&index_name).await?;
    println!("Index name: {}", index_name.bold().green());
    println!(
        "Index start version: {}",
        format!("{}", response.0).bold().green()
    );
    println!(
        "Index end version: {}",
        format!("{}", response.1).bold().green()
    );
    Ok(())
}

pub const REMAINS_META_VERSION_COUNT: u64 = 50;

pub(crate) async fn compact_index(storage_path: String, index_name: String) -> Result<()> {
    let s3service = Arc::new(cloud::s3::S3Service::default());
    let latest_storage_meta = match StorageMeta::load_latest_cloud(s3service.clone()).await? {
        Some(storage_meta) => storage_meta,
        None => {
            let msg = "must create schema first: ctl create_index <index_name>";
            warn!("{}", msg);
            return Err(anyhow::anyhow!(msg));
        }
    };

    if !latest_storage_meta.indexes.contains_key(&index_name) {
        let msg = format!("Index {} not found", index_name);
        warn!("{}", msg);
        return Err(anyhow::anyhow!(msg));
    }

    info!("Start to compact index {}", index_name);
    latest_storage_meta
        .download_all_incrementals_and_merge(s3service.clone(), &storage_path)
        .await?;

    apply_deleteds(&storage_path, s3service.clone(), &latest_storage_meta).await?;

    let index_meta =
        if let Some(meta) = fts_index::index::compact_index(&storage_path, &index_name).await? {
            meta
        } else {
            let msg = format!("Index {} not found", index_name);
            warn!("{}", msg);
            return Err(anyhow::anyhow!(msg));
        };

    // Save incremental meta and upload to cloud
    let index = latest_storage_meta.indexes.get(&index_name).unwrap();
    let mut incremental_meta = index_meta.incremental_meta();
    incremental_meta.save(&storage_path).await?;
    incremental_meta
        .upload(&storage_path, s3service.clone())
        .await?;

    // Update storage meta
    let mut new_index = index.clone();
    new_index.incrementals = vec![Incremental::from(&incremental_meta)];
    let mut new_storage_meta = latest_storage_meta.clone();
    new_storage_meta
        .indexes
        .insert(index_name.clone(), new_index);
    new_storage_meta.version += 1;
    new_storage_meta.upload(s3service.clone()).await?;

    // Reset deletes
    Deletes::reset_deletes(s3service.clone()).await?;

    // Activate new storage meta
    StorageMeta::activate(s3service.clone(), new_storage_meta.version).await?;

    // Expired old storage meta
    let versions = StorageMeta::get_cloud_versions(s3service.clone()).await?;
    for version in versions
        .iter()
        .skip(REMAINS_META_VERSION_COUNT as usize * 2)
    {
        Deletes::gc(s3service.clone(), *version).await?;
    }

    print_index_meta(&index_name, &index_meta);

    Ok(())
}

fn print_index_meta(index_name: impl AsRef<str>, index_meta: &IndexMeta) {
    println!(
        "Index {} start-version: {}, end-version: {}",
        index_name.as_ref().bold().green(),
        format!("{}", index_meta.start_version()).bold().green(),
        format!("{}", index_meta.end_version()).bold().green(),
    );
}