use std::{
    fmt::Debug,
    sync::{Arc, atomic::AtomicU16},
    time::{Duration, SystemTime},
};

use anyhow::Result;
use derivative::Derivative;
use log::{debug, info, warn};
use tici_common::index::IndexKey;
use tici_grpc::{
    CompactFragRequest, CompactType, ErrorCode, FinishCompactFragRequest,
    FinishCompactFragResponse, FragMeta, S3Location, ShardManifestHeader,
};

use crate::{
    HandleRange, ShardInfo, SplittableFrag,
    common::shard::FragmentLocation,
    meta::{
        metrics::*,
        writer::{
            CompactShardReason, CompactionWork, FTSIndex, MaybeSplitOrCompactReason, ShardTaskCtx,
            ShardTaskManager, mgr::WorkerNodeMgr,
        },
    },
};

#[derive(Debug, thiserror::Error)]
pub enum CompactionScheduleError {
    #[error(
        "Compaction overlaps with running compaction: shard_id={shard_id}, input={input:?}, conflicting={conflicting:?}"
    )]
    OverlapDetected {
        shard_id: u64,
        input: Vec<String>,       // Fragment paths
        conflicting: Vec<String>, // Conflicting compaction fragment paths
    },
    #[error("Other error: {0}")]
    Other(#[from] anyhow::Error),
}

#[derive(Clone, PartialEq, Hash, Eq, serde::Serialize, serde::Deserialize)]
pub struct CompactionStateKey {
    shard_id: u64,
    seq: u64,
    start_frag: String,
    end_frag: String,
    worker: String,
}

impl heapsize::HeapSizeOf for CompactionStateKey {
    fn heap_size_of_children(&self) -> usize {
        self.start_frag.capacity() + self.end_frag.capacity() + self.worker.capacity()
    }
}

impl Debug for CompactionStateKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_fmt(format_args!(
            "(shard_id={}/seq={}/frags={}..{}/worker={})",
            self.shard_id, self.seq, &self.start_frag, &self.end_frag, &self.worker
        ))
    }
}

impl CompactionStateKey {
    pub fn get_worker(&self) -> String {
        self.worker.clone()
    }

    pub fn get_shard_id(&self) -> u64 {
        self.shard_id
    }
}

impl TryFrom<&FinishCompactFragRequest> for CompactionStateKey {
    type Error = anyhow::Error;
    fn try_from(value: &FinishCompactFragRequest) -> Result<Self, Self::Error> {
        if let Some(shard) = value.shard.as_ref() {
            Ok(Self {
                shard_id: shard.shard_id,
                seq: value.seq,
                start_frag: value
                    .input_frags
                    .first()
                    .map_or(String::new(), |f| f.frag_path.clone()),
                end_frag: value
                    .input_frags
                    .last()
                    .map_or(String::new(), |f| f.frag_path.clone()),
                worker: value.addr.clone(),
            })
        } else {
            Err(anyhow::anyhow!(
                "Shard is None in FinishCompactFragRequest: {:?}",
                value
            ))
        }
    }
}

impl TryFrom<&CompactFragRequest> for CompactionStateKey {
    type Error = anyhow::Error;
    fn try_from(value: &CompactFragRequest) -> Result<Self, Self::Error> {
        if let Some(shard) = value.shard.as_ref() {
            Ok(Self {
                shard_id: shard.shard_id,
                seq: value.seq,
                start_frag: value
                    .input_frags
                    .first()
                    .map_or(String::new(), |f| f.frag_path.clone()),
                end_frag: value
                    .input_frags
                    .last()
                    .map_or(String::new(), |f| f.frag_path.clone()),
                worker: value.worker.clone(),
            })
        } else {
            Err(anyhow::anyhow!(
                "Shard is None in CompactFragRequest: {:?}",
                value
            ))
        }
    }
}

