use std::{collections::HashSet, sync::Arc, time::Instant};

use anyhow::{Context, Result};
use log::{debug, info};
use storage_backend::storage::{STORAGE_SERVICE_META, create_backend_storage};
use tici_common::cdc_file_name::CDCFileName;
use tokio::sync::RwLock;

use crate::{
    ShardInfo,
    meta::{
        metrics::{SHARD_EVENT_COUNTER_VEC, SHARD_EVENT_DURATION_HISTOGRAM},
        writer::{FRAG_FILE_PREFIX, FTSIndex, ShardTaskManager, gc_stale_fragment},
    },
};

impl ShardTaskManager {
    pub async fn manual_gc(&self) -> Result<Vec<(String, Vec<String>)>> {
        let storage = create_backend_storage(&self.config.load().s3, STORAGE_SERVICE_META).await?;
        let prefix = format!("{}/{}", self.config.load().s3.prefix, FRAG_FILE_PREFIX);
        gc_stale_fragment(
            self.meta_store.as_ref(),
            storage.as_ref(),
            &self.config.load().s3.bucket,
            &prefix,
            600_000,
        )
        .await
    }

    async fn scan_cdc_files(
        &self,
        keyspace_id: u32,
        physical_table_id: i64,
    ) -> Result<(Vec<Arc<RwLock<ShardInfo>>>, (String, String))> {
        let indices: Vec<Arc<FTSIndex>> = {
            let guard = self.indices.read().await;
            guard
                .values()
                .filter(|index| {
                    index.keyspace_id() == keyspace_id && index.table_id() == physical_table_id
                })
                .cloned()
                .collect()
        };
        if indices.is_empty() {
            return Ok((vec![], ("".to_string(), "".to_string())));
        }
        // We expected all indices under the same table_id to have the same cdc prefix.
        let cdc_buckets_and_prefixes: HashSet<(String, String)> = indices
            .iter()
            .map(|x| {
                (
                    x.get_cdc_s3_bucket().to_string(),
                    x.get_cdc_s3_prefix().to_string(),
                )
            })
            .collect();
        let cdc_buckets_and_prefixes: Vec<(String, String)> =
            cdc_buckets_and_prefixes.into_iter().collect();

        let (cdc_bucket, cdc_prefix) = if let [(cdc_bucket, cdc_prefix)] =
            cdc_buckets_and_prefixes.as_slice()
        {
            (cdc_bucket.clone(), cdc_prefix.clone())
        } else {
            anyhow::bail!(
                "index must have the only bucket and prefix in all shards, cdc_buckets_and_prefixes={:?}, keyspace_id={}, table_id={}",
                cdc_buckets_and_prefixes,
                keyspace_id,
                physical_table_id
            );
        };
        let mut shards: Vec<Arc<RwLock<ShardInfo>>> = vec![];
        for index in indices.iter() {
            let mut s = index.clone_shards().await;
            shards.append(&mut s);
        }
        Ok((shards, (cdc_bucket, cdc_prefix)))
    }

    pub async fn gc_cdc_files(&self) -> (u64, u64) {
        SHARD_EVENT_COUNTER_VEC
            .with_label_values(&["gc_cdcs"])
            .inc();
        let start = Instant::now();
        let tables = self.get_all_table_ids().await;
        let mut failed_table = 0u64;
        let mut suc_table = 0u64;
        for (k, t) in tables {
            match self.gc_cdc_files_by_table(k, t).await {
                Ok(_) => suc_table += 1,
                Err(_) => failed_table += 1,
            }
        }
        let e = start.elapsed().as_secs_f64();
        SHARD_EVENT_DURATION_HISTOGRAM
            .with_label_values(&["gc_cdcs"])
            .observe(e);
        (suc_table, failed_table)
    }

    pub async fn gc_cdc_files_by_table(
        &self,
        keyspace_id: u32,
        physical_table_id: i64,
    ) -> Result<Vec<String>> {
        info!(
            "gc_cdc_files: start table, keyspace_id={}, table_id={}",
            keyspace_id, physical_table_id,
        );
        let storage = create_backend_storage(&self.config.load().s3, STORAGE_SERVICE_META).await?;
        let (shards, (cdc_bucket, cdc_prefix)) =
            self.scan_cdc_files(keyspace_id, physical_table_id).await?;
        if shards.is_empty() {
            // Index exists but shards are not bootstrapped yet. Be conservative.
            return Ok(vec![]);
        }

        // Determine the GC watermark from shard progress: the minimum processed file
        // across all shards of all indices for this table.
        let mut watermark: Option<CDCFileName> = None;
        for shard in shards {
            let shard = shard.read().await;
            let last = shard.progress.cdc_s3_last_file.as_str();
            if last.is_empty() {
                // At least a shard is reading from the beginning, we can not gc.
                return Ok(vec![]);
            }

            let parsed = CDCFileName::parse(last).with_context(|| {
                format!(
                    "gc_cdc_files: invalid shard progress cdc file, shard_id={}, index_key={:?}, last_file={}",
                    shard.shard_id,
                    shard.index_key,
                    last
                )
            })?;

            if parsed.prefix() != cdc_prefix {
                anyhow::bail!(
                    "gc_cdc_files: shard progress prefix mismatch, keyspace_id={}, table_id={}, shard_id={}, index_key={:?}, expected_prefix={}, got_prefix={}, last_file={}",
                    keyspace_id,
                    physical_table_id,
                    shard.shard_id,
                    shard.index_key,
                    cdc_prefix,
                    parsed.prefix(),
                    last
                );
            }

            watermark = Some(match watermark {
                None => parsed,
                Some(prev) => std::cmp::min(prev, parsed),
            });
        }

        let watermark = watermark.ok_or_else(|| anyhow::anyhow!(
            "can not determine min watermark, can't gc. table_id={physical_table_id}, keyspace_id={keyspace_id}"
        ))?;

        // List all CDC objects under this table prefix and delete those older than
        // watermark.
        let keys = storage.list_all(&cdc_bucket, &cdc_prefix).await?;
        let mut to_delete = Vec::new();
        for key in keys {
            let parsed = match CDCFileName::parse(&key) {
                Ok(v) => {
                    if v.prefix() != cdc_prefix {
                        info!(
                            "gc_cdc_files: skip cdc key with unexpected prefix, expected_prefix={}, key={}",
                            cdc_prefix, key
                        );
                        continue;
                    }
                    v
                }
                Err(e) => {
                    // Maybe something like
                    // tici_default_prefix/cdc/123/meta/
                    // schema_463597423226781697_3984470366.json.
                    debug!(
                        "gc_cdc_files: skip invalid cdc key, key={}, err={:?}",
                        key, e
                    );
                    continue;
                }
            };
            if parsed < watermark {
                to_delete.push(key);
            }
        }

        if to_delete.is_empty() {
            return Ok(vec![]);
        }
        to_delete.sort();

        info!(
            "gc_cdc_files: keyspace_id={}, table_id={}, watermark={}, deleted=[{}]{:?}",
            keyspace_id,
            physical_table_id,
            watermark,
            to_delete.len(),
            to_delete,
        );

        // Delete in batches of 1000 (S3 DeleteObjects limit).
        for chunk in to_delete.chunks(1000) {
            storage.delete_objects(&cdc_bucket, chunk).await?;
        }

        Ok(to_delete)
    }
}