use std::sync::Arc;

use log::{debug, info, warn};
use tici_grpc::{AppendFragMetaResponse, ErrorCode, FinishCompactFragResponse};
use tici_system::readable::ReadableSize;
use tokio::sync::{RwLock, mpsc::UnboundedReceiver};

use crate::{
    HandleRange, RangeBound, ShardInfo, SplittableFrag,
    meta::{
        metrics::{
            SHARD_EVENT_COUNTER_VEC, SHARD_TASK_BY_RUNNER_COUNTER_VEC, SHARD_TASK_QUEUE_HISTOGRAM,
        },
        util::BgRunner,
        writer::{CompactionStateKey, FTSIndex, ShardSplitReason, ShardTaskManager, shard_task::*},
    },
};

pub enum ShardSplitResult {
    NoNeed(u64, u64),
    CannotSplit,
    Invalid,
    Yield(Vec<(CompactionStateKey, f64)>),
    Ok(Vec<ShardInfo>),
}

pub enum ShardMergeResult {
    NoNeed,
    Merged(Arc<RwLock<ShardInfo>>),
    NeedMergeOverlapCompaction(Box<(SplittableFrag, SplittableFrag)>),
}

#[derive(Clone, Debug)]
pub enum LocalRunnerRouter {
    Sender(Vec<tokio::sync::mpsc::UnboundedSender<ShardTask>>),
    Receiver(u64),
}

impl LocalRunnerRouter {
    pub fn ident(&self, tag: &str) -> String {
        match &self {
            LocalRunnerRouter::Sender(_) => {
                if tag.is_empty() {
                    "Sche[G]".to_string()
                } else {
                    format!("Sche[G]({})", tag)
                }
            }
            LocalRunnerRouter::Receiver(id) => {
                if tag.is_empty() {
                    format!("Sche[L{}]", id)
                } else {
                    format!("Sche[L{}]({})", id, tag)
                }
            }
        }
    }

    pub fn is_global(&self) -> bool {
        matches!(self, LocalRunnerRouter::Sender(_))
    }
}

pub struct ShardTaskRunner {
    _task_runner: BgRunner,
    router: LocalRunnerRouter,
}

#[derive(Debug)]
pub struct ShardTaskRunnerCtx {
    mgr: Arc<ShardTaskManager>,
    router: LocalRunnerRouter,
}

impl ShardTaskRunnerCtx {
    pub fn new(mgr: Arc<ShardTaskManager>, router: LocalRunnerRouter) -> Arc<Self> {
        Arc::new(Self { mgr, router })
    }
}

fn build_split_ranges(shard: &HandleRange, split_keys: &[Vec<u8>]) -> Vec<HandleRange> {
    let mut new_ranges = Vec::with_capacity(split_keys.len() + 1);
    let mut start = shard.encoded_start();
    for split_key in split_keys {
        new_ranges.push(HandleRange::new(start, split_key.clone()));
        start = split_key.clone();
    }
    new_ranges.push(HandleRange::new(start, shard.encoded_end()));
    new_ranges
}

async fn handle_manual_compaction(mgr: &ShardTaskManager, task: ManualCompactionTask) {
    match mgr
        .get_index_and_shard(&task.header.index_key, &task.header.start_key)
        .await
    {
        Ok((ind, c)) => {
            match mgr
                .maybe_manual_schedule_compaction(ind, &mut *c.write().await, &task.range)
                .await
            {
                Ok(_) => {
                    info!(
                        "manual_compaction: succeed, header={}",
                        task.header.to_long_ident()
                    );
                    task.ctx.report_ok(ErrorCode::Success);
                }
                Err(e) => {
                    info!(
                        "manual_compaction: failed, header={}, e={:?}",
                        task.header.to_long_ident(),
                        e
                    );
                    task.ctx.report_ok(ErrorCode::ShardNotFound);
                }
            }
        }
        Err(e) => {
            info!(
                "manual_compaction: can't find shard, maybe is already dismissed, header={}, e={:?}",
                task.header.to_long_ident(),
                e
            );
            task.ctx.report_ok(ErrorCode::ShardNotFound);
        }
    }
}