#[derive(Clone, Derivative, serde::Serialize)]
#[derivative(PartialEq)]
pub struct CompactionState {
    pub input_fragments: Vec<SplittableFrag>,
    pub shard_id: u64,
    pub index_key: IndexKey,
    pub range: HandleRange,
    pub epoch: u64,
    pub seq: u64,
    pub start_time: SystemTime,
    pub worker: String,
    pub upload_location: S3Location,
    pub level: Option<u8>,
    pub current_level_states: [u64; 2],
    pub min_split_row_count: u64,
    pub max_split_point_count: u64,
    pub reason: CompactShardReason,
    pub remaining_iterations: usize,
    pub compact_type: CompactType,
    #[serde(skip)]
    #[derivative(PartialEq = "ignore")]
    pub index: Arc<FTSIndex>,
}

impl heapsize::HeapSizeOf for CompactionState {
    fn heap_size_of_children(&self) -> usize {
        self.input_fragments.heap_size_of_children()
    }
}

impl Debug for CompactionState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CompactionState")
            .field("index", &self.index_key)
            .field_with("shard", |f| {
                f.write_fmt(format_args!(
                    "shard_id={}/range={:?}/epoch={}/seq={}",
                    &self.shard_id, self.range, self.epoch, self.seq
                ))
            })
            .field("start_time", &self.get_start_ts())
            .field("worker", &self.worker)
            .field_with("compaction", |f| {
                f.write_fmt(format_args!(
                    "[{}]({}..{}),reason={:?},level={:?},type={:?}",
                    self.input_fragments.len(),
                    self.get_first_name(),
                    self.get_last_name(),
                    self.reason,
                    self.level,
                    self.compact_type
                ))
            })
            .field("lvls", &self.current_level_states)
            .field_with("args", |f| {
                f.write_fmt(format_args!(
                    "mprc={}/mspc={}/remain={}",
                    self.min_split_row_count, self.max_split_point_count, self.remaining_iterations
                ))
            })
            .field("upload_location", &self.upload_location)
            .finish()
    }
}

impl CompactionState {
    pub fn new(
        shard: &ShardInfo,
        index: Arc<FTSIndex>,
        input_fragments: Vec<SplittableFrag>,
        level: Option<u8>,
        min_split_row_count: u64,
        max_split_point_count: u64,
        reason: CompactShardReason,
        remaining_iterations: usize,
    ) -> Self {
        let compact_type = Self::resolve_compact_type(shard, index.as_ref(), &input_fragments);
        Self {
            shard_id: shard.shard_id,
            index_key: shard.index_key.clone(),
            range: shard.range.clone(),
            input_fragments,
            epoch: shard.epoch,
            seq: shard.seq,
            start_time: SystemTime::now(),
            worker: shard.writer.clone(),
            upload_location: shard.get_upload_location(),
            level,
            current_level_states: shard.manifest.current_level_sizes(),
            min_split_row_count,
            max_split_point_count,
            reason,
            remaining_iterations,
            compact_type,
            index,
        }
    }

    fn resolve_compact_type(
        shard: &ShardInfo,
        index: &FTSIndex,
        input_fragments: &[SplittableFrag],
    ) -> CompactType {
        if !index.is_finished_append_base() || !shard.manifest.base_slice().is_empty() {
            return CompactType::Delta;
        }

        let Some(first_frag) = input_fragments.first() else {
            return CompactType::Delta;
        };

        if matches!(
            shard.manifest.find_pos_by_frag_path(&first_frag.frag_path),
            Some(FragmentLocation::Delta(0))
        ) {
            CompactType::Full
        } else {
            CompactType::Delta
        }
    }

    pub fn compact_type(&self) -> CompactType {
        self.compact_type
    }

    pub fn get_header(&self) -> ShardManifestHeader {
        ShardManifestHeader {
            shard_id: self.shard_id,
            start_key: self.range.encoded_start(),
            end_key: self.range.encoded_end(),
            epoch: self.epoch,
            seq: self.seq,
        }
    }

    fn get_first_name(&self) -> String {
        self.input_fragments
            .first()
            .map_or("empty".to_string(), |x| x.frag_path.clone())
    }

    fn get_last_name(&self) -> String {
        self.input_fragments
            .last()
            .map_or("empty".to_string(), |x| x.frag_path.clone())
    }

