// Keep this as a separate integration-test binary because failpoints are
// process-global. If this test panics or leaves a failpoint configured, the
// contamination stays within this binary instead of breaking unrelated tests.
use std::{path::Path, sync::Arc, time::Duration};

use anyhow::{Context, Result};
use fail::FailScenario;
use scopeguard::defer;
use tempfile::TempDir;
use tici_grpc::KeyRange;
use tici_shard::{
    config::WorkerNodeConfig,
    reader::frag_data_cache::FragDataCache,
    test_utils::{
        TEST_HDFSLOGS_INDEX_META, TEST_S3_BUCKET, test_create_backend_storage,
        test_get_shard_writer_meta, write_test_data,
    },
    utils::meta_client_pool::{MetaClientPool, MockMetaClientPool},
};
use tici_system::readable::ReadableSize;

async fn wait_for_frag_file(path: &Path) -> Result<()> {
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            if path.exists() {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .with_context(|| format!("Timed out waiting for frag file {}", path.display()))?;
    Ok(())
}

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

    rt.block_on(async {
        let shard_meta = test_get_shard_writer_meta();
        let index_meta = TEST_HDFSLOGS_INDEX_META.clone();
        let worker_data_dir = TempDir::new()?;
        let frag_cache_dir = TempDir::new()?;
        let config = WorkerNodeConfig::new_for_test(
            worker_data_dir
                .path()
                .to_str()
                .context("worker_data_dir to_str failed")?,
        );

        let backend_storage = test_create_backend_storage().await?;
        let meta_client_pool: Arc<dyn MetaClientPool> = Arc::new(MockMetaClientPool::new());
        write_test_data(
            &config,
            std::slice::from_ref(&shard_meta),
            &index_meta,
            backend_storage.clone(),
            meta_client_pool.clone(),
        )
        .await?;

        let frag_data_cache = Arc::new(FragDataCache::new(
            backend_storage,
            frag_cache_dir
                .path()
                .to_str()
                .context("frag_cache_dir to_str failed")?
                .to_string(),
            config.storage.temporary_path(),
            Duration::ZERO,
            ReadableSize::mb(64),
        )?);

        let mut mock_client = meta_client_pool.get_meta_client().await?;
        let (frag_metas, _seq) = mock_client
            .get_frags_by_keyrange(
                shard_meta.shard_header.shard_id,
                &shard_meta.index_key,
                KeyRange::default(),
            )
            .await?;
        let frag_meta = frag_metas
            .into_iter()
            .next()
            .context("expected at least one frag meta")?;

        let local_frag_path = frag_data_cache.local_frag_path(&frag_meta.frag_path);
        let local_frag_meta_path = Path::new(&local_frag_path).join("index/meta.json");

        fail::cfg("frag_data_cache_after_mark_frag_ready", "pause").unwrap();
        defer!({
            fail::remove("frag_data_cache_after_mark_frag_ready");
        });

        let frag_data_cache_for_task = frag_data_cache.clone();
        let frag_meta_for_task = frag_meta.clone();
        let acquire_handle = tokio::spawn(async move {
            frag_data_cache_for_task
                .acquire_frag(TEST_S3_BUCKET, &frag_meta_for_task)
                .await
        });

        wait_for_frag_file(&local_frag_meta_path).await?;

        frag_data_cache.remove_zero_ref_frags();
        assert!(
            local_frag_meta_path.exists(),
            "GC removed freshly downloaded frag while acquire_frag was still returning"
        );

        fail::remove("frag_data_cache_after_mark_frag_ready");
        let acquired_lease = acquire_handle.await.context("acquire task join failed")??;
        assert_eq!(acquired_lease.local_frag_path(), local_frag_path.as_str());
        assert!(
            local_frag_meta_path.exists(),
            "Downloaded frag should still exist"
        );

        drop(acquired_lease);
        frag_data_cache.remove_zero_ref_frags();

        Ok(())
    })
}