// returns `(succeed, stop_next_compaction)`.
async fn handle_split(
    mgr: &ShardTaskManager,
    router: &LocalRunnerRouter,
    shard: Arc<RwLock<ShardInfo>>,
    index: Arc<FTSIndex>,
    reason: ShardSplitReason,
) -> (bool, bool) {
    let (shard_ident, shard_mb, shard_rows) = {
        let shard = shard.read().await;
        (
            shard.to_long_ident(),
            ReadableSize(shard.manifest.get_total_estimated_bytes()).as_mb_f64(),
            shard.manifest.get_total_estimated_rows(),
        )
    };
    if mgr.config.load().shard.enable_split {
        if !mgr.is_topology_change_enabled() && !matches!(reason, ShardSplitReason::Manual) {
            warn!(
                "{}: rejected because restore is still rebuilding topology state, shard={}, reason={:?}",
                router.ident("split"),
                shard_ident,
                reason,
            );
            return (false, false);
        }
        if index.is_topology_frozen() && !matches!(reason, ShardSplitReason::Manual) {
            warn!(
                "{}: rejected due to topology freeze, {}, reason={:?}",
                router.ident("split"),
                shard_ident,
                reason,
            );
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["split_failed_freeze"])
                .inc();
            return (false, false);
        }
        match mgr.maybe_split_shards(shard, reason, 0).await {
            Ok(ShardSplitResult::NoNeed(size, rows)) => {
                debug!(
                    "{}: no need, {}, size: {}/{}, rows={}",
                    router.ident("split"),
                    shard_ident,
                    size,
                    mgr.config.load().shard.get_split_size(),
                    rows
                );
                index.stats.observe_split_no_need();
                (false, false)
            }
            Ok(ShardSplitResult::Yield(v)) => {
                // Yield because there is an ongoing compaction.
                info!(
                    "{}: yield, {}, current_size={:.2}MB, current_rows={}, exist_compactions={:?}",
                    router.ident("split"),
                    shard_ident,
                    shard_mb,
                    shard_rows,
                    v,
                );
                index.stats.observe_split_yield();
                (false, true)
            }
            Ok(ShardSplitResult::CannotSplit) | Ok(ShardSplitResult::Invalid) => {
                // Don't have a split point, or the split point is invalid.
                info!(
                    "{}: can not split / invalid split, {}",
                    router.ident("split"),
                    shard_ident,
                );
                index.stats.observe_split_unable();
                (false, false)
            }
            Ok(ShardSplitResult::Ok(_)) => {
                // If splitted, the next compaction won't succeed at all.
                // So skip the compaction on the old shard.
                (true, true)
            }
            Err(e) => {
                warn!(
                    "{}: failed to split shard, {}, e={:?}",
                    router.ident("split"),
                    shard_ident,
                    e
                );
                (false, false)
            }
        }
    } else {
        (false, false)
    }
}

async fn handle_resche(
    mgr: &ShardTaskManager,
    task: NeedRescheduleTask,
    router: &LocalRunnerRouter,
) {
    match mgr
        .get_shard(&task.header.index_key, &task.header.start_key)
        .await
    {
        Ok(shard) => {
            let reason = task.reason.clone();
            if let Err(e) = mgr
                .resche_shards(
                    &mut *shard.write().await,
                    &task.header.index_key,
                    task.designated_writer,
                    task.reason,
                )
                .await
            {
                info!(
                    "{}: failed to reschedule, header={}, reason={:?}, e={:?}",
                    router.ident("resche"),
                    task.header.to_long_ident(),
                    reason,
                    e
                );
                task.ctx.report_err(e);
            } else {
                task.ctx.report_ok(());
            };
        }
        Err(e) => {
            // This shard may be already scheduled away.
            info!(
                "{}: failed to find shard, header={:?}, e={:?}",
                router.ident("reche"),
                task.header.to_long_ident(),
                e
            );
            task.ctx.report_err(e);
        }
    }
}