    pub fn get_start_ts(&self) -> u64 {
        self.start_time
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64
    }

    pub fn get_elapsed_secs(&self) -> f64 {
        self.start_time
            .elapsed()
            .unwrap_or(Duration::from_secs(0))
            .as_secs_f64()
    }

    pub fn get_input_size(&self) -> u64 {
        self.input_fragments.iter().map(|x| x.bytes()).sum()
    }

    pub fn get_weighted_input_size(&self) -> u64 {
        self.input_fragments
            .iter()
            .map(SplittableFrag::weighted_bytes)
            .sum()
    }

    pub fn to_short_string(&self) -> String {
        format!(
            "({:?},shard_id={},Start={:?})(R:{:?})(seq={},epoch={},count={},L{:?}),{}..{}",
            self.index_key,
            self.shard_id,
            self.range.sortable_start_ref(),
            self.reason,
            self.seq,
            self.epoch,
            self.input_fragments.len(),
            self.level.map_or("NaN".to_string(), |x| x.to_string()),
            self.get_first_name(),
            self.get_last_name(),
        )
    }

    pub fn get_compaction_state_key(&self) -> CompactionStateKey {
        CompactionStateKey {
            shard_id: self.shard_id,
            seq: self.seq,
            start_frag: self.get_first_name(),
            end_frag: self.get_last_name(),
            worker: self.worker.clone(),
        }
    }

    pub fn overlap_with(&self, input_fragments: &[SplittableFrag]) -> bool {
        if let (Some(l1), Some(r1)) = (self.input_fragments.first(), self.input_fragments.last()) {
            for f in input_fragments.iter() {
                if f == l1 || f == r1 {
                    return true;
                }
            }
            false
        } else {
            false
        }
    }
}

pub struct AppliedCompactionState {
    pub state: CompactionState,
    output_frag: Vec<FragMeta>,
    after_compaction: Option<MaybeSplitOrCompactReason>,
}

impl AppliedCompactionState {
    pub fn new(
        state: CompactionState,
        output_frag: Vec<FragMeta>,
        after_compaction: Option<MaybeSplitOrCompactReason>,
    ) -> Self {
        Self {
            state,
            output_frag,
            after_compaction,
        }
    }

    pub fn get_output_size(&self) -> u64 {
        self.output_frag.iter().map(|x| x.bytes()).sum()
    }

    fn get_size_in_kb(&self) -> f64 {
        self.state.get_input_size() as f64 / 1024.0
    }

    fn get_size_out_kb(&self) -> f64 {
        self.get_output_size() as f64 / 1024.0
    }

    pub fn compute_compaction_speed(&self) -> f64 {
        let elapsed = self.state.get_elapsed_secs();
        let size_in = self.get_size_in_kb();
        size_in / elapsed
    }

    pub fn compute_effective_speed(&self) -> f64 {
        let elapsed = self.state.get_elapsed_secs();
        let size_in = self.get_size_in_kb();
        let size_out = self.get_size_out_kb();
        if size_out > size_in {
            0.0
        } else {
            (size_in - size_out) / elapsed
        }
    }

    pub fn describe_compaction(&self) -> String {
        format!(
            "compaction_size={:.4} KB->{:.4} KB, rate={:.2}, process={:.2} kbps, effective={:.2} kbps, frag_change: [{}]{:?}->[{}]{:?}",
            self.get_size_in_kb(),
            self.get_size_out_kb(),
            self.get_size_out_kb() / self.get_size_in_kb(),
            self.compute_compaction_speed(),
            self.compute_effective_speed(),
            self.state.input_fragments.len(),
            self.state
                .input_fragments
                .iter()
                .map(|x| x.frag_path.clone())
                .collect::<Vec<_>>()
                .join(","),
            self.output_frag.len(),
            self.output_frag
                .iter()
                .map(|x| x.frag_path.clone())
                .collect::<Vec<_>>()
                .join(","),
        )
    }

    pub fn take_after_compaction(&mut self) -> Option<MaybeSplitOrCompactReason> {
        self.after_compaction.take()
    }
}

