use std::future::Future;

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},
    worker::mocks::make_frag_meta_by_range,
};
use tici_common::{
    cdc_file_name::{CDC_FILENAME_MATCH, CDCFileName},
    index::IndexKey,
};
use tici_test_utils::test_utils::{
    mock_frag_path, partition_table::make_create_partition_index_req, test_suite::E2ETestSuite,
};

const TEST_TABLE_VERSION: u64 = 461_869_219_056_451_591_u64;

fn make_cdc_key(prefix: &str, table_version: u64, seq: u64) -> String {
    CDCFileName::new(prefix.to_string(), table_version, format!("{seq:020}"))
        .unwrap()
        .to_string()
}

async fn list_cdc_files(
    storage: &std::sync::Arc<dyn storage_backend::storage::BackendStorage>,
    bucket: &str,
    prefix: &str,
) -> Vec<String> {
    let mut keys = storage
        .list_all(bucket, prefix)
        .await
        .unwrap()
        .into_iter()
        .filter(|k| CDC_FILENAME_MATCH.is_match(k))
        .collect::<Vec<_>>();
    keys.sort();
    keys
}

fn run_with_mock_suite<F>(tag: &str, f: F) -> Result<()>
where
    F: for<'a> FnOnce(
        &'a mut E2ETestSuite,
    ) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + 'a>>,
{
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;

    fail::cfg("worker_service_mock_meta_service_url", "return").unwrap();
    defer!({
        fail::remove("worker_service_mock_meta_service_url");
    });

    let mut meta_cfg = make_default_test_meta_config(true);
    let mut worker_cfg = make_default_test_worker_config(true);
    let unique_prefix = format!("tici_default_prefix_cdc_gc_{tag}_{}", uuid::Uuid::new_v4());
    meta_cfg.s3.prefix = unique_prefix.clone();
    worker_cfg.s3.prefix = unique_prefix;

    rt.block_on(async move {
        let mut suite = E2ETestSuite::new_with_mock_cfg(meta_cfg, worker_cfg).await?;
        f(&mut suite).await?;
        suite.stop().await?;
        Ok(())
    })
}

#[test]
fn test_gc_cdc_files_delete_iff_all_shards_advanced() -> Result<()> {
    run_with_mock_suite("all_shards_advanced", |suite| {
        Box::pin(async move {
            let index_key = IndexKey::new_nullspace(1, 1001);
            suite.must_create_index(&index_key).await?;
            let index = suite.get_index(&index_key).await;
            let cdc_location = index.get_cdc_location();

            let storage = create_backend_storage_for_test(&suite.get_meta_config().s3).await?;

            let cdc_key = |seq| make_cdc_key(&cdc_location.prefix, TEST_TABLE_VERSION, seq);
            let [cdc_old, cdc_curr, cdc_next] = [9_u64, 10, 11].map(cdc_key);
            for key in [&cdc_old, &cdc_curr, &cdc_next] {
                storage
                    .put_data(&cdc_location.bucket, key, b"{}".to_vec())
                    .await?;
            }
            assert_eq!(
                list_cdc_files(&storage, &cdc_location.bucket, &cdc_location.prefix).await,
                vec![cdc_old.clone(), cdc_curr.clone(), cdc_next.clone()],
            );

            // Split into 2 shards so we can simulate "not all shards advanced".
            let first = index.select_first_shard().await.unwrap();
            suite
                .must_split_shard_simple(&index_key, first.read().await.range.clone(), Some(b"A"))
                .await
                .unwrap();

            let shards = index.clone_shards().await;
            assert_eq!(shards.len(), 2);
            let s0 = shards[0].read().await.clone();
            let s1 = shards[1].read().await.clone();

            let addr = suite.get_one_worker_node_addr();

            // Shard A advanced to 11, shard B only to 10. GC should delete files < 10
            // (i.e., delete `cdc_old`), but must keep `cdc_curr`.
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s0.shard_id,
                    &index_key,
                    cdc_next.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_s0_0"),
                        &s0.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s1.shard_id,
                    &index_key,
                    cdc_curr.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_s1_0"),
                        &s1.range.debug_derive_frag_range(),
                    )],
                )
                .await?;

            let deleted = suite
                .get_meta_service()
                .shard_task_manager
                .gc_cdc_files_by_table(index_key.keyspace_id, index_key.table_id)
                .await?;
            assert_eq!(deleted, vec![cdc_old.clone()]);
            assert_eq!(
                list_cdc_files(&storage, &cdc_location.bucket, &cdc_location.prefix).await,
                vec![cdc_curr.clone(), cdc_next.clone()],
            );

            // Now shard B also advances to 11, GC should delete files < 11 (i.e., delete
            // 10).
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s1.shard_id,
                    &index_key,
                    cdc_next.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_s1_1"),
                        &s1.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            let deleted = suite
                .get_meta_service()
                .shard_task_manager
                .gc_cdc_files_by_table(index_key.keyspace_id, index_key.table_id)
                .await?;
            assert_eq!(deleted, vec![cdc_curr.clone()]);
            assert_eq!(
                list_cdc_files(&storage, &cdc_location.bucket, &cdc_location.prefix).await,
                vec![cdc_next.clone()],
            );

            // Writes and reads still work after GC: append once more and read progress back
            // from meta.
            let cdc_latest = cdc_key(12);
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s0.shard_id,
                    &index_key,
                    cdc_latest.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_s0_2"),
                        &s0.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            let restored = suite.get_shard_from_restore(s0.shard_id).await?;
            assert_eq!(restored.progress.cdc_s3_last_file, cdc_latest);
            Ok(())
        })
    })
}