async fn handle_explicit_split(
    mgr: &ShardTaskManager,
    router: &LocalRunnerRouter,
    task: ExplicitSplitTask,
) {
    if !mgr.is_topology_change_enabled() {
        let err = anyhow::anyhow!(
            "{}: rejected because restore is still rebuilding topology state, header={}, reason={:?}",
            router.ident("explicit_split"),
            task.header.to_long_ident(),
            task.reason,
        );
        warn!("{:#}", err);
        task.ctx.report_err(err);
        return;
    }
    if task.index.is_topology_frozen()
        && !matches!(task.reason, ExplicitSplitReason::PresplitImport)
    {
        let err = anyhow::anyhow!(
            "{}: rejected due to topology freeze, header={}, reason={:?}",
            router.ident("explicit_split"),
            task.header.to_long_ident(),
            task.reason,
        );
        warn!("{:#}", err);
        task.ctx.report_err(err);
        return;
    }

    let Some(shard) = task.index.get_shard(&task.header.start_key).await else {
        info!(
            "{}: skip explicit split because shard is gone, header={}, reason={:?}",
            router.ident("explicit_split"),
            task.header.to_long_ident(),
            task.reason,
        );
        task.ctx.report_ok(ExplicitSplitResult { split_count: 0 });
        return;
    };

    let guard = shard.read().await;
    if guard.shard_id != task.header.shard_id {
        info!(
            "{}: skip explicit split because shard_id changed, expected={}, got={}, header={}, reason={:?}",
            router.ident("explicit_split"),
            task.header.shard_id,
            guard.shard_id,
            task.header.to_long_ident(),
            task.reason,
        );
        task.ctx.report_ok(ExplicitSplitResult { split_count: 0 });
        return;
    }

    let local_split_keys = task
        .split_keys
        .iter()
        .filter(|split_key| {
            split_key.as_slice() > guard.range.encoded_start().as_slice()
                && split_key.as_slice() < guard.range.encoded_end().as_slice()
        })
        .cloned()
        .collect::<Vec<_>>();
    if local_split_keys.is_empty() {
        info!(
            "{}: skip explicit split because no local split keys remain, header={}, reason={:?}",
            router.ident("explicit_split"),
            task.header.to_long_ident(),
            task.reason,
        );
        task.ctx.report_ok(ExplicitSplitResult { split_count: 0 });
        return;
    }

    let new_ranges = build_split_ranges(&guard.range, &local_split_keys);
    let split_count = (new_ranges.len() - 1) as u32;

    match mgr.explicit_split_shards(guard, new_ranges, 0).await {
        Ok(_) => {
            info!(
                "{}: explicit split succeed, header={}, split_count={}, reason={:?}",
                router.ident("explicit_split"),
                task.header.to_long_ident(),
                split_count,
                task.reason,
            );
            task.ctx.report_ok(ExplicitSplitResult { split_count });
        }
        Err(e) => {
            warn!(
                "{}: explicit split failed, header={}, split_keys={}, reason={:?}, err={:?}",
                router.ident("explicit_split"),
                task.header.to_long_ident(),
                local_split_keys.len(),
                task.reason,
                e,
            );
            task.ctx.report_err(e);
        }
    }
}

