use anyhow::Result;
use tici::{
    config::{make_default_test_meta_config, make_default_test_worker_config},
    meta::writer::{CompactShardReason, CompactionState, build_compact_frag_req},
};
use tici_common::index::IndexKey;
use tici_grpc::CompactType;
use tici_test_utils::test_utils::{AppendExtraParam, test_suite::E2ETestSuite};

#[test]
fn test_full_compaction_type_switches_after_finish_append_base() -> Result<()> {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()?;

    rt.block_on(async {
        let mut suite = E2ETestSuite::new_with_mock_cfg(
            make_default_test_meta_config(false),
            make_default_test_worker_config(false),
        )
        .await
        .unwrap();

        let index_key = IndexKey::new_nullspace(1, 1004);
        suite.must_create_index(&index_key).await.unwrap();
        suite.validate_add_index_result(&index_key).await;

        let shard = suite.get_first_shard().await;
        let shard_id = shard.shard_id;
        suite
            .must_append_delta_by_shard(&shard, "F0", 1, AppendExtraParam::default())
            .await;
        suite
            .must_append_delta_by_shard(&shard, "F1", 2, AppendExtraParam::default())
            .await;
        suite
            .must_append_delta_by_shard(&shard, "F2", 3, AppendExtraParam::default())
            .await;

        let index = suite.get_index(&index_key).await;
        let shard_before_finish = suite
            .get_shard_from_index(&index_key, shard_id)
            .await
            .unwrap();
        let shard_before_finish = shard_before_finish.read().await;

        let delta_before_finish = CompactionState::new(
            &shard_before_finish,
            index.clone(),
            shard_before_finish.manifest.delta_slice().to_vec(),
            None,
            0,
            0,
            CompactShardReason::Uni,
            1,
        );
        assert_eq!(delta_before_finish.compact_type(), CompactType::Delta);
        drop(shard_before_finish);

        assert!(index.finish_append_base().await.unwrap());
        assert!(!index.finish_append_base().await.unwrap());

        let shard_after_finish = suite
            .get_shard_from_index(&index_key, shard_id)
            .await
            .unwrap();
        let shard_after_finish = shard_after_finish.read().await;

        let full_state = CompactionState::new(
            &shard_after_finish,
            index.clone(),
            shard_after_finish.manifest.delta_slice().to_vec(),
            None,
            0,
            0,
            CompactShardReason::Uni,
            1,
        );
        assert_eq!(full_state.compact_type(), CompactType::Full);

        let delta_from_middle = CompactionState::new(
            &shard_after_finish,
            index.clone(),
            shard_after_finish.manifest.delta_slice()[1..].to_vec(),
            None,
            0,
            0,
            CompactShardReason::Uni,
            1,
        );
        assert_eq!(delta_from_middle.compact_type(), CompactType::Delta);

        let req = build_compact_frag_req(
            shard_after_finish.get_header(),
            &full_state,
            String::new(),
            index.table_info.clone(),
            index.index_info.clone(),
        );
        assert_eq!(req.compact_type, CompactType::Full as i32);

        drop(shard_after_finish);
        suite.stop().await.unwrap();
    });

    Ok(())
}