use std::{
    collections::{
        HashMap, HashSet,
        hash_map::Entry::{Occupied, Vacant},
    },
    sync::{Arc, Weak},
    time::Instant,
};

use anyhow::Result;
use arc_swap::ArcSwap;
use log::{debug, info, warn};
use storage_backend::storage::BackendStorage;
use tici_shard::utils::frag_path_generator::FragPathGenerator;
use tokio::sync::mpsc::UnboundedSender;
use txn_types::TimeStamp;

use crate::{
    SplittableFrag,
    config::MetaServiceConfig,
    meta::{
        MetaStore,
        metrics::{SHARD_EVENT_COUNTER_VEC, SHARD_EVENT_DURATION_HISTOGRAM},
        util::BgRunner,
        writer::{FRAG_FILE_PREFIX, ShardTaskManager},
    },
};

pub enum GcTask {
    Manual,
    GetConfig(tokio::sync::oneshot::Sender<Arc<MetaServiceConfig>>),
}

pub struct GcWorker {
    _runner: BgRunner,
    tx: UnboundedSender<GcTask>,
}

pub async fn gc_stale_fragment(
    meta_store: &dyn MetaStore,
    storage: &dyn BackendStorage,
    bucket: &str,
    prefix: &str,
    gc_ts: u64,
) -> Result<Vec<(String, Vec<String>)>> {
    let start = Instant::now();
    SHARD_EVENT_COUNTER_VEC
        .with_label_values(&["gc_frags"])
        .inc();
    let shards = meta_store.restore_shards().await?;
    let shards_len = shards.len();
    let valid_fragment_map: HashSet<_> = shards
        .into_iter()
        .flat_map(|x| {
            let y: Vec<SplittableFrag> = x.manifest.into();
            y
        })
        .map(|x: SplittableFrag| x.f.frag_path.clone())
        .filter(|x| !x.is_empty())
        .collect();

    // TODO(P1) LIST command can be amplified, so we need to introduced a layered
    // strategy to amortize the overall cost.
    // However, it is hard to do that because S3 can not LIST with a depth limit.
    // The only way is to list by layer, which includes to many requests.
    let present_fragments = storage.list_all(bucket, prefix).await?;
    let mut present_fragments_map = HashMap::<String, Vec<String>>::new();
    for f in present_fragments {
        let frag_path = FragPathGenerator::extract_base(f.as_str()).unwrap_or_default();
        if frag_path.is_empty() {
            // TODO(P2) These directories are to be removed also.
            // Parent directory like `tici_default_prefix/frags/A1001_0`.
            continue;
        }
        match present_fragments_map.entry(frag_path) {
            Occupied(mut o) => o.get_mut().push(f),
            Vacant(v) => {
                let vec = vec![f];
                v.insert(vec);
            }
        }
    }

    let mut to_delete = vec![];
    let now = txn_types::TimeStamp::physical_now();
    for (to_gc_frag, owned_files) in present_fragments_map {
        if !valid_fragment_map.contains(&to_gc_frag) {
            // An old fragment to be gc-ed, or new fragment which has not been appended.
            match FragPathGenerator::extract_timestamp(&to_gc_frag) {
                None => {
                    // Invalid fragments, delete anyway.
                    to_delete.push((to_gc_frag, owned_files));
                }
                Some(x) => {
                    let x = TimeStamp::new(x);
                    let physical_millis = x.physical();
                    if physical_millis + gc_ts < now {
                        // Obsolette, to remove
                        debug!(
                            "delete fragment: frag={}, objects={}, physical_millis={}, gc_ts={}, final={}",
                            to_gc_frag,
                            owned_files.len(),
                            physical_millis,
                            gc_ts,
                            physical_millis + gc_ts,
                        );
                        to_delete.push((to_gc_frag, owned_files));
                    }
                }
            }
        }
    }
    if !to_delete.is_empty() {
        info!(
            "gc fragments: shard_len={}, valid_frags={}, safe_gc_ts={}, deleted_frags=[{}]{}",
            shards_len,
            valid_fragment_map.len(),
            gc_ts,
            to_delete.len(),
            to_delete
                .iter()
                .map(|x| { FragPathGenerator::extract_timestamp(x.0.as_str()).unwrap_or(0) })
                .map(|x| x.to_string())
                .collect::<Vec<_>>()
                .join(",")
        );
        let e = start.elapsed().as_secs_f64();
        SHARD_EVENT_DURATION_HISTOGRAM
            .with_label_values(&["gc_frags"])
            .observe(e);
    }
    for (i, (_frag, objects)) in to_delete.iter().enumerate() {
        storage.delete_objects(bucket, objects.as_slice()).await?;
        if i % 100 == 0 {
            tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
        }
    }
    info!(
        "gc fragments finished: shard_len={}, valid_frags={}, safe_gc_ts={}, deleted_frags={}",
        shards_len,
        valid_fragment_map.len(),
        gc_ts,
        to_delete.len(),
    );
    Ok(to_delete)
}

impl GcWorker {
    pub fn new(
        meta_config: Arc<ArcSwap<MetaServiceConfig>>,
        meta_store: Arc<dyn MetaStore>,
        backend_storage: Arc<dyn BackendStorage>,
        shard_task_manager: Weak<ShardTaskManager>,
    ) -> Self {
        // TODO(P2) make this config modificable
        let duration = meta_config
            .load()
            .shard
            .gc_config
            .gc_worker_check_interval
            .0;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        let runner = BgRunner::spawn(move |mut rf| async move {
            use tokio::time::interval;
            let mut interval = interval(duration);
            let prefix = format!("{}/{}", meta_config.load().s3.prefix, FRAG_FILE_PREFIX);
            loop {
                let safe_gc_ts_diff = meta_config
                    .load()
                    .shard
                    .gc_config
                    .safe_gc_ts_diff
                    .as_millis();
                tokio::select! {
                    _ = interval.tick() => {
                        info!("start gc worker task, safe_gc_ts_diff={}", safe_gc_ts_diff);
                        if let Err(e) = gc_stale_fragment(meta_store.as_ref(), backend_storage.as_ref(), &meta_config.load().s3.bucket, &prefix, safe_gc_ts_diff).await {
                            info!("gc_stale_fragment error, {:?}", e);
                        }
                        if let Some(mgr) = shard_task_manager.upgrade() {
                            mgr.gc_cdc_files().await;
                        }
                        info!("finished gc worker task");
                    },
                    msg = rx.recv() => {
                        match msg {
                            Some(GcTask::Manual) => {
                                if let Err(e) = gc_stale_fragment(meta_store.as_ref(), backend_storage.as_ref(), &meta_config.load().s3.bucket, &prefix, safe_gc_ts_diff).await {
                                    info!("gc error, {:?}", e);
                                }
                            }
                            Some(GcTask::GetConfig(r)) => {
                                if let Err(e) = r.send(meta_config.load_full()) {
                                    warn!("GcTask::GetConfig error, e={:?}", e);
                                }
                            }
                            None => {
                                break;
                            }
                        }
                    }
                    _ = &mut rf => {
                        return;
                    }
                }
            }
        });
        Self {
            _runner: runner,
            tx,
        }
    }

    pub fn get_tx(&self) -> UnboundedSender<GcTask> {
        self.tx.clone()
    }
}