async fn handle_merge(mgr: &ShardTaskManager, router: &LocalRunnerRouter, task: MergeTask) {
    if !mgr.is_topology_change_enabled() && !matches!(task.reason, MergeShardReason::Manual) {
        warn!(
            "{}: rejected because restore is still rebuilding topology state, index={:?}, table_id={}, shard_id={}",
            router.ident("merge"),
            task.index.index_key(),
            task.index.table_id(),
            task.shard_id
        );
        return;
    }
    if task.index.is_topology_frozen() && !matches!(task.reason, MergeShardReason::Manual) {
        warn!(
            "{}: rejected due to topology freeze, index={:?}, table_id={}, shard_id={}",
            router.ident("merge"),
            task.index.index_key(),
            task.index.table_id(),
            task.shard_id
        );
        SHARD_EVENT_COUNTER_VEC
            .with_label_values(&["merge_failed_freeze"])
            .inc();
        return;
    }
    match mgr
        .get_shard(&task.index.index_key(), &task.start_key)
        .await
    {
        Ok(c) => {
            let (index_key, end_key) = {
                let shard_info = c.read().await;
                if shard_info.shard_id != task.shard_id {
                    // Reject after split -> merge -> split
                    return;
                }
                (
                    shard_info.index_key.clone(),
                    shard_info.range.sortable_end(),
                )
            };
            // TODO(P2) Need to support merge into the left?
            if let Ok(shard_on_the_right) = mgr.get_shard(&index_key, &end_key).await {
                let right_shard_id = {
                    let other = shard_on_the_right.read().await;
                    other.shard_id
                };
                let merge_result = mgr
                    .maybe_merge_shards(vec![c.clone(), shard_on_the_right.clone()], &task.reason)
                    .await;

                match merge_result {
                    Ok(ShardMergeResult::NoNeed) => {}
                    Ok(ShardMergeResult::Merged(_merged)) => {}
                    Ok(ShardMergeResult::NeedMergeOverlapCompaction(_inputs)) => {
                        let index = match mgr.get_index(&index_key).await {
                            Ok(index) => index,
                            Err(e) => {
                                warn!(
                                    "{}: merge overlap compaction skipped, err={:?}",
                                    router.ident("merge"),
                                    e
                                );
                                return;
                            }
                        };

                        let mut shard_info = c.write().await;
                        if shard_info.shard_id != task.shard_id {
                            return;
                        }
                        if let Err(e) = mgr
                            .maybe_schedule_merge_overlap_compaction(index.clone(), &mut shard_info)
                            .await
                        {
                            warn!(
                                "{}: merge overlap compaction failed, shard={}, err={:?}",
                                router.ident("merge"),
                                shard_info.to_long_ident(),
                                e
                            );
                        }
                        drop(shard_info);

                        let mut right_shard = shard_on_the_right.write().await;
                        if right_shard.shard_id != right_shard_id {
                            return;
                        }
                        if let Err(e) = mgr
                            .maybe_schedule_merge_overlap_compaction(index, &mut right_shard)
                            .await
                        {
                            warn!(
                                "{}: merge overlap compaction failed, shard={}, err={:?}",
                                router.ident("merge"),
                                right_shard.to_long_ident(),
                                e
                            );
                        }
                    }
                    Err(e) => {
                        warn!("{}: error={:?}", router.ident("merge"), e);
                    }
                }
            }
        }
        Err(_) => {
            // This could happen, for example, the shard has
            // been merged alreay, or is has been splitted
            // before.
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["merge_failed_not_found"])
                .inc();
        }
    }
}

