use anyhow::{Ok, Result};
use scopeguard::defer;
use storage_backend::storage::create_backend_storage_for_test;
use tici::{
    config::{make_default_test_meta_config, make_default_test_worker_config},
    meta::{
        metrics::INDEX_COUNT_GAUGE_VEC,
        writer::{FRAG_FILE_PREFIX, gc_stale_fragment},
    },
};
use tici_common::index::IndexKey;
use tici_test_utils::test_utils::{test_suite::E2ETestSuite, *};

#[test]
fn test_gc_multiple_indexes() -> Result<()> {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;
    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    fail::cfg("mock_etcd_leader_term", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
        fail::remove("mock_etcd_leader_term");
    });

    rt.block_on(async {
        let mut meta_cfg = make_default_test_meta_config(false);
        let mut worker_cfg = make_default_test_worker_config(false);
        let unique_prefix = format!("test_gc_multiple_indexes_{}", uuid::Uuid::new_v4());
        meta_cfg.s3.prefix = unique_prefix.clone();
        worker_cfg.s3.prefix = unique_prefix;
        let mut test_suite = E2ETestSuite::new_with_mock_cfg(meta_cfg, worker_cfg)
            .await
            .unwrap();

        assert_eq!(
            INDEX_COUNT_GAUGE_VEC
                .get_metric_with_label_values(&["total"])
                .iter()
                .next()
                .unwrap()
                .get(),
            0
        );
        let index_key1 = IndexKey::new_nullspace(1, 1001);
        test_suite.must_create_index(&index_key1).await.unwrap();
        let index_key2 = IndexKey::new_nullspace(1, 1002);
        test_suite.must_create_index(&index_key2).await.unwrap();

        let shard1 = test_suite
            .get_index(&index_key1)
            .await
            .select_first_shard()
            .await
            .unwrap();
        let shard1_unwrap = shard1.read().await.clone();
        let shard2 = test_suite
            .get_index(&index_key2)
            .await
            .select_first_shard()
            .await
            .unwrap();
        let shard2_unwrap = shard2.read().await.clone();
        // fragment 0 of index i
        test_suite
            .must_append_delta_by_shard(
                &shard1_unwrap,
                &format!("A{}_{}", 1001, 0),
                1,
                AppendExtraParam::default(),
            )
            .await;
        test_suite
            .must_append_delta_by_shard(
                &shard2_unwrap,
                &format!("A{}_{}", 1001, 0),
                1,
                AppendExtraParam::default(),
            )
            .await;

        let meta_config = test_suite.get_meta_config();
        let storage = create_backend_storage_for_test(&meta_config.s3)
            .await
            .unwrap();
        let prefix = format!("{}/{}", meta_config.s3.prefix, FRAG_FILE_PREFIX);
        let safe_gc_ts_diff = meta_config.shard.gc_config.safe_gc_ts_diff.as_millis();
        gc_stale_fragment(
            test_suite.get_meta_service().meta_store.as_ref(),
            storage.as_ref(),
            &meta_config.s3.bucket,
            &prefix,
            safe_gc_ts_diff,
        )
        .await
        .unwrap();

        let present_fragments = storage
            .list_all(&meta_config.s3.bucket, prefix.as_str())
            .await
            .unwrap();
        assert_eq!(present_fragments.len(), 1);

        test_suite.must_drop_index(&index_key1).await.unwrap();
        test_suite.must_drop_index(&index_key2).await.unwrap();

        test_suite.stop_worker().await.unwrap();
        assert_eq!(test_suite.collect_shards_from_restore().await.len(), 0);
        assert_eq!(test_suite.collect_indexes_from_restore().await.len(), 0);
        test_suite.stop_meta().await.unwrap();
    });

    Ok(())
}