use std::{sync::Arc, time::Duration};

use anyhow::{Context, Result, bail, ensure};
use storage_backend::storage::BackendStorage;
use tici::meta::writer::{FTSIndex, ShardTaskManager};
use tici_common::cdc_file_name::{CDC_FILENAME_MATCH, CDCFileName};

async fn compute_cdc_watermark(index: &Arc<FTSIndex>) -> Result<Option<CDCFileName>> {
    let shards = index.clone_shards().await;
    if shards.is_empty() {
        return Ok(None);
    }

    let mut watermark: Option<CDCFileName> = None;
    for shard in shards {
        let last = shard.read().await.progress.cdc_s3_last_file.clone();
        if last.is_empty() {
            // Progress not ready yet.
            return Ok(None);
        }
        let parsed = CDCFileName::parse(&last)?;
        watermark = Some(match watermark {
            None => parsed,
            Some(prev) => std::cmp::min(prev, parsed),
        });
    }

    Ok(watermark)
}

fn synthesize_deletable_key(watermark: &CDCFileName) -> Result<String> {
    // Avoid relying on "sequence 0 is never produced". Instead, synthesize a key
    // that is guaranteed to be strictly older than the current watermark.
    let mut dummy = CDCFileName::new(
        watermark.prefix().to_string(),
        watermark.table_version(),
        "00000000000000000000".to_string(),
    )?;
    if dummy < *watermark {
        return Ok(dummy.to_string());
    }

    // 2) If sequence cannot be lowered (e.g. watermark itself is already 0), lower
    //    the table version.
    let Some(prev_table_version) = watermark.table_version().checked_sub(1) else {
        bail!("cannot synthesize deletable CDC key from watermark={watermark}");
    };
    dummy = CDCFileName::new(
        watermark.prefix().to_string(),
        prev_table_version,
        "99999999999999999999".to_string(),
    )?;
    if dummy < *watermark {
        return Ok(dummy.to_string());
    }

    bail!("cannot synthesize deletable CDC key from watermark={watermark}");
}

fn find_existing_deletable_key(
    keys: Vec<String>,
    cdc_prefix: &str,
    watermark: &CDCFileName,
) -> Option<String> {
    // Pick the newest deletable file (< watermark) if possible.
    let mut candidate: Option<(CDCFileName, String)> = None;
    for key in keys.into_iter() {
        if !CDC_FILENAME_MATCH.is_match(&key) {
            continue;
        }
        let parsed = match CDCFileName::parse(&key) {
            Ok(v) => v,
            Err(_) => continue,
        };
        if parsed.prefix() != cdc_prefix {
            continue;
        }
        if &parsed < watermark {
            match &candidate {
                None => candidate = Some((parsed, key)),
                Some((best, _)) => {
                    if &parsed > best {
                        candidate = Some((parsed, key));
                    }
                }
            }
        }
    }
    candidate.map(|(_, key)| key)
}

pub async fn must_check_gc_files(
    shard_task_manager: &ShardTaskManager,
    index: Arc<FTSIndex>,
    storage: &Arc<dyn BackendStorage>,
) -> Result<bool> {
    let cdc_location = index.get_cdc_location();

    // Wait a bit for shard progress to become available.
    for _ in 0..30 {
        let Some(watermark) = compute_cdc_watermark(&index).await? else {
            tokio::time::sleep(Duration::from_millis(5000)).await;
            continue;
        };

        let keys = storage
            .list_all(&cdc_location.bucket, &cdc_location.prefix)
            .await
            .context("must_check_gc_files: list cdc keys")?;

        let mut real = false;
        let candidate_key = if let Some(key) =
            find_existing_deletable_key(keys, &cdc_location.prefix, &watermark)
        {
            real = true;
            key
        } else {
            let dummy_key = synthesize_deletable_key(&watermark)?;
            storage
                .put_data(&cdc_location.bucket, &dummy_key, b"{}".to_vec())
                .await
                .context("must_check_gc_files: put dummy cdc key")?;
            dummy_key
        };

        let before = storage
            .list_all(&cdc_location.bucket, &cdc_location.prefix)
            .await
            .context("must_check_gc_files: list before gc")?;
        ensure!(
            before.contains(&candidate_key),
            "candidate CDC file should exist before GC: {candidate_key}"
        );

        shard_task_manager.manual_gc().await.unwrap();
        let deleted = shard_task_manager
            .gc_cdc_files_by_table(index.keyspace_id(), index.table_id())
            .await
            .context("must_check_gc_files: gc_cdc_files_by_table")?;

        let after = storage
            .list_all(&cdc_location.bucket, &cdc_location.prefix)
            .await
            .context("must_check_gc_files: list after gc")?;
        ensure!(
            !after.contains(&candidate_key),
            "candidate CDC file should be deleted after GC: {candidate_key}"
        );
        ensure!(
            deleted.contains(&candidate_key),
            "must_check_gc_files should report deleted candidate key: {candidate_key}"
        );

        return Ok(real);
    }

    bail!("must_check_gc_files: shard progress did not become available in time");
}