impl ShardTaskManager {
    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn maybe_manual_schedule_compaction(
        &self,
        index: Arc<FTSIndex>,
        shard: &mut ShardInfo,
        range: &str,
    ) -> Result<ErrorCode> {
        if shard.can_compaction(&self.config.load().shard) {
            let mut scheduled = vec![];
            let inputs = match range {
                "delta" => shard.manifest.delta_cloned(),
                _ => anyhow::bail!("unsupported manual compaction range: {}", range),
            };
            match self
                .schedule_compaction_if_not_overlapped(
                    index,
                    &mut scheduled,
                    shard,
                    inputs,
                    CompactShardReason::Manual,
                    0,
                )
                .await
            {
                Ok(()) => Ok(ErrorCode::Success),
                Err(CompactionScheduleError::OverlapDetected {
                    shard_id,
                    input,
                    conflicting,
                }) => {
                    debug!(
                        "Manual compaction skipped due to overlap: shard_id={}, input={:?}, conflicting={:?}",
                        shard_id, input, conflicting
                    );
                    Ok(ErrorCode::Success)
                }
                Err(CompactionScheduleError::Other(e)) => Err(e),
            }
        } else {
            anyhow::bail!("can't schedule compaction");
        }
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn maybe_schedule_merge_overlap_compaction(
        &self,
        index: Arc<FTSIndex>,
        shard: &mut ShardInfo,
    ) -> Result<bool, anyhow::Error> {
        let policy = self.config.load().shard.compaction_policy.clone();
        let Some(inputs) = shard.merge_overlap_compaction_inputs(policy.as_str()) else {
            return Ok(false);
        };
        if !shard.can_compaction(&self.config.load().shard) {
            return Ok(false);
        }

        let mut scheduled = vec![];
        match self
            .schedule_compaction_if_not_overlapped(
                index,
                &mut scheduled,
                shard,
                inputs,
                CompactShardReason::MergeOverlap,
                0,
            )
            .await
        {
            Ok(()) => Ok(true),
            Err(CompactionScheduleError::OverlapDetected {
                shard_id,
                input,
                conflicting,
            }) => {
                debug!(
                    "Merge overlap compaction skipped due to confliction: shard_id={}, input={:?}, conflicting={:?}",
                    shard_id, input, conflicting
                );
                Ok(false)
            }
            Err(CompactionScheduleError::Other(e)) => Err(e),
        }
    }

    #[cfg(any(test, feature = "testexport"))]
    pub async fn debug_maybe_schedule_compaction(
        &self,
        index: Arc<FTSIndex>,
        shard: &mut ShardInfo,
    ) -> Result<Vec<CompactionState>, anyhow::Error> {
        self.maybe_schedule_compaction(index, shard).await
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn maybe_schedule_compaction(
        &self,
        index: Arc<FTSIndex>,
        shard: &mut ShardInfo,
    ) -> Result<Vec<CompactionState>, anyhow::Error> {
        let mut scheduled = vec![];
        let shard_cfg = &self.config.load().shard;
        if !shard.can_compaction(shard_cfg) {
            return Ok(scheduled);
        }

        let picked = shard.pick_uni_compaction(&shard_cfg.uni_compaction);
        if picked.is_empty() {
            return Ok(scheduled);
        }

        let remaining_iterations = shard_cfg.uni_compaction.max_compaction_iterations;
        match self
            .schedule_compaction_if_not_overlapped(
                index,
                &mut scheduled,
                shard,
                picked,
                CompactShardReason::Uni,
                remaining_iterations,
            )
            .await
        {
            Ok(()) => {}
            Err(CompactionScheduleError::OverlapDetected { .. }) => {
                // Expected during concurrent operations, skip silently
                debug!("Uni compaction skipped due to overlap");
            }
            Err(CompactionScheduleError::Other(e)) => return Err(e),
        }
        Ok(scheduled)
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn maybe_schedule_compact_by_uni_policy_with_trigger(
        &self,
        index: Arc<FTSIndex>,
        shard: &mut ShardInfo,
        trigger_frag_path: &str,
        remaining_iterations: usize,
    ) -> Result<Vec<CompactionState>, anyhow::Error> {
        let mut scheduled = vec![];
        let shard_cfg = &self.config.load().shard;
        if shard_cfg.compaction_policy.as_str() != "uni" {
            return Ok(scheduled);
        }
        if remaining_iterations == 0 {
            return Ok(scheduled);
        }
        if !shard.can_compaction(shard_cfg) {
            return Ok(scheduled);
        }

        let picked =
            shard.pick_uni_compaction_at(&shard_cfg.uni_compaction, trigger_frag_path, false);
        if picked.is_empty() {
            return Ok(scheduled);
        }

        match self
            .schedule_compaction_if_not_overlapped(
                index,
                &mut scheduled,
                shard,
                picked,
                CompactShardReason::UniChain,
                remaining_iterations,
            )
            .await
        {
            Ok(()) => {}
            Err(CompactionScheduleError::OverlapDetected { .. }) => {
                // Expected during concurrent operations, skip silently
                debug!("Uni chain compaction skipped due to overlap");
            }
            Err(CompactionScheduleError::Other(e)) => return Err(e),
        }
        Ok(scheduled)
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    #[cfg(any(test, feature = "testexport"))]
    pub async fn debug_schedule_compaction_if_not_overlapped(
        &self,
        index: Arc<FTSIndex>,
        scheduled: &mut Vec<CompactionState>,
        shard: &mut ShardInfo,
        input_fragments: Vec<SplittableFrag>,
        reason: CompactShardReason,
        remaining_iterations: usize,
    ) -> Result<(), CompactionScheduleError> {
        self.schedule_compaction_if_not_overlapped(
            index,
            scheduled,
            shard,
            input_fragments,
            reason,
            remaining_iterations,
        )
        .await
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn schedule_compaction_if_not_overlapped(
        &self,
        index: Arc<FTSIndex>,
        scheduled: &mut Vec<CompactionState>,
        shard: &mut ShardInfo,
        input_fragments: Vec<SplittableFrag>,
        reason: CompactShardReason,
        remaining_iterations: usize,
    ) -> Result<(), CompactionScheduleError> {
        if let Some(conflicting_state) = shard.find_overlapping_compaction(&input_fragments) {
            // Increment overlap metric
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["compaction_failed_sche_overlap"])
                .inc();
            // Return error with details
            return Err(CompactionScheduleError::OverlapDetected {
                shard_id: shard.shard_id,
                input: input_fragments
                    .iter()
                    .map(|f| f.frag_path.clone())
                    .collect(),
                conflicting: conflicting_state
                    .input_fragments
                    .iter()
                    .map(|f| f.frag_path.clone())
                    .collect(),
            });
        }

        SHARD_TASK_TRIGGER_REASON_COUNTER_VEC
            .with_label_values(&[reason.to_metrics_label()])
            .inc();
        let c = self
            .schedule_compaction(index, shard, input_fragments, reason, remaining_iterations)
            .await?;
        scheduled.push(c);
        Ok(())
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    async fn schedule_compaction(
        &self,
        index: Arc<FTSIndex>,
        shard: &mut ShardInfo,
        input_fragments: Vec<SplittableFrag>,
        reason: CompactShardReason,
        remaining_iterations: usize,
    ) -> Result<CompactionState, anyhow::Error> {
        let config = self.config.load();
        let mut update = shard.process_add_compaction(
            index,
            input_fragments,
            config.shard.compactor.min_split_row_count,
            config.shard.compactor.max_split_point_count,
            reason,
            remaining_iterations,
        )?;
        let (compaction_key, compaction_state) =
            if let Some((comp, comp2)) = update.compaction.as_ref() {
                (comp.clone(), comp2.clone())
            } else {
                return Err(anyhow::anyhow!("can not find compaction"));
            };
        self.meta_store
            .update_shard("schedule_compaction", shard, update.clone())
            .await?;

        update.mark_progress_cdc_persisted();
        shard.apply_update(update);
        shard.stats.observe_compaction_start();
        let compaction_work = Arc::new(CompactionWork {
            state: compaction_state.clone(),
            tick: AtomicU16::new(0),
        });

        let input_first_manifest_pos = compaction_state
            .input_fragments
            .first()
            .and_then(|frag| shard.manifest.find_pos_by_frag_path(&frag.frag_path));
        let input_last_manifest_pos = compaction_state
            .input_fragments
            .last()
            .and_then(|frag| shard.manifest.find_pos_by_frag_path(&frag.frag_path));
        info!(
            "schedule_compaction: start compaction={}, compaction_key={:?}, origin_fragments={}, input_len={}, input_(size|weighted)={}/{}, input_(first|last)_pos={:?}/{:?}, current={}",
            compaction_state.to_short_string(),
            &compaction_key,
            shard.manifest.get_fragments().len(),
            compaction_state.input_fragments.len(),
            compaction_state.get_input_size(),
            compaction_state.get_weighted_input_size(),
            input_first_manifest_pos,
            input_last_manifest_pos,
            shard.progress.compaction_state.len(),
        );
        // TODO(P1) This may fail, for example, the worker may reach its limit.
        // As a result, compaction should be aborted then by heartbeat.
        if let Err(e) = WorkerNodeMgr::send_add_compaction(
            self.wn_task_tx.clone(),
            shard.writer.clone(),
            shard.get_header(),
            compaction_state.get_compaction_state_key(),
            compaction_work,
        )
        .await
        {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["compaction_failed_worker"])
                .inc();
            return Err(e);
        }
        SHARD_EVENT_COUNTER_VEC
            .with_label_values(&["compaction_scheduled"])
            .inc();
        Ok(compaction_state)
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn apply_compaction(
        &self,
        ctx: ShardTaskCtx<FinishCompactFragResponse>,
        shard: &mut ShardInfo,
        index: Arc<FTSIndex>,
        compaction_key: CompactionStateKey,
        compaction_state: CompactionState,
        output_frags: Vec<FragMeta>,
    ) -> Option<AppliedCompactionState> {
        let mut output_frags = output_frags;
        output_frags.retain(|frag| {
            let size = frag.property.as_ref().map_or(0, |p| p.size);
            !(frag.property.is_none() && size == 0)
        });
        for frag in &output_frags {
            if let Err(e) = frag.validate() {
                warn!(
                    "apply_compaction: invalid output frag, shard={}, frag={:?}, err={:#}",
                    shard.to_long_ident(),
                    frag,
                    e,
                );
                ctx.report_err(e);
                return None;
            }
        }
        COMPACTION_INPUT_FRAGMENT_COUNT_HISTOGRAM
            .observe(compaction_state.input_fragments.len() as f64);
        COMPACTION_INPUT_SIZE_HISTOGRAM.observe(compaction_state.get_input_size() as f64);
        let output_size: u64 = output_frags.iter().map(|f| f.bytes()).sum();
        if output_size > 0 {
            let input_size = compaction_state.get_input_size();
            if input_size > 0 {
                COMPACTION_RATIO_HISTOGRAM
                    .with_label_values(&["physical"])
                    .observe(input_size as f64 / output_size as f64);
            }

            let weighted_input_size = compaction_state.get_weighted_input_size();
            if weighted_input_size > 0 {
                COMPACTION_RATIO_HISTOGRAM
                    .with_label_values(&["weighted"])
                    .observe(weighted_input_size as f64 / output_size as f64);
            }
        }
        let elapsed = compaction_state.get_elapsed_secs();
        info!(
            "apply_compaction: start shard_id={}, key={:?}, state={:?}, elapsed={:.2?}s, shard_levels={:?}, output_frags={:?}",
            shard.shard_id,
            compaction_key,
            compaction_state,
            elapsed,
            shard.manifest.describe_level(),
            output_frags,
        );
        let mut update = match shard.process_apply_compaction(
            &compaction_key,
            &compaction_state,
            output_frags.as_slice(),
        ) {
            Err(e) => {
                SHARD_EVENT_COUNTER_VEC
                    .with_label_values(&["compaction_apply_failed_check"])
                    .inc();
                ctx.report_err(e);
                return None;
            }
            Ok(u) => u,
        };

        let current_lvls = shard.manifest.current_level_sizes();
        let level_diffs: Vec<i64> = current_lvls
            .iter()
            .zip(compaction_state.current_level_states.iter())
            .map(|(cur, prev)| *cur as i64 - *prev as i64)
            .collect();
        if let Err(e) = self
            .meta_store
            .update_shard("commit_compact", shard, update.clone())
            .await
        {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["compaction_apply_failed_backend"])
                .inc();
            ctx.report_err(e);
            return None;
        }

        SHARD_EVENT_DURATION_HISTOGRAM
            .with_label_values(&["compaction"])
            .observe(elapsed);

        let previous_size = shard.get_total_estimated_bytes();
        let previous_rows = shard.get_total_estimated_rows();
        update.mark_progress_cdc_persisted();
        shard.apply_update(update);
        let after_size = shard.get_total_estimated_bytes();
        let after_rows = shard.get_total_estimated_rows();

        self.helper_after_compaction(true, shard, index).await;
        let mut after_compaction = None;
        if compaction_state.remaining_iterations > 1 {
            let remaining_iterations = compaction_state.remaining_iterations - 1;
            let trigger_frag_path = output_frags.last().map(|f| f.frag_path.clone());
            if let Some(trigger_frag_path) = trigger_frag_path {
                let config = self.config.load();
                let picked = shard.pick_uni_compaction_at(
                    &config.shard.uni_compaction,
                    trigger_frag_path.as_str(),
                    false,
                );
                if !picked.is_empty() {
                    after_compaction = Some(MaybeSplitOrCompactReason::AfterCompaction {
                        trigger_frag_path,
                        remaining_iterations,
                    });
                }
            }
        }
        let applied_compaction = AppliedCompactionState::new(
            compaction_state.clone(),
            output_frags.clone(),
            after_compaction,
        );

        info!(
            "apply_compaction: finished compaction={}, shard_id={}, elapsed={} s, shard_levels={}, desc: ({}), level_diffs={:?}, shard_size={}->{}, shard_rows={}->{}, remain_compactions={}",
            applied_compaction.state.to_short_string(),
            shard.shard_id,
            applied_compaction.state.get_elapsed_secs(),
            shard.manifest.describe_level(),
            applied_compaction.describe_compaction(),
            level_diffs,
            previous_size,
            after_size,
            previous_rows,
            after_rows,
            shard.progress.compaction_state.len(),
        );

        ctx.report_ok(FinishCompactFragResponse {
            status: tici_grpc::ErrorCode::Success.into(),
        });

        Some(applied_compaction)
    }

    // NOTE Do not directly call this, you have to schedule a ShardTask.
    pub(super) async fn abort_compaction(
        &self,
        shard_info: &mut ShardInfo,
        compaction_key: CompactionStateKey,
        index: Arc<FTSIndex>,
    ) {
        if shard_info
            .progress
            .compaction_state
            .remove(&compaction_key)
            .is_none()
        {
            warn!(
                "abort_compaction: failed to find compaction, shard={}, compaction_key={:?}",
                shard_info.to_long_ident(),
                compaction_key
            );
        } else {
            self.helper_after_compaction(false, shard_info, index).await;
            info!(
                "abort_compaction: compaction aborted, shard={}, compaction_key={:?}, len={}",
                shard_info.to_long_ident(),
                compaction_key,
                shard_info.progress.compaction_state.len(),
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use tici_common::config::S3Config;
    use tici_grpc::FragMeta;
    use tici_shard::test_utils::test_mock_frag_storage_refs;

    use super::*;

    fn frag_path(path: &str) -> String {
        format!("{path}/f_1")
    }

    fn make_test_frag(path: &str) -> SplittableFrag {
        let frag = FragMeta::new(
            0,
            frag_path(path),
            tici_grpc::FragProperty {
                size: 1,
                split_handles: vec![vec![]],
                ..Default::default()
            },
            test_mock_frag_storage_refs(),
            1,
        )
        .unwrap();
        SplittableFrag::new(frag, 1, 1)
    }

    /// Create a minimal CompactionState for testing overlap_with()
    fn make_test_compaction_state(frag_paths: Vec<&str>) -> CompactionState {
        let input_fragments: Vec<SplittableFrag> =
            frag_paths.iter().map(|path| make_test_frag(path)).collect();

        let default_s3 = S3Config::default();
        CompactionState {
            input_fragments,
            shard_id: 1,
            index_key: IndexKey::new_nullspace(1, 1),
            range: HandleRange::new(vec![1], vec![2]),
            epoch: 1,
            seq: 1,
            start_time: SystemTime::now(),
            worker: "test_worker".to_string(),
            upload_location: S3Location {
                bucket: default_s3.bucket,
                prefix: default_s3.prefix,
            },
            level: Some(0),
            current_level_states: [0, 0],
            min_split_row_count: 0,
            max_split_point_count: 0,
            reason: CompactShardReason::Manual,
            remaining_iterations: 0,
            compact_type: CompactType::Delta,
            index: Arc::new(crate::meta::writer::FTSIndex::new_for_test()),
        }
    }

    #[test]
    fn test_overlap_with_non_overlapping() {
        // [F1, F2] vs [F3, F4] → no overlap (no shared endpoints)
        let state = make_test_compaction_state(vec!["F1", "F2"]);
        let input = vec![make_test_frag("F3"), make_test_frag("F4")];
        assert!(!state.overlap_with(&input));
    }

    #[test]
    fn test_overlap_with_endpoint_match_first() {
        // [F1, F2] vs [F1, F3] → overlap (F1 is first endpoint of state)
        let state = make_test_compaction_state(vec!["F1", "F2"]);
        let input = vec![make_test_frag("F1"), make_test_frag("F3")];
        assert!(state.overlap_with(&input));
    }

    #[test]
    fn test_overlap_with_endpoint_match_last() {
        // [F1, F2] vs [F2, F3] → overlap (F2 is last endpoint of state)
        let state = make_test_compaction_state(vec!["F1", "F2"]);
        let input = vec![make_test_frag("F2"), make_test_frag("F3")];
        assert!(state.overlap_with(&input));
    }

    #[test]
    fn test_overlap_with_middle_fragment_no_match() {
        // [F1, F2, F3] vs [F2] → no overlap
        // Endpoint-only check: state endpoints are F1 and F3, F2 is not an endpoint
        let state = make_test_compaction_state(vec!["F1", "F2", "F3"]);
        let input = vec![make_test_frag("F2")];
        assert!(!state.overlap_with(&input));
    }

    #[test]
    fn test_overlap_with_endpoint_in_longer_range() {
        // [F1, F2, F3] vs [F1] or [F3] → overlap (matches endpoints)
        let state = make_test_compaction_state(vec!["F1", "F2", "F3"]);

        let input_first = vec![make_test_frag("F1")];
        assert!(state.overlap_with(&input_first));

        let input_last = vec![make_test_frag("F3")];
        assert!(state.overlap_with(&input_last));
    }

    #[test]
    fn test_overlap_with_same_range() {
        // [F1, F2] vs [F1, F2] → overlap (both endpoints match)
        let state = make_test_compaction_state(vec!["F1", "F2"]);
        let input = vec![make_test_frag("F1"), make_test_frag("F2")];
        assert!(state.overlap_with(&input));
    }

    #[test]
    fn test_overlap_with_empty_input() {
        // [F1] vs [] → no overlap (empty input has no fragments to check)
        let state = make_test_compaction_state(vec!["F1"]);
        let input: Vec<SplittableFrag> = vec![];
        assert!(!state.overlap_with(&input));
    }

    #[test]
    fn test_overlap_single_fragment_match() {
        // [F5] vs [F5] → overlap (single fragment is both first and last endpoint)
        let state = make_test_compaction_state(vec!["F5"]);
        let input = vec![make_test_frag("F5")];
        assert!(state.overlap_with(&input));
    }
}