// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

#[cfg(any(test, feature = "testexport"))]
use std::collections::HashSet;
use std::{
    collections::VecDeque,
    fmt,
    fmt::Debug,
    sync::Arc,
    time::{Duration, Instant as StdInstant},
};

use cloud_encryption::EncryptionState;
use health_controller::{InspectFactor, LatencyInspector};
use kvengine::{
    FilePrepareType,
    table::{TxnFile, schema_file::SchemaFile},
};
use kvenginepb::TxnFileRef;
use kvproto::{
    kvrpcpb::{ExtraOp as TxnExtraOp, KeyRange},
    metapb, pdpb,
    raft_cmdpb::{RaftCmdRequest, RaftCmdResponse},
    raft_serverpb::{self as rspb, RaftMessage},
};
use pd_client::{BucketMeta, BucketStat};
use protobuf::Message;
use raft_proto::eraftpb;
use raftstore::store::{
    fsm::ChangeObserver,
    local_metrics::{RaftMetrics, TimeTracker},
};
use smallvec::{SmallVec, smallvec};
use tikv_util::time::Instant;
use trace_event::types::TraceContext;
use tracker::{TrackerToken, get_tls_tracker_token};
use uuid::Uuid;

use super::{Peer, PeerInbox, RaftApplyState};
use crate::store::{
    ApplyMetrics, ExecResult, Proposal, RegionIdVer, RegionSnapshot, TrimOverBoundParameter,
    util::KeysInfoFormatter,
};

#[derive(Debug)]
pub enum PeerMsg {
    /// Do not construct this variant directly to send raft messages; use
    /// `send_raft_msg` or `force_send_raft_msg` instead.
    RaftMessage(rspb::RaftMessage),
    RaftCommand(RaftCommand),
    Tick,
    Start,
    ApplyResult(MsgApplyResult),
    CasualMessage(CasualMessage),
    /// Message that can't be lost but rarely created. If they are lost, real
    /// bad things happen like some peers will be considered dead in the
    /// group.
    SignificantMsg(SignificantMsg),
    GenerateEngineChangeSet(kvenginepb::ChangeSet, Callback),
    ApplySnapshotResult(kvenginepb::ChangeSet),
    PrepareChangeSetResult(
        kvengine::Result<kvengine::ChangeSet>,
        u64, // peer_id
    ),
    PrepareCommitMergeResult(
        kvengine::Result<kvengine::ChangeSet>,
        u64, // commit index
    ),
    PrepareTxnFileResult {
        entry_index: u64,
        peer_id: u64,
        apply_generation: u64,
        txn_file: TxnFile,
    },
    Persisted(PersistReady),
    RaftlogFetched(crate::store::worker::FetchedLogs),
    ColdRegionTick(u64 /* sweep id */),
    ColdReadIndexTimeout,
    ColdLeaderProbeTimeout,
    Idle(PeerInbox),
    WakeUp(PeerInbox),
    StoreMsgForWakeUp(StoreMsg),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ColdCampaignReason {
    ReadIndexTimeout,
    LeaderUnknown,
    LeaderStoreDown,
}

#[derive(Clone, Copy, Debug)]
pub(crate) struct ColdLeaderProbeTask {
    /// Time at which the follower may campaign if the probe has not completed.
    pub(crate) deadline: StdInstant,
    pub(crate) probe_id: Uuid,
    pub(crate) leader_id: u64,
    pub(crate) campaign_reason: ColdCampaignReason,
}

impl PeerMsg {
    pub(crate) fn size(&self) -> usize {
        match self {
            PeerMsg::RaftMessage(msg) => {
                let entries = msg.get_message().get_entries();
                let mut size = 0;
                for entry in entries {
                    size += entry.data.len();
                }
                size
            }
            PeerMsg::RaftCommand(cmd) => cmd.request_size,
            PeerMsg::RaftlogFetched(res) => {
                if let Ok(ents) = &res.logs.ents {
                    let mut size = 0;
                    for entry in ents.iter() {
                        size += entry.data.len();
                    }
                    return size;
                }
                0
            }
            PeerMsg::Idle(inbox) | PeerMsg::WakeUp(inbox) => {
                inbox.msgs.iter().map(|msg| msg.size()).sum()
            }
            _ => 0,
        }
    }