async fn handle_compaction(
    mgr: &ShardTaskManager,
    router: &LocalRunnerRouter,
    index: Arc<FTSIndex>,
    shard: &mut ShardInfo,
    reason: MaybeSplitOrCompactReason,
) -> bool {
    // We handle split first, because:
    // - split is faster than compaction
    // - currently, all compaction will fail, after a split
    // - if split is delayed, the size of the shard could be very large
    let cfg = mgr.config.load();
    if !cfg.shard.auto_compaction {
        return false;
    }
    if index.is_finished_append_base() && shard.manifest.has_base() {
        let base_len = shard.manifest.base_slice().len();
        match shard.switch_base_to_delta(&mgr.meta_store).await {
            Ok(true) => {
                info!(
                    "{}: {}, switched base to delta before compaction, reason={:?}, base_len={}",
                    router.ident("compact_sche"),
                    shard.to_long_ident(),
                    reason,
                    base_len,
                );
            }
            Ok(false) => {}
            Err(e) => {
                info!(
                    "{}: {}, failed to switch base to delta before compaction, reason={:?}, error={:?}",
                    router.ident("compact_sche"),
                    shard.to_long_ident(),
                    reason,
                    e
                );
                return false;
            }
        }
    }
    match mgr.maybe_schedule_compaction(index, shard).await {
        Ok(res) => {
            if res.is_empty() {
                info!(
                    "{}: {}, Rejected, policy=uni, compaction_state={:?}, reason={:?}",
                    router.ident("compact_sche"),
                    shard.to_long_ident(),
                    shard.progress.collect_compaction_states(),
                    reason
                );
                false
            } else {
                info!(
                    "{}: {}, OK, policy=uni, compaction_state={:?}, compactions={:?}, reason={:?}",
                    router.ident("compact_sche"),
                    shard.to_long_ident(),
                    shard.progress.collect_compaction_states(),
                    res,
                    reason
                );
                true
            }
        }
        Err(e) => {
            info!(
                "{}: {}, error, policy=uni, reason={:?}, compaction_state={:?}, error={:?}",
                router.ident("compact_sche"),
                shard.to_long_ident(),
                reason,
                shard.progress.collect_compaction_states(),
                e
            );
            false
        }
    }
}

async fn handle_compaction_after_compaction(
    mgr: &ShardTaskManager,
    router: &LocalRunnerRouter,
    index: Arc<FTSIndex>,
    shard: &mut ShardInfo,
    trigger_frag_path: String,
    remaining_iterations: usize,
) -> bool {
    let cfg = mgr.config.load();
    if !cfg.shard.auto_compaction {
        return false;
    }
    match mgr
        .maybe_schedule_compact_by_uni_policy_with_trigger(
            index,
            shard,
            trigger_frag_path.as_str(),
            remaining_iterations,
        )
        .await
    {
        Ok(res) => {
            if res.is_empty() {
                info!(
                    "{}: {}, Rejected, policy=uni_chain, compaction_state={:?}, trigger={}, remaining={}",
                    router.ident("compact_chain"),
                    shard.to_long_ident(),
                    shard.progress.collect_compaction_states(),
                    trigger_frag_path,
                    remaining_iterations
                );
                false
            } else {
                info!(
                    "{}: {}, OK, policy=uni_chain, compaction_state={:?}, compactions={:?}, trigger={}, remaining={}",
                    router.ident("compact_chain"),
                    shard.to_long_ident(),
                    shard.progress.collect_compaction_states(),
                    res,
                    trigger_frag_path,
                    remaining_iterations
                );
                true
            }
        }
        Err(e) => {
            info!(
                "{}: {}, error, policy=uni_chain, trigger={}, remaining={}, compaction_state={:?}, error={:?}",
                router.ident("compact_chain"),
                shard.to_long_ident(),
                trigger_frag_path,
                remaining_iterations,
                shard.progress.collect_compaction_states(),
                e
            );
            false
        }
    }
}