#[test]
fn test_gc_cdc_files_two_indices_under_same_table() -> Result<()> {
    run_with_mock_suite("two_indices", |suite| {
        Box::pin(async move {
            let index_key_1 = IndexKey::new_nullspace(3, 1003);
            let index_key_2 = IndexKey::new_nullspace(3, 1004);
            suite.must_create_index(&index_key_1).await?;
            suite.must_create_index(&index_key_2).await?;

            let index_1 = suite.get_index(&index_key_1).await;
            let index_2 = suite.get_index(&index_key_2).await;
            let cdc_location_1 = index_1.get_cdc_location();
            let cdc_location_2 = index_2.get_cdc_location();
            assert_eq!(cdc_location_1.bucket, cdc_location_2.bucket);
            assert_eq!(cdc_location_1.prefix, cdc_location_2.prefix);

            let storage = create_backend_storage_for_test(&suite.get_meta_config().s3).await?;

            let cdc_key = |seq| make_cdc_key(&cdc_location_1.prefix, TEST_TABLE_VERSION, seq);
            let [cdc_old, cdc_curr, cdc_next] = [9_u64, 10, 11].map(cdc_key);
            for key in [&cdc_old, &cdc_curr, &cdc_next] {
                storage
                    .put_data(&cdc_location_1.bucket, key, b"{}".to_vec())
                    .await?;
            }
            assert_eq!(
                list_cdc_files(&storage, &cdc_location_1.bucket, &cdc_location_1.prefix).await,
                vec![cdc_old.clone(), cdc_curr.clone(), cdc_next.clone()],
            );

            let s0 = index_1.select_first_shard().await.unwrap();
            let s1 = index_2.select_first_shard().await.unwrap();
            let s0 = s0.read().await.clone();
            let s1 = s1.read().await.clone();
            let addr = suite.get_one_worker_node_addr();

            // Index-1 shard advanced to 11, index-2 shard only to 10. GC should delete
            // files < 10 (i.e., delete `cdc_old`), but must keep `cdc_curr`.
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s0.shard_id,
                    &index_key_1,
                    cdc_next.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_mi_s0_0"),
                        &s0.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s1.shard_id,
                    &index_key_2,
                    cdc_curr.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_mi_s1_0"),
                        &s1.range.debug_derive_frag_range(),
                    )],
                )
                .await?;

            let deleted = suite
                .get_meta_service()
                .shard_task_manager
                .gc_cdc_files_by_table(index_key_1.keyspace_id, index_key_1.table_id)
                .await?;
            assert_eq!(deleted, vec![cdc_old.clone()]);
            assert_eq!(
                list_cdc_files(&storage, &cdc_location_1.bucket, &cdc_location_1.prefix).await,
                vec![cdc_curr.clone(), cdc_next.clone()],
            );

            // Now index-2 shard also advances to 11, GC should delete files < 11 (i.e.,
            // delete `cdc_curr`).
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s1.shard_id,
                    &index_key_2,
                    cdc_next.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_mi_s1_1"),
                        &s1.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            let deleted = suite
                .get_meta_service()
                .shard_task_manager
                .gc_cdc_files_by_table(index_key_1.keyspace_id, index_key_1.table_id)
                .await?;
            assert_eq!(deleted, vec![cdc_curr.clone()]);
            assert_eq!(
                list_cdc_files(&storage, &cdc_location_1.bucket, &cdc_location_1.prefix).await,
                vec![cdc_next.clone()],
            );

            // Writes and reads still work after GC: append once more and read progress back
            // from meta.
            let cdc_latest = cdc_key(12);
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    s0.shard_id,
                    &index_key_1,
                    cdc_latest.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_mi_s0_2"),
                        &s0.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            let restored = suite.get_shard_from_restore(s0.shard_id).await?;
            assert_eq!(restored.progress.cdc_s3_last_file, cdc_latest);

            Ok(())
        })
    })
}