    pub(crate) fn type_str(&self) -> &str {
        match self {
            PeerMsg::RaftMessage(_) => "RaftMessage",
            PeerMsg::RaftCommand(_) => "RaftCommand",
            PeerMsg::Tick => "Tick",
            PeerMsg::Start => "Start",
            PeerMsg::ApplyResult(_) => "ApplyResult",
            PeerMsg::CasualMessage(_) => "CasualMessage",
            PeerMsg::SignificantMsg(_) => "SignificantMsg",
            PeerMsg::GenerateEngineChangeSet(..) => "GenerateEngineChangeSet",
            PeerMsg::ApplySnapshotResult(_) => "ApplySnapshotResult",
            PeerMsg::PrepareChangeSetResult(..) => "PrepareChangeSetResult",
            PeerMsg::PrepareCommitMergeResult(..) => "PrepareCommitMergeResult",
            PeerMsg::PrepareTxnFileResult { .. } => "PrepareTxnFileResult",
            PeerMsg::Persisted(_) => "Persisted",
            PeerMsg::RaftlogFetched(_) => "RaftlogFetched",
            PeerMsg::ColdRegionTick(_) => "ColdRegionTick",
            PeerMsg::ColdReadIndexTimeout => "ColdReadIndexTimeout",
            PeerMsg::ColdLeaderProbeTimeout => "ColdLeaderProbeTimeout",
            PeerMsg::Idle(_) => "Idle",
            PeerMsg::WakeUp(_) => "WakeUp",
            PeerMsg::StoreMsgForWakeUp(_) => "StoreMsgForWakeUp",
        }
    }

    pub(crate) fn type_code(&self) -> u8 {
        match self {
            PeerMsg::RaftMessage(_) => 1,
            PeerMsg::RaftCommand(_) => 2,
            PeerMsg::Tick => 3,
            PeerMsg::Start => 4,
            PeerMsg::ApplyResult(_) => 5,
            PeerMsg::CasualMessage(_) => 6,
            PeerMsg::SignificantMsg(_) => 7,
            PeerMsg::GenerateEngineChangeSet(..) => 8,
            PeerMsg::ApplySnapshotResult(_) => 9,
            PeerMsg::PrepareChangeSetResult(..) => 10,
            PeerMsg::PrepareCommitMergeResult(..) => 11,
            PeerMsg::PrepareTxnFileResult { .. } => 12,
            PeerMsg::Persisted(_) => 13,
            PeerMsg::RaftlogFetched(_) => 17,
            PeerMsg::ColdRegionTick(_) => 18,
            PeerMsg::ColdReadIndexTimeout => 19,
            PeerMsg::ColdLeaderProbeTimeout => 20,
            PeerMsg::Idle(_) => 14,
            PeerMsg::WakeUp(_) => 15,
            PeerMsg::StoreMsgForWakeUp(_) => 16,
        }
    }
}

#[derive(Debug)]
pub(crate) struct PendingSplit {
    pub cs: kvenginepb::ChangeSet,
    /// Child shard IDs whose rfengine metadata write was intentionally skipped
    /// during merged-engine preprocessing. Kvengine split must not materialize
    /// them later, or replay can recreate a stale child after rfstore has
    /// already advanced the child region to a newer lineage.
    pub skip_new_shard_ids: Vec<u64>,
}

#[derive(Debug)]
pub(crate) enum ApplyMsg {
    Apply {
        start: std::time::Instant,
        apply: MsgApply,
    },
    Registration(MsgRegistration),
    PendingSplit(PendingSplit),
    ApplyChangeSet(kvengine::ChangeSet),
    PrepareChangeSet {
        cs: kvenginepb::ChangeSet,
        encryption_state: EncryptionState,
        reload_snap: Option<kvenginepb::Snapshot>, /* The snap contains the current files that
                                                    * need to be reloaded. */
        prepare_type: FilePrepareType,
    },
    PrepareMerge,
    PrepareCommitMerge {
        source: kvenginepb::ChangeSet,
        commit_index: u64,
    },
    ResumeCommitMerge {
        source: kvengine::ChangeSet,
        commit_index: u64,
    },
    PrepareRollbackMerge(u64 /* initial flush sequence */),
    UnsafeDestroy {
        region_id: u64,
    },
    Maintenance {
        region_id: u64,
    },
    PrepareTxnFile {
        txn_file_ref: TxnFileRef,
        commit_index: u64,
        encryption_state: EncryptionState,
    },
    ResumeTxnFile {
        commit_index: u64,
        apply_generation: u64,
        txn_file: TxnFile,
    },
    TriggerRefreshShardStates,
    Change {
        cmd: ChangeObserver,
        region_epoch: metapb::RegionEpoch,
        cb: Callback,
    },
}

impl ApplyMsg {
    pub fn estimated_size(&self) -> usize {
        match self {
            // Currently only MsgApply (potentially large) is
            // considered.
            ApplyMsg::Apply { apply: m, .. } => m.estimated_size(),
            _ => 0,
        }
    }

