pub(crate) mod jwt;
pub(crate) mod pager;
pub(crate) mod sink;
pub(crate) mod source;

use super::pipeline::source::*;
use anyhow::Result;
use collection::CollectionSource;
use colored::Colorize;
use fts_index::storage::deletes::Deletes;
use fts_index::storage::meta::Index;
use fts_index::{
    schema::SCHEMA_VERSION,
    storage::meta::{Incremental, StorageMeta},
};
use log::*;
use model::meta::*;
use model::report::PipelineReport;
use sink::{Sink, SinkReport};
use sqlx::MySqlPool;
use std::sync::Arc;
use tokio::time::Instant;
use tokio::{join, try_join};

pub(crate) async fn index_updated_collections(
    dsn: String,
    versions: (Option<u64>, Option<u64>),
    memory_budget: Option<usize>,
    storage_path: String,
    index_name: String,
    concurrency: usize,
    whole_deleted: bool,
) -> Result<u64> {
    let instant = Instant::now();
    let s3service = Arc::new(cloud::s3::S3Service::default());
    let mut 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));
        }
    };

    let db = Arc::new(MySqlPool::connect(&dsn).await?);

    let start_version = match versions.0 {
        Some(s) => s,
        None => latest_storage_meta.latest_end_version(&index_name),
    };
    let end_version = get_end_version(&db, start_version, versions.1).await?;
    debug!(
        "Got start_version: {}, end_version: {}",
        start_version, end_version
    );
    if let Some(message) = unnecessary_do(start_version, end_version) {
        println!("{}", message.bold().yellow());
        return Ok(0);
    }

    let mut collection_source = CollectionSource::new(db.clone());
    let (collection_sender, collection_receiver) = tokio::sync::mpsc::channel(concurrency);
    let collection_handler = collection_source.send(collection_sender, start_version, end_version);

    let mut document_source =
        CollectionDocumentSource::new(db.clone(), collection_receiver, whole_deleted);
    let (document_sender, document_receiver) = tokio::sync::mpsc::channel(concurrency);
    let document_handler = document_source.send(document_sender, start_version, end_version);

    let mut document_sink = sink::CollectionIndexSink::new(
        storage_path.clone(),
        index_name.clone(),
        start_version,
        end_version,
        concurrency,
        memory_budget,
    );
    let document_sink_handler = document_sink.sink(document_receiver);

    let (collections_report, docs_report, sink_report) =
        try_join!(collection_handler, document_handler, document_sink_handler)?;

    let minimal_successes = std::env::var("MINIMAL_SUCCESSES")
        .unwrap_or("1000".into())
        .parse::<i64>()
        .unwrap_or(1000);

    // not enough docs OR all docs are deleted
    if sink_report.successes < minimal_successes || sink_report.successes == sink_report.deletes {
        warn!(
            "Not enough docs OR all docs are deleted, successes: {}, deletes: {}",
            sink_report.successes, sink_report.deletes
        );
        return Ok(0);
    }

    // save storage-meta
    latest_storage_meta.version += 1;
    latest_storage_meta
        .indexes
        .entry(index_name.clone())
        .and_modify(|index_meta| {
            if start_version == 0 {
                index_meta.incrementals.clear();
            }
            index_meta
                .incrementals
                .push(Incremental::new(start_version, end_version));
        })
        .or_insert({
            Index::new(
                SCHEMA_VERSION.into(),
                vec![Incremental::new(start_version, end_version)],
            )
        });
    latest_storage_meta.save(&storage_path)?;

    // save incremental-meta
    let mut incremental_meta = sink_report.index_meta.clone().unwrap().current().clone();
    incremental_meta.set_meta(
        index_name.clone(),
        start_version,
        end_version,
        SCHEMA_VERSION.into(),
    );
    incremental_meta.save(&storage_path).await?;

    // update incremental-meta to cloud
    incremental_meta
        .upload(&storage_path, s3service.clone())
        .await?;

    // upload storage-meta to cloud
    latest_storage_meta.upload(s3service.clone()).await?;

    // upload deletes
    Deletes::upload_deleteds(
        latest_storage_meta.version,
        s3service.clone(),
        sink_report.deleteds.clone(),
    )
    .await?;

    // activate new version
    StorageMeta::activate(s3service, latest_storage_meta.version).await?;

    let version = latest_storage_meta.version;
    join!(report(
        db,
        collections_report,
        docs_report,
        sink_report,
        instant.elapsed(),
        start_version,
        latest_storage_meta,
    ));

    Ok(version)
}

async fn get_end_version(
    pool: &MySqlPool,
    start_version: u64,
    end_version: Option<u64>,
) -> Result<u64> {
    match end_version {
        Some(end_version) => Ok(end_version),
        None => Ok(CollectionMeta::new_index_version(pool, start_version).await?),
    }
}

async fn report(
    pool: Arc<MySqlPool>,
    collections_report: SourceReport,
    docs_report: SourceReport,
    sink_report: SinkReport,
    all_cost: std::time::Duration,
    start_version: u64,
    latest_storage_meta: StorageMeta,
) {
    info!("Collection report: {:?}", collections_report);
    info!("Docs report: {:?}", docs_report);
    info!("Sink report: {:?}", sink_report);

    match PipelineReport::init(&pool).await {
        Ok(_) => {
            info!("Init pipeline report success");
        }
        Err(e) => {
            warn!("Failed to init pipeline report: {:?}", e);
            return;
        }
    };
    match PipelineReport::new(
        serde_json::to_string(&collections_report).unwrap_or_default(),
        serde_json::to_string(&docs_report).unwrap_or_default(),
        serde_json::to_string(&sink_report).unwrap_or_default(),
        all_cost.as_millis() as i64,
        latest_storage_meta.version as i64,
        start_version as i64,
        latest_storage_meta.latest_end_version("shard0") as i64,
    )
    .save(&pool)
    .await
    {
        Ok(_) => {
            info!("Save pipeline report success");
        }
        Err(e) => {
            warn!("Failed to save pipeline report: {:?}", e);
        }
    };
}

fn unnecessary_do(start_version: u64, end_version: u64) -> Option<String> {
    if end_version <= start_version {
        Some(format!(
            "No collection was updated since version {}",
            start_version
        ))
    } else {
        None
    }
}
#[cfg(test)]
mod tests {
    use std::env;

    use super::*;

    #[tokio::test]
    async fn test_index_updated_collections() {
        let dsn = if let Ok(e) = env::var("DATABASE_URL") {
            e
        } else {
            return assert!(true);
        };
        let storage_path = "/tmp/storage".to_string();
        let index_name = "test_index".to_string();
        let concurrency = 4;

        let result = index_updated_collections(
            dsn,
            (None, None),
            None,
            storage_path,
            index_name,
            concurrency,
            false,
        )
        .await;

        assert!(result.is_ok());
    }
}