#[test]
fn test_gc_cdc_files_partition_table_gc_separately() -> Result<()> {
    run_with_mock_suite("partition_table", |suite| {
        Box::pin(async move {
            // One logical table with 2 partitions:
            // - partition-1 has 1 shard
            // - partition-2 has 2 shards (after split)
            let logical_index_key = IndexKey::new_nullspace(9000, 9010);
            let physical_table_id_1 = 9001;
            let physical_table_id_2 = 9002;
            suite
                .must_create_index_with_req(make_create_partition_index_req(
                    &logical_index_key,
                    &[physical_table_id_1, physical_table_id_2],
                ))
                .await?;

            let index_key_1 =
                IndexKey::new_nullspace(physical_table_id_1, logical_index_key.index_id);
            let index_key_2 =
                IndexKey::new_nullspace(physical_table_id_2, logical_index_key.index_id);
            let index_1 = suite.get_index(&index_key_1).await;
            let index_2 = suite.get_index(&index_key_2).await;

            let cdc_location_1 = index_1.get_cdc_location();
            let cdc_location_2 = index_2.get_cdc_location();
            assert_eq!(cdc_location_1.bucket, cdc_location_2.bucket);
            assert_ne!(cdc_location_1.prefix, cdc_location_2.prefix);

            let storage = create_backend_storage_for_test(&suite.get_meta_config().s3).await?;

            // Upload a minimal set of CDC files for both partitions under their
            // respective table prefixes. Note sequences across partitions are
            // unrelated; only the per-partition ordering matters.
            let cdc_key_p1 =
                |seq: u64| make_cdc_key(&cdc_location_1.prefix, TEST_TABLE_VERSION, seq);
            let cdc_key_p2 =
                |seq: u64| make_cdc_key(&cdc_location_2.prefix, TEST_TABLE_VERSION, seq);
            let p1_keep = cdc_key_p1(100);
            let [p2_old, p2_curr, p2_next] = [1_u64, 2, 3].map(cdc_key_p2);
            for key in [&p1_keep, &p2_old, &p2_curr] {
                storage
                    .put_data(&cdc_location_1.bucket, key, b"{}".to_vec())
                    .await?;
            }
            assert_eq!(
                list_cdc_files(&storage, &cdc_location_1.bucket, &cdc_location_1.prefix).await,
                vec![p1_keep.clone()],
            );
            assert_eq!(
                list_cdc_files(&storage, &cdc_location_2.bucket, &cdc_location_2.prefix).await,
                vec![p2_old.clone(), p2_curr.clone()],
            );

            // Split partition-2 into 2 shards.
            let first = index_2.select_first_shard().await.unwrap();
            suite
                .must_split_shard_simple(&index_key_2, first.read().await.range.clone(), Some(b"A"))
                .await
                .unwrap();

            let shards_1 = index_1.clone_shards().await;
            assert_eq!(shards_1.len(), 1);

            let shards_2 = index_2.clone_shards().await;
            assert_eq!(shards_2.len(), 2);
            let shard_2_0 = shards_2[0].read().await.clone();
            let shard_2_1 = shards_2[1].read().await.clone();

            let addr = suite.get_one_worker_node_addr();

            // We only need partition-1 CDC objects to exist in S3 to assert that a GC
            // call on partition-2 will never delete them.

            // Partition-2 has 2 shards; only one advanced to `p2_next`. GC for
            // partition-2 must:
            // - delete `p2_old` (older than watermark)
            // - keep `p2_curr` (watermark itself)
            // - never delete partition-1 files
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    shard_2_0.shard_id,
                    &index_key_2,
                    p2_next.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_pt_p2_s0_0"),
                        &shard_2_0.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    shard_2_1.shard_id,
                    &index_key_2,
                    p2_curr.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_pt_p2_s1_0"),
                        &shard_2_1.range.debug_derive_frag_range(),
                    )],
                )
                .await?;

            let deleted_p2 = suite
                .get_meta_service()
                .shard_task_manager
                .gc_cdc_files_by_table(index_key_2.keyspace_id, index_key_2.table_id)
                .await?;
            assert_eq!(deleted_p2, vec![p2_old.clone()]);

            assert_eq!(
                list_cdc_files(&storage, &cdc_location_1.bucket, &cdc_location_1.prefix).await,
                vec![p1_keep.clone()],
            );
            assert_eq!(
                list_cdc_files(&storage, &cdc_location_2.bucket, &cdc_location_2.prefix).await,
                vec![p2_curr.clone()],
            );

            // Advance the remaining partition-2 shard to `p2_next`. GC should now
            // delete `p2_curr`.
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    shard_2_1.shard_id,
                    &index_key_2,
                    p2_next.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_pt_p2_s1_1"),
                        &shard_2_1.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            let deleted_p2 = suite
                .get_meta_service()
                .shard_task_manager
                .gc_cdc_files_by_table(index_key_2.keyspace_id, index_key_2.table_id)
                .await?;
            assert_eq!(deleted_p2, vec![p2_curr.clone()]);

            assert_eq!(
                list_cdc_files(&storage, &cdc_location_1.bucket, &cdc_location_1.prefix).await,
                vec![p1_keep.clone()],
            );

            // Writes and reads still work after GC.
            let p2_after_gc = cdc_key_p2(4);
            suite
                .do_append_delta_frag(
                    addr.as_str(),
                    shard_2_0.shard_id,
                    &index_key_2,
                    p2_after_gc.clone(),
                    vec![make_frag_meta_by_range(
                        &mock_frag_path("cdc_gc_pt_p2_s0_1"),
                        &shard_2_0.range.debug_derive_frag_range(),
                    )],
                )
                .await?;
            let restored = suite.get_shard_from_restore(shard_2_0.shard_id).await?;
            assert_eq!(restored.progress.cdc_s3_last_file, p2_after_gc);

            Ok(())
        })
    })
}