    pub fn on_schedule(&mut self, metrics: &RaftMetrics, now: std::time::Instant) {
        if let ApplyMsg::Apply { apply: m, .. } = self {
            for cb in &mut m.cbs {
                for tracker in cb.cb.write_trackers_mut() {
                    tracker.observe(now, &metrics.store_time, |t| {
                        t.metrics.write_instant = Some(now);
                        &mut t.metrics.store_time_nanos
                    });
                    tracker.reset(now);
                }
            }
        }
    }
}

pub enum StoreMsg {
    Tick,
    Start {
        store: metapb::Store,
    },
    StoreUnreachable {
        store_id: u64,
    },
    GenerateEngineChangeSet(kvenginepb::ChangeSet, Callback),
    ProposeEngineChangeSet {
        region_id: u64,
        cs: kvenginepb::ChangeSet,
        expected_region_epoch: Option<metapb::RegionEpoch>,
        callback: Callback,
    },
    RaftMessage(kvproto::raft_serverpb::RaftMessage),
    SnapshotReady(u64),
    GetRegionsInRange {
        start: Vec<u8>,
        end: Vec<u8>,
        callback: Box<dyn FnOnce(Vec<RegionIdVer>) + Send>,
    },
    GetRegionById {
        region_id: u64,
        callback: Box<dyn FnOnce(Option<metapb::Region>) + Send>,
    },
    ApplyResult {
        region_id: u64,
        peer_id: u64,
    },
    PeerStartupActivated(u64 /* region id */),
    PeerStartupConnected(u64 /* region id */),
    DependentsEmpty(u64 /* region id */),
    PrepareMerge {
        region_id: u64,
        req: RaftCmdRequest,
        callback: Callback,
    },
    CheckMerge(u64),
    /// Inspect the latency of rfstore.
    LatencyInspect {
        factor: InspectFactor,
        send_time: Instant,
        // Maximum allowed delay between dispatch and actual execution in
        // DiskCheckRunner. Stale inspect tasks are dropped instead of
        // reporting a late callback.
        ttl: Duration,
        inspector: LatencyInspector,
    },
    Stop,
    #[cfg(any(test, feature = "testexport"))]
    QueryColdPeerStates {
        region_ids: Arc<HashSet<u64>>,
        callback: Box<dyn FnOnce(ColdPeerStateSnapshot) + Send>,
    },
    #[cfg(any(test, feature = "testexport"))]
    PdStoreHeartbeatTick,
}

/// Observable non-Cold states exposed only to tests.
#[cfg(any(test, feature = "testexport"))]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ColdPeerStateKind {
    Regular,
    TempRegular,
}

/// Identifies a tracked peer that is not currently Cold.
#[cfg(any(test, feature = "testexport"))]
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct NonColdPeerState {
    /// Region containing the peer.
    pub region_id: u64,
    /// Peer ID local to the region.
    pub peer_id: u64,
    /// Current non-Cold scheduling state.
    pub state: ColdPeerStateKind,
}