async fn handle_check_after_append(
    mgr: &ShardTaskManager,
    task: MaybeSplitOrCompactTask,
    router: &LocalRunnerRouter,
) {
    match mgr
        .get_index_and_shard(&task.header.index_key, &task.header.start_key)
        .await
    {
        Ok((index, shard)) => {
            match task.reason {
                MaybeSplitOrCompactReason::StallTooManyFrags => {
                    // If there are too many fragments, we must stall write for compactions,
                    // otherwise, we will get "entry too large" error.
                    handle_compaction(
                        mgr,
                        router,
                        index.clone(),
                        &mut *shard.write().await,
                        task.reason,
                    )
                    .await;
                }
                MaybeSplitOrCompactReason::AfterAppend => {
                    let (_splitted, stop_next_compaction) = handle_split(
                        mgr,
                        router,
                        shard.clone(),
                        index.clone(),
                        ShardSplitReason::NewAppend,
                    )
                    .await;
                    // If the shard needs to be splitted but yields because there is an ongoing
                    // compaction, we need to wait the compaction finish. And
                    // new compaction will not be scheduled.
                    if !stop_next_compaction {
                        handle_compaction(
                            mgr,
                            router,
                            index.clone(),
                            &mut *shard.write().await,
                            task.reason,
                        )
                        .await;
                    } else {
                        index.stats.observe_compaction_yield();
                    }
                    task.ctx.report_ok(());
                }
                MaybeSplitOrCompactReason::AfterCompaction {
                    trigger_frag_path,
                    remaining_iterations,
                } => {
                    handle_compaction_after_compaction(
                        mgr,
                        router,
                        index.clone(),
                        &mut *shard.write().await,
                        trigger_frag_path,
                        remaining_iterations,
                    )
                    .await;
                    task.ctx.report_ok(());
                }
                MaybeSplitOrCompactReason::ManualSplit => {
                    info!("manual_split: shard={}", shard.read().await.to_long_ident());
                    let (ok, _) = handle_split(
                        mgr,
                        router,
                        shard.clone(),
                        index.clone(),
                        ShardSplitReason::Manual,
                    )
                    .await;
                    if ok {
                        task.ctx.report_ok(());
                    } else {
                        info!(
                            "manual_split: failed, shard={}",
                            shard.read().await.to_long_ident()
                        );
                        task.ctx.report_err(anyhow::anyhow!("maunal_split failed"));
                    }
                }
            };
        }
        Err(e) => {
            warn!(
                "{}: failed to find shard, header={}, e={:?}",
                router.ident("after_append"),
                task.header.to_long_ident(),
                e
            );
            task.ctx.report_err(e);
        }
    }
}

