#![allow(dead_code)]

mod cdc_gc;
mod config_load;
pub mod logger;
pub mod meta_failover;
mod meta_handle;
pub mod meta_scaffold;
pub mod partition_table;
pub mod reader_handle;
pub mod shard_cache_scaffold;
pub mod test_suite;
pub mod worker_handle;
pub mod workload_ci;

use std::{backtrace, sync::Arc};

pub use cdc_gc::*;
pub use config_load::*;
use either::Either;
pub use logger::{init_logger_for_test, init_logger_for_ut};
use tici_common::index::IndexKey;
use tici_grpc::ColumnInfo;
use tokio::sync::RwLock;

pub mod runner;

pub async fn must_finally_ex<F, Fut, SuccT, FailT>(f: F) -> Either<SuccT, FailT>
where
    F: Fn(i32) -> Fut,
    Fut: std::future::Future<Output = Either<SuccT, FailT>>,
{
    let mut att = 0;
    let att_max = 10;
    loop {
        match f(att).await {
            Either::Left(v) => return Either::Left(v),
            Either::Right(e) => {
                if att >= att_max {
                    return Either::Right(e);
                }
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(1000));
        att += 1;
    }
}

pub async fn must_finally<F, Fut>(f: F)
where
    F: Fn() -> Fut,
    Fut: std::future::Future<Output = bool>,
{
    let mut att = 0;
    let att_max = 10;
    loop {
        if f().await {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(1000));
        att += 1;
        assert!(
            att < att_max,
            "must_finally: exceeded max attempts {att_max}, backtrace={:#?}",
            backtrace::Backtrace::capture().to_string(),
        );
    }
}

#[derive(Clone, Copy)]
pub struct AppendExtraParam {
    pub upload_to_s3: bool,
    // Check if all levels in the manifest have consistent fragment count
    pub do_manifest_check: bool,
    pub call_check_append_frag: bool,
    pub dedup: (usize, usize, usize),
    pub size: Option<u64>,
    pub count: Option<u64>,
    pub segs_count: Option<usize>,
}

impl Default for AppendExtraParam {
    fn default() -> Self {
        AppendExtraParam {
            upload_to_s3: true,
            do_manifest_check: true,
            call_check_append_frag: true,
            dedup: (0, 0, 0),
            size: None,
            count: None,
            segs_count: None,
        }
    }
}

/// Helper function to create test ColumnInfo
pub fn create_column_info(
    id: i64,
    name: &str,
    column_type: i32,
    is_primary_key: bool,
) -> ColumnInfo {
    ColumnInfo {
        column_id: id,
        column_name: name.to_string(),
        r#type: column_type,
        collation: "utf8mb4_bin".to_string(),
        column_length: 0,
        decimal: 0,
        flag: 0,
        elems: vec![],
        default_val: vec![],
        is_primary_key,
        is_array: false,
    }
}

pub fn create_cdc_file_name(id: u64) -> String {
    format!(
        "tici_default_prefix/cdc/10313/461869219056451591/CDC{:020}.json",
        id
    )
}

pub fn mock_frag_path(path: impl AsRef<str>) -> String {
    format!("{}/f_1", path.as_ref())
}

pub fn mock_frag_paths(paths: &[&str]) -> Vec<String> {
    paths.iter().map(mock_frag_path).collect()
}

use storage_backend::storage::BackendStorage;

// TODO(P2) Integrate with storage module
pub async fn s3_list_frags(
    storage: &Arc<dyn BackendStorage>,
    bucket: &str,
    prefix: &str,
    start_after: Option<&str>,
    index_key: &IndexKey,
) -> anyhow::Result<Vec<String>> {
    let frag_prefix = format!(
        "{}/{}/{}/",
        prefix,
        tici::meta::writer::FRAG_FILE_PREFIX,
        index_key.to_s3_prefix()
    );

    let frags = storage
        .list_with_delimiter(bucket, &frag_prefix, start_after, "/")
        .await?;

    Ok(frags)
}

pub async fn check_reader_writer_manifest_consistent(
    test_suite: Arc<RwLock<test_suite::E2ETestSuite>>,
    index_key: IndexKey,
) -> Either<(), String> {
    for writer_shard in test_suite
        .read()
        .await
        .collect_shards_from_index(&index_key)
        .await
    {
        if let Some(read_shard) = test_suite
            .read()
            .await
            .get_meta_service()
            .shard_cache_manager
            .get_shard_by_id(&writer_shard.shard_id)
            .await
        {
            let reader_delta_paths = read_shard.inner.read().await.manifest.delta_frag_paths();
            let writer_delta_paths = writer_shard.manifest.delta_frag_paths();
            if writer_delta_paths != reader_delta_paths {
                return Either::Right(format!(
                    "error manifest result: shard_id={}, W={:?} <==> R={:?}",
                    writer_shard.shard_id, writer_delta_paths, reader_delta_paths
                ));
            }
        } else {
            return Either::Right(format!(
                "missing reader shard: shard_id={}, writer_delta_paths={:?}",
                writer_shard.shard_id,
                writer_shard.manifest.delta_frag_paths()
            ));
        }
    }
    Either::Left(())
}