/// Aggregate Cold state for the tracked peers on one store.
#[cfg(any(test, feature = "testexport"))]
#[derive(Debug, Default)]
pub struct ColdPeerStateSnapshot {
    /// Number of tracked peers found on the store.
    pub peer_count: usize,
    /// Number of tracked peers in Cold.
    pub cold_count: usize,
    /// Number of tracked peers in Regular.
    pub regular_count: usize,
    /// Number of tracked peers in TempRegular.
    pub temp_regular_count: usize,
    /// Details for Regular and TempRegular peers.
    pub non_cold_peers: Vec<NonColdPeerState>,
}

impl fmt::Debug for StoreMsg {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        // dummy implementation for debug peer message.
        write!(f, "StoreMsg")
    }
}

#[derive(Debug)]
pub struct PersistReady {
    pub(crate) region_id: u64,
    pub(crate) peer_id: u64,
    pub(crate) ready_number: u64,
    pub(crate) _commit_idx: u64,
    pub(crate) raft_messages: Vec<RaftMessage>,
}

/// IOTask contains I/O tasks which need to be persisted to raft db.
pub(crate) struct IoTask {
    pub(crate) readies: Vec<PersistReady>,
    pub(crate) raft_wb: rfengine::WriteBatch,
    pub(crate) remove_dependents: Vec<(u64 /* parent_id */, u64 /* dependent_id */)>,
    pub(crate) raft_wb_entry_trackers: Vec<Vec<TimeTracker>>,
    pub(crate) send_time: Instant,
}

pub(crate) enum IoWorkerTask {
    IoTask(IoTask),
    UpdateConfig {
        max_batch_size: Option<usize>,
        min_write_duration: Option<Duration>,
    },
}

#[derive(Debug)]
pub struct RaftCommand {
    pub send_time: Instant,
    pub(crate) request: RaftCmdRequest,
    pub trace_ctx: TraceContext,
    pub callback: Callback,
    request_size: usize,
}

impl RaftCommand {
    pub fn new(request: RaftCmdRequest, trace_ctx: TraceContext, callback: Callback) -> Self {
        let send_time = Instant::now();
        let request_size = request.compute_size() as usize;
        Self {
            request,
            trace_ctx,
            callback,
            send_time,
            request_size,
        }
    }
}

#[derive(Debug)]
pub struct MsgApply {
    pub(crate) term: u64,
    pub(crate) entries: Vec<eraftpb::Entry>,
    pub(crate) new_role: Option<raft::StateRole>,
    pub(crate) cbs: Vec<Proposal>,
    pub(crate) bucket_meta: Option<Arc<BucketMeta>>,
}

impl MsgApply {
    /// Split off [-, raft-index). The rest [raft-index, +) is left in `self`.
    pub fn split_off(&mut self, raft_idx: u64) -> Option<MsgApply> {
        let pos = self
            .entries
            .iter()
            .position(|e| e.get_index() >= raft_idx)
            .unwrap_or(self.entries.len());
        if pos == 0 {
            return None;
        }
        let mut split_entries = self.entries.split_off(pos);
        std::mem::swap(&mut split_entries, &mut self.entries);

        let cbs_pos = self
            .cbs
            .iter()
            .position(|cb| cb.index >= raft_idx)
            .unwrap_or(self.cbs.len());
        let mut split_cbs = self.cbs.split_off(cbs_pos);
        std::mem::swap(&mut split_cbs, &mut self.cbs);

        Some(MsgApply {
            term: self.term,
            entries: split_entries,
            new_role: self.new_role,
            cbs: split_cbs,
            bucket_meta: self.bucket_meta.clone(),
        })
    }

    /// Return `None` when there is no entry.
    #[inline]
    pub fn first_raft_index(&self) -> Option<u64> {
        self.entries.first().map(|e| e.get_index())
    }

    /// Return `None` when there is no entry.
    #[inline]
    pub fn last_raft_index(&self) -> Option<u64> {
        self.entries.last().map(|e| e.get_index())
    }

    pub fn new_for_replication(entries: Vec<eraftpb::Entry>) -> Self {
        Self {
            term: 1,
            entries,
            new_role: None,
            cbs: vec![],
            bucket_meta: None,
        }
    }

    pub fn estimated_size(&self) -> usize {
        let entries_size: usize = self.entries.iter().map(|e| e.data.len()).sum();
        entries_size
    }
}

#[derive(Debug)]
pub struct MsgApplyResult {
    pub(crate) peer_id: u64,
    pub(crate) results: VecDeque<ExecResult>,
    pub(crate) apply_state: RaftApplyState,
    pub(crate) metrics: ApplyMetrics,
    pub(crate) bucket_stat: Option<Box<BucketStat>>,
    pub(crate) encryption_paused_index: Option<u64>,
}

#[derive(Derivative)]
#[derivative(Debug)]
pub struct MsgRegistration {
    pub(crate) peer: metapb::Peer,
    pub(crate) term: u64,
    pub(crate) apply_generation: u64,
    pub(crate) apply_state: RaftApplyState,
    pub(crate) region: metapb::Region,
    #[derivative(Debug = "ignore")]
    pub(crate) encryption_state: EncryptionState,
}

impl MsgRegistration {
    pub(crate) fn new(peer: &Peer) -> Self {
        Self {
            peer: peer.peer.clone(),
            term: peer.term(),
            apply_generation: peer.apply_generation,
            apply_state: peer.get_store().apply_state(),
            region: peer.get_store().region().clone(),
            encryption_state: peer.encryption_state.clone(),
        }
    }
}

#[derive(Debug)]
pub struct ReadResponse {
    pub response: RaftCmdResponse,
    pub snapshot: Option<RegionSnapshot>,
    pub txn_extra_op: TxnExtraOp,
}

#[derive(Debug)]
pub struct WriteResponse {
    pub response: RaftCmdResponse,

    // Used for split_region only.
    pub key_errors: Option<Vec<kvproto::kvrpcpb::KeyError>>,
}

pub type ReadCallback = Box<dyn FnOnce(ReadResponse) + Send>;
pub type WriteCallback = Box<dyn FnOnce(WriteResponse) + Send>;
pub type ExtCallback = Box<dyn FnOnce() + Send>;

/// Variants of callbacks for `Msg`.
///  - `Read`: a callback for read only requests including `StatusRequest`,
///    `GetRequest` and `SnapRequest`
///  - `Write`: a callback for write only requests including `AdminRequest`
///    `PutRequest`, `DeleteRequest` and `DeleteRangeRequest`.
pub enum Callback {
    /// No callback.
    None,
    /// Read callback.
    Read {
        cb: ReadCallback,

        tracker: TrackerToken,
    },
    /// Write callback.
    Write {
        cb: WriteCallback,
        /// `proposed_cb` is called after a request is proposed to the raft
        /// group successfully. It's used to notify the caller to move
        /// on early because it's very likely the request
        /// will be applied to the raftstore.
        proposed_cb: Option<ExtCallback>,
        /// `committed_cb` is called after a request is committed and before
        /// it's being applied, and it's guaranteed that the request
        /// will be successfully applied soon.
        committed_cb: Option<ExtCallback>,

        trackers: SmallVec<[TimeTracker; 4]>,
    },
}

impl Callback {
    pub fn read(cb: ReadCallback) -> Self {
        let tracker = get_tls_tracker_token();
        Callback::Read { cb, tracker }
    }

    pub fn write(cb: WriteCallback) -> Self {
        Self::write_ext(cb, None, None)
    }

    pub fn write_ext(
        cb: WriteCallback,
        proposed_cb: Option<ExtCallback>,
        committed_cb: Option<ExtCallback>,
    ) -> Self {
        let tracker = TimeTracker::default();
        Callback::Write {
            cb,
            proposed_cb,
            committed_cb,
            trackers: smallvec![tracker],
        }
    }

    pub fn invoke_with_response(self, resp: RaftCmdResponse) {
        self.invoke_with_response_ext(resp, None);
    }