impl ShardTaskRunner {
    // TODO(P2) Introduce some generic type
    fn observe_task_queue(task: &ShardTask) {
        let (metric_label, queued_secs) = match task {
            ShardTask::Hup => return,
            ShardTask::CompactionFinish(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::AppendDeltaFrag(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::AppendBaseFrag(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::ExplicitSplit(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::MaybeSplitOrCompact(task) => {
                (task.ctx.metric_label(), task.ctx.queued_secs())
            }
            ShardTask::NeedReschedule(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::CompactionAborted(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::TryMerge(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
            ShardTask::ManualCompaction(task) => (task.ctx.metric_label(), task.ctx.queued_secs()),
        };
        let Some(metric_label) = metric_label else {
            return;
        };
        SHARD_TASK_QUEUE_HISTOGRAM
            .with_label_values(&[metric_label])
            .observe(queued_secs);
    }

    pub fn new(
        shard_task_rx: UnboundedReceiver<ShardTask>,
        router: LocalRunnerRouter,
    ) -> (Self, tokio::sync::oneshot::Sender<Arc<ShardTaskRunnerCtx>>) {
        let ident = router.ident("");
        let (task_runner, stx) = BgRunner::wrapper(
            ident,
            shard_task_rx,
            |ctx: Arc<ShardTaskRunnerCtx>, x: ShardTask| async move {
                if Self::handle_task(ctx.mgr.as_ref(), x, &ctx.router).await {
                    return true;
                }
                false
            },
            |ctx: Arc<ShardTaskRunnerCtx>| {
                ctx.mgr
                    .stopped
                    .store(true, std::sync::atomic::Ordering::Relaxed);
            },
        );

        (
            Self {
                _task_runner: task_runner,
                router,
            },
            stx,
        )
    }

    pub fn router_cloned(&self) -> LocalRunnerRouter {
        self.router.clone()
    }

    pub fn try_dispatch_to_local_runner(
        router: &LocalRunnerRouter,
        task: ShardTask,
    ) -> Option<ShardTask> {
        match router {
            LocalRunnerRouter::Sender(s) => {
                if s.is_empty() {
                    // No local runner
                    Some(task)
                } else {
                    let local_id = task.get_local_id(s.len());
                    if let Some(local_id) = local_id {
                        if let Err(e) = s[local_id].send(task) {
                            warn!(
                                "dispatch: failed to dispath to {}, e={:?}, task={:?}",
                                local_id, e, e.0
                            );
                        }
                        None
                    } else {
                        // Must be handled
                        Some(task)
                    }
                }
            }
            LocalRunnerRouter::Receiver(_) => Some(task),
        }
    }

    async fn handle_task(
        mgr: &ShardTaskManager,
        task: ShardTask,
        router: &LocalRunnerRouter,
    ) -> bool {
        let task = if let Some(task) = Self::try_dispatch_to_local_runner(router, task) {
            // - I am the local runner, and the task is sent from global.
            // - I am the golbal runner, and the task can't be handled by local.
            task
        } else {
            return false;
        };
        Self::observe_task_queue(&task);
        let runner_id = match router {
            LocalRunnerRouter::Sender(_) => "G".to_string(),
            LocalRunnerRouter::Receiver(id) => id.to_string(),
        };
        SHARD_TASK_BY_RUNNER_COUNTER_VEC
            .with_label_values(&[runner_id.as_str(), "tot"])
            .inc();

        match task {
            ShardTask::Hup => {
                info!("{}: runner hup", router.ident("hup"));
                if let LocalRunnerRouter::Sender(v) = router {
                    for r in v.iter() {
                        if let Err(e) = r.send(ShardTask::Hup) {
                            warn!("{}: send hup error, e={:?}", router.ident("hup"), e);
                        }
                    }
                }
                return true;
            }
            ShardTask::MaybeSplitOrCompact(task) => {
                handle_check_after_append(mgr, task, router).await
            }
            ShardTask::CompactionAborted(task) => {
                match mgr
                    .get_index_and_shard(&task.header.index_key, &task.header.start_key)
                    .await
                {
                    Ok((index, shard)) => {
                        let mut s = shard.write().await;
                        mgr.abort_compaction(&mut s, task.key, index).await;
                        task.ctx.report_ok(());
                    }
                    Err(e) => {
                        warn!(
                            "{} failed to find shard, header={}, e={:?}",
                            router.ident("compact_abort"),
                            task.header.to_long_ident(),
                            e
                        );
                        task.ctx.report_err(e);
                    }
                }
            }
            ShardTask::CompactionFinish(task) => {
                let start_key = RangeBound::make_start(task.header.start_key.clone());
                if let Some(s) = task.index.get_shard(&start_key).await {
                    let mut shard = s.write().await;
                    let applied_compaction = mgr
                        .apply_compaction(
                            task.ctx,
                            &mut shard,
                            task.index.clone(),
                            task.compaction_key,
                            task.compaction_state,
                            task.output_frags,
                        )
                        .await;
                    if let Some(mut applied_compaction) = applied_compaction {
                        let compaction_speed = applied_compaction.compute_compaction_speed();
                        let append_speed = shard.stats.get_append_bps();
                        let should_split = compaction_speed
                            * mgr.config.load().shard.compactor.compaction_speed_factor
                            < append_speed;
                        let after_compaction = applied_compaction.take_after_compaction();
                        drop(shard);

                        let mut stop_next_compaction = false;
                        if should_split {
                            info!(
                                "{}: append heavier than compaction",
                                router.ident("compaction_apply"),
                            );
                            let (_, stop) = handle_split(
                                mgr,
                                router,
                                s.clone(),
                                task.index.clone(),
                                ShardSplitReason::WritePressure,
                            )
                            .await;
                            stop_next_compaction = stop;
                        }

                        // If a compaction is ongoing, the split could yield, so we need to check if
                        // we need to split first.
                        if !stop_next_compaction {
                            if let Some(MaybeSplitOrCompactReason::AfterCompaction {
                                trigger_frag_path,
                                remaining_iterations,
                            }) = after_compaction
                            {
                                if let Some(current) = task.index.get_shard(&start_key).await {
                                    let mut shard = current.write().await;
                                    handle_compaction_after_compaction(
                                        mgr,
                                        router,
                                        task.index.clone(),
                                        &mut shard,
                                        trigger_frag_path,
                                        remaining_iterations,
                                    )
                                    .await;
                                }
                            }
                        }
                    } else {
                        drop(shard);
                    }
                } else {
                    // It's common case, when the shard has been splitted/merged in the meantime.
                    info!(
                        "{}: can't find shard={:?}",
                        router.ident("compaction_apply"),
                        &task.compaction_key
                    );
                    task.ctx.report_ok(FinishCompactFragResponse {
                        status: tici_grpc::ErrorCode::ShardNotFound.into(),
                    });
                };
            }
            ShardTask::NeedReschedule(task) => {
                handle_resche(mgr, task, router).await;
            }
            ShardTask::AppendDeltaFrag(task) => {
                fail::fail_point!("append_delta_frag_meta_before_get_shard");
                if let Some(s) = task.index.get_shard(&task.header.start_key).await {
                    let mut guard = s.write().await;
                    // NOTE: a stale task can arrive after the shard is merged. In this can, we can
                    // just ignore this task  as the merged write will re-sync
                    // cdc logs from the smallest progess index.
                    if guard.shard_id != task.header.shard_id {
                        info!(
                            "append_frag_meta(delta): shard_id changed, expected={}, got={}, header={}",
                            task.header.shard_id,
                            guard.shard_id,
                            task.header.to_long_ident()
                        );
                        task.ctx.report_ok((
                            AppendFragMetaResponse {
                                status: ErrorCode::ShardNotFound.into(),
                            },
                            None,
                        ));
                    } else {
                        mgr.append_delta_frag_meta(task.ctx, &mut guard, task.index, task.req)
                            .await;
                    }
                } else {
                    info!(
                        "append_frag_meta(delta): can't find shard, maybe is already dismissed, header={:?}",
                        task.header.to_long_ident()
                    );
                    task.ctx.report_ok((
                        AppendFragMetaResponse {
                            status: tici_grpc::ErrorCode::ShardNotFound.into(),
                        },
                        None,
                    ));
                };
            }
            ShardTask::AppendBaseFrag(task) => {
                fail::fail_point!("append_base_frag_meta_before_get_shard");
                if let Some(s) = task.index.get_shard(&task.header.start_key).await {
                    let mut guard = s.write().await;
                    if guard.shard_id != task.header.shard_id {
                        info!(
                            "append_frag_meta(base): shard_id changed, expected={}, got={}, header={}",
                            task.header.shard_id,
                            guard.shard_id,
                            task.header.to_long_ident()
                        );
                        task.ctx.report_ok((
                            AppendFragMetaResponse {
                                status: ErrorCode::ShardNotFound.into(),
                            },
                            None,
                        ));
                    } else {
                        mgr.append_base_frag_meta(
                            task.ctx,
                            &mut guard,
                            task.index,
                            task.frags.as_slice(),
                        )
                        .await;
                    }
                } else {
                    info!(
                        "append_frag_meta(base): can't find shard, maybe is already dismissed, header={}",
                        task.header.to_long_ident()
                    );
                    task.ctx.report_ok((
                        AppendFragMetaResponse {
                            status: tici_grpc::ErrorCode::ShardNotFound.into(),
                        },
                        None,
                    ));
                };
            }
            ShardTask::ExplicitSplit(task) => {
                handle_explicit_split(mgr, router, task).await;
            }
            ShardTask::TryMerge(task) => handle_merge(mgr, router, task).await,
            ShardTask::ManualCompaction(task) => handle_manual_compaction(mgr, task).await,
        }
        false
    }
}