    pub fn invoke_with_response_ext(
        self,
        resp: RaftCmdResponse,
        key_errs: Option<Vec<kvproto::kvrpcpb::KeyError>>,
    ) {
        match self {
            Callback::None => (),
            Callback::Read { cb, .. } => {
                let resp = ReadResponse {
                    response: resp,
                    snapshot: None,
                    txn_extra_op: TxnExtraOp::Noop,
                };
                cb(resp);
            }
            Callback::Write { cb, .. } => {
                let resp = WriteResponse {
                    response: resp,
                    key_errors: key_errs,
                };
                cb(resp);
            }
        }
    }

    pub fn has_proposed_cb(&mut self) -> bool {
        if let Callback::Write { proposed_cb, .. } = self {
            proposed_cb.is_some()
        } else {
            false
        }
    }

    pub fn invoke_proposed(&mut self) {
        if let Callback::Write { proposed_cb, .. } = self {
            if let Some(cb) = proposed_cb.take() {
                cb()
            }
        }
    }

    pub fn invoke_committed(&mut self) {
        if let Callback::Write { committed_cb, .. } = self {
            if let Some(cb) = committed_cb.take() {
                cb()
            }
        }
    }

    pub fn invoke_read(self, args: ReadResponse) {
        match self {
            Callback::Read { cb, .. } => cb(args),
            other => panic!("expect Callback::Read(..), got {:?}", other),
        }
    }

    pub fn is_none(&self) -> bool {
        matches!(self, Callback::None)
    }

    pub fn read_tracker(&self) -> Option<TrackerToken> {
        let Callback::Read { tracker, .. } = self else {
            return None;
        };
        Some(*tracker)
    }

    #[inline]
    pub fn write_trackers(&self) -> &[TimeTracker] {
        match self {
            Callback::Write { trackers, .. } => trackers,
            _ => &[],
        }
    }

    #[inline]
    pub fn write_trackers_mut(&mut self) -> &mut [TimeTracker] {
        match self {
            Callback::Write { trackers, .. } => trackers.as_mut(),
            _ => &mut [],
        }
    }
}

impl fmt::Debug for Callback {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        match *self {
            Callback::None => write!(fmt, "Callback::None"),
            Callback::Read { .. } => write!(fmt, "Callback::Read(..)"),
            Callback::Write { .. } => write!(fmt, "Callback::Write(..)"),
        }
    }
}

/// Some significant messages sent to raftstore. Raftstore will dispatch these
/// messages to Raft groups to update some important internal status.
#[derive(Debug)]
pub enum SignificantMsg {
    StoreUnreachable {
        store_id: u64,
    },
    /// Capture changes of a region.
    CaptureChange {
        cmd: ChangeObserver,
        region_epoch: metapb::RegionEpoch,
        callback: Callback,
        can_apply: bool,
    },
    LeaderCallback(Callback),
}

/// The initiator of a split request carried in [`CasualMessage::SplitRegion`] /
/// [`CasualMessage::HalfSplitRegion`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SplitSource {
    Load,
    Schema,
    /// PD-initiated split; carries a label identifying the specific code path
    /// (e.g. `"heartbeat.use_key"`, `"heartbeat.half_split"`) for logging.
    Pd(&'static str),
    /// Direct split request from a gRPC client; carries the peer address for
    /// logging.
    Peer(String),
}

impl fmt::Display for SplitSource {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SplitSource::Load => f.write_str("load"),
            SplitSource::Schema => f.write_str("schema"),
            SplitSource::Pd(label) => write!(f, "pd.{}", label),
            SplitSource::Peer(peer) => write!(f, "peer.{}", peer),
        }
    }
}

/// Message that will be sent to a peer.
///
/// These messages are not significant and can be dropped occasionally.
pub enum CasualMessage {
    /// Split the target region into several partitions.
    SplitRegion {
        region_epoch: metapb::RegionEpoch,
        // It's an encoded key.
        // TODO: support meta key.
        split_keys: Vec<Vec<u8>>,
        callback: Callback,
        source: SplitSource,
    },

    /// Half split the target region.
    HalfSplitRegion {
        region_epoch: metapb::RegionEpoch,
        policy: pdpb::CheckPolicy,
        source: SplitSource,
    },
    /// Half split the target region within the given key range.
    HalfSplitKeyRange {
        region_epoch: metapb::RegionEpoch,
        key_range: KeyRange,
    },
    DeletePrefix {
        region_version: u64,
        prefix: Vec<u8>,
        callback: Callback,
    },
    /// IngestFiles from load_data worker.
    IngestFiles {
        cs: kvenginepb::ChangeSet,
        callback: Callback,
    },
    /// Restore from backup data.
    RestoreShard {
        cs: kvenginepb::ChangeSet,
        /// Expected target membership captured when the RestoreShard payload
        /// was generated. `None` preserves compatibility for offline restore.
        expected_conf_ver: Option<u64>,
        callback: Callback,
    },
    TriggerTrimOverBound(TrimOverBoundParameter),
    MajorCompact {
        major_compact: bool,
        columnar: bool,
        callback: Callback,
    },
    UpdateSchemaFile(SchemaFile),
    CheckLeader {
        shard_ver: u64,
        callback: Callback,
    },
    ClearColumnar {
        restore_version: Option<u64>,
    },
    ClearFtsIndex,
    TriggerRefreshShardStates,
    ForceSwitchMemTable {
        current_size: u64,
        callback: Callback,
    },
}

impl fmt::Debug for CasualMessage {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            CasualMessage::SplitRegion {
                ref split_keys,
                source,
                ..
            } => write!(
                fmt,
                "Split region with {} from {}",
                KeysInfoFormatter(split_keys.iter()),
                source,
            ),
            CasualMessage::HalfSplitRegion { source, .. } => {
                write!(fmt, "Half Split from {}", source)
            }
            CasualMessage::HalfSplitKeyRange { .. } => write!(fmt, "Half Split Key Range"),
            CasualMessage::DeletePrefix { prefix, .. } => {
                write!(fmt, "delete prefix {:?}", prefix)
            }
            CasualMessage::IngestFiles { cs, .. } => {
                write!(fmt, "ingest files {:?}", cs)
            }
            CasualMessage::RestoreShard { cs, .. } => {
                write!(fmt, "restore shard from {:?}", cs)
            }
            CasualMessage::TriggerTrimOverBound(param) => {
                write!(fmt, "trigger trim over bound {:?}", param)
            }
            CasualMessage::MajorCompact {
                major_compact,
                columnar,
                ..
            } => {
                write!(
                    fmt,
                    "major compact {:?}, columnar: {}",
                    major_compact, columnar
                )
            }
            CasualMessage::UpdateSchemaFile(schema_file) => {
                write!(
                    fmt,
                    "update schema file id:{}, ver:{}",
                    schema_file.get_file_id(),
                    schema_file.get_version()
                )
            }
            CasualMessage::CheckLeader { shard_ver, .. } => {
                write!(fmt, "check leader with version {}", shard_ver)
            }
            CasualMessage::ClearColumnar { restore_version } => {
                write!(
                    fmt,
                    "clear columnar, restore_version: {:?}",
                    restore_version
                )
            }
            CasualMessage::ClearFtsIndex => {
                write!(fmt, "clear fts index")
            }
            CasualMessage::TriggerRefreshShardStates => {
                write!(fmt, "trigger refresh shard states")
            }
            CasualMessage::ForceSwitchMemTable { .. } => {
                write!(fmt, "manual flush")
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use kvproto::raft_cmdpb::{CmdType, Request};

    use super::*;

    #[test]
    fn test_raft_command_size_includes_normal_write_value() {
        let value = vec![b'x'; 1024 * 1024];
        let mut request = Request::default();
        request.set_cmd_type(CmdType::Put);
        request.mut_put().set_key(b"key".to_vec());
        request.mut_put().set_value(value.clone());
        let mut command = RaftCmdRequest::default();
        command.set_requests(vec![request].into());
        let expected_size = command.compute_size() as usize;

        let message = PeerMsg::RaftCommand(RaftCommand::new(
            command,
            TraceContext::default(),
            Callback::None,
        ));
        assert_eq!(message.size(), expected_size);
        assert!(message.size() >= value.len());
    }
}