// Copyright 2025 PingCAP, Inc.

use std::{
    cmp,
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
};

use chrono::NaiveDateTime;
use collections::{HashMap, HashMapExt, HashSet};
use kvengine::dfs::Dfs;
use kvproto::metapb;
use merged_engine::StoreProgress;
use native_br::{backup::IncrementalBackupFile, common::get_all_incremental_backups};
use pd_client::{PdClient, util::get_all_stores_except_tiflash_async};
use protobuf::Message;
use rfengine::service_worker::WalProgress;
use security::HttpClient;
use serde_derive::{Deserialize, Serialize};
use tikv_util::{
    box_err, box_try, box_try_join,
    config::{AbsoluteOrPercentSize, ReadableDuration, ReadableSize},
    debug, error, info,
    time::Instant,
    trace, warn,
};
use tokio::{
    sync::{mpsc, watch},
    task::JoinSet,
};
use txn_types::TimeStamp;

use crate::{
    Error, ReplicationWorkerConfig, Result, error::error_is_force_advance_eligible,
    metrics::REP_FEATCH_WAL_TARGET_COUNTER, util::send_request_to_store,
};

const MAX_PENDING_TARGETS: usize = 64;

const SEARCH_BACKUPS_LIMIT: usize = 50;

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct WalConfig {
    /// Whether to tolerate store errors (no more than 1 store) during update
    /// WAL.
    pub tolerate_store_err: bool,

    /// The conservative upper bound for a single WAL fetch response size.
    ///
    /// This should match the cluster's rfengine WAL file target size
    /// (`rfengine.target-file-size`).
    pub tikv_rfengine_wal_size: ReadableSize,
    /// The size limit of a WAL batch used to update stores.
    ///
    /// The concurrency of fetching WAL is `update_stores_wal_size_limit /
    /// tikv_rfengine_wal_size`.
    pub update_stores_wal_size_limit: AbsoluteOrPercentSize,
    /// The hard cap for concurrent remote WAL fetches.
    pub max_fetch_concurrency: usize,

    /// Throughput limit (bytes/sec) for fetching WAL bytes from an upstream
    /// TiKV store.
    ///
    /// This limit is enforced independently per store (i.e. each store has its
    /// own limiter with the same configured rate).
    ///
    /// Set to 0 to disable (unlimited).
    pub store_fetch_max_throughput: ReadableSize,

    /// Timeout for request `track_wal_progress` interface.
    pub track_wal_progress_timeout: ReadableDuration,

    /// The min/max time span of a target since the last one.
    ///
    /// Time span exceed the max limit will not be safe. We would miss the
    /// stores scale up then down during this period.
    ///
    /// As the targets fetched from rfengine are always of the current time, if
    /// the last target is too old, target will be fetched from backup.
    pub min_target_time_span: ReadableDuration,
    pub max_target_time_span: ReadableDuration,
    /// Whether to fetch WAL target from backup. Used to work around broken
    /// backups or bugs.
    pub fetch_target_from_backup: bool,
}

impl Default for WalConfig {
    fn default() -> Self {
        Self {
            tolerate_store_err: true,
            tikv_rfengine_wal_size: ReadableSize::mb(512),
            update_stores_wal_size_limit: AbsoluteOrPercentSize::Percent(20.0),
            max_fetch_concurrency: 9,
            store_fetch_max_throughput: ReadableSize::mb(256),
            track_wal_progress_timeout: ReadableDuration::secs(30),
            min_target_time_span: ReadableDuration::minutes(5),
            max_target_time_span: ReadableDuration::minutes(20),
            fetch_target_from_backup: true,
        }
    }
}

impl WalConfig {
    pub(crate) fn calc_fetch_wal_concurrency(&self) -> usize {
        ((self.update_stores_wal_size_limit.as_memory_size() / self.tikv_rfengine_wal_size.0)
            as usize)
            .min(self.max_fetch_concurrency)
            .max(1)
    }
}

// None: When the store is not ready.
pub(crate) type StoreWalProgresses = HashMap<u64 /* store_id */, Option<StoreProgress>>;

#[derive(Debug)]
pub(crate) struct WalTarget {
    pub(crate) ts: TimeStamp,
    pub(crate) store_targets: StoreWalProgresses,
}

#[derive(Debug)]
pub(crate) struct StoreLag {
    pub(crate) epoch_lag: u32,
    pub(crate) offset_lag: i64, // Can be negative.
}

impl StoreLag {
    pub(crate) fn compare(&self, other: &Self) -> cmp::Ordering {
        self.epoch_lag
            .cmp(&other.epoch_lag)
            .then_with(|| self.offset_lag.cmp(&other.offset_lag))
    }
}

// WalTargetFetcher -> WalFetcher
#[derive(Clone)]
pub(crate) struct WalTargetSender {
    tx: mpsc::UnboundedSender<WalTarget>,
    pending_count: Arc<AtomicUsize>,
}

// WalFetcher <- WalTargetFetcher.
pub(crate) struct WalTargetReceiver {
    rx: mpsc::UnboundedReceiver<WalTarget>,
    pending_count: Arc<AtomicUsize>,
}

pub(crate) fn wal_target_channel() -> (WalTargetSender, WalTargetReceiver) {
    let (tx, rx) = mpsc::unbounded_channel();
    let pending_count = Arc::new(AtomicUsize::new(0));
    (
        WalTargetSender {
            tx,
            pending_count: pending_count.clone(),
        },
        WalTargetReceiver { rx, pending_count },
    )
}

impl WalTargetSender {
    pub(crate) fn send(
        &self,
        item: WalTarget,
    ) -> std::result::Result<(), mpsc::error::SendError<WalTarget>> {
        self.pending_count.fetch_add(1, Ordering::Relaxed);
        match self.tx.send(item) {
            Ok(()) => Ok(()),
            Err(err) => {
                self.pending_count.fetch_sub(1, Ordering::Relaxed);
                Err(err)
            }
        }
    }

    pub(crate) fn len(&self) -> usize {
        self.pending_count.load(Ordering::Relaxed)
    }
}

impl WalTargetReceiver {
    pub(crate) async fn recv(&mut self) -> Option<WalTarget> {
        let item = self.rx.recv().await;
        if item.is_some() {
            self.pending_count.fetch_sub(1, Ordering::Relaxed);
        }
        item
    }
}

pub(crate) struct WalTargetFetcherHandle {
    join_handle: tokio::task::JoinHandle<()>,
}

impl WalTargetFetcherHandle {
    pub(crate) async fn wait(self) {
        if let Err(err) = self.join_handle.await {
            if err.is_panic() {
                panic!("WalTargetFetcher panic: {err:?}");
            }
        }
    }
}

pub(crate) struct WalTargetFetcher {
    pd: Arc<dyn PdClient>,
    config: WalConfig,
    thread_pool: tokio::runtime::Handle,
    fs: Arc<dyn Dfs>,
    http_client: HttpClient,

    skip_store_addr_keywords: Vec<String>,
    include_only_store_addr_keywords: Vec<String>,
}

impl WalTargetFetcher {
    pub(crate) fn spawn(
        pd: Arc<dyn PdClient>,
        init_target_ts: TimeStamp,
        config: &ReplicationWorkerConfig,
        thread_pool: tokio::runtime::Handle,
        fs: Arc<dyn Dfs>,
        target_tx: WalTargetSender,
        mut stop_rx: watch::Receiver<bool>,
    ) -> WalTargetFetcherHandle {
        let interval = config.sync_interval.0;
        let wal_config = &config.wal;
        let fetcher = Arc::new(Self::new(
            pd.clone(),
            wal_config.clone(),
            thread_pool.clone(),
            fs.clone(),
            config.skip_store_addr_keywords.clone(),
            config.include_only_store_addr_keywords.clone(),
        ));
        let mut last_target_ts = init_target_ts;
        let mut skipped_stores = SkippedStores::new(
            pd,
            config.skip_store_addr_keywords.clone(),
            config.include_only_store_addr_keywords.clone(),
            wal_config.min_target_time_span.0,
        );
        let join_handle = thread_pool.spawn(async move {
            loop {
                if *stop_rx.borrow() {
                    info!("WalTargetFetcher: stop requested, exit");
                    return;
                }

                let start_time = Instant::now_coarse();

                let fetch_res = tokio::select! {
                    res = fetcher.fetch_target_ts_and_progress(&target_tx, last_target_ts, &mut skipped_stores) => res,
                    res = stop_rx.changed() => {
                        if res.is_err() || *stop_rx.borrow() {
                            info!("WalTargetFetcher: stop requested, exit");
                            return;
                        }
                        continue;
                    }
                };

                match fetch_res {
                    Ok(Some(wal_target)) => {
                        let ts = wal_target.ts;
                        if let Err(mpsc::error::SendError(_item)) =
                            target_tx.send(wal_target)
                        {
                            info!("WalTargetFetcher: targets channel closed, exit");
                            return;
                        }
                        last_target_ts = ts;
                    }
                    Ok(None) => {}
                    Err(err) => {
                        error!("WalTargetFetcher: fetch target failed: {:?}", err);
                        REP_FEATCH_WAL_TARGET_COUNTER
                            .with_label_values(&["error", ""])
                            .inc();
                    }
                }

                let elapsed = start_time.saturating_elapsed();
                tokio::select! {
                    _ = tokio::time::sleep(interval.saturating_sub(elapsed)) => {}
                    res = stop_rx.changed() => {
                        if res.is_err() || *stop_rx.borrow() {
                            info!("WalTargetFetcher: stop requested, exit");
                            return;
                        }
                    }
                }
            }
        });
        WalTargetFetcherHandle { join_handle }
    }

    fn new(
        pd: Arc<dyn PdClient>,
        config: WalConfig,
        thread_pool: tokio::runtime::Handle,
        fs: Arc<dyn Dfs>,
        skip_store_addr_keywords: Vec<String>,
        include_only_store_addr_keywords: Vec<String>,
    ) -> Self {
        let http_client = pd
            .get_security_mgr()
            .http_client(hyper::Client::builder())
            .unwrap();
        Self {
            pd,
            config,
            http_client,
            thread_pool,
            fs,
            skip_store_addr_keywords,
            include_only_store_addr_keywords,
        }
    }

    fn tolerate_store_err(&self) -> usize {
        self.config.tolerate_store_err as usize
    }

    async fn fetch_target_ts_and_progress(
        self: &Arc<Self>,
        target_tx: &WalTargetSender,
        last_target_ts: TimeStamp,
        skipped_stores: &mut SkippedStores,
    ) -> Result<Option<WalTarget>> {
        if target_tx.len() >= MAX_PENDING_TARGETS {
            debug!(
                "WalTargetFetcher: skip fetch, too many pending targets: {}",
                target_tx.len()
            );
            REP_FEATCH_WAL_TARGET_COUNTER
                .with_label_values(&["warn", "too_many_pending"])
                .inc();
            return Ok(None);
        }

        debug_assert!(!last_target_ts.is_zero());

        if self.should_fetch_from_backup(last_target_ts) {
            self.fetch_from_backup(last_target_ts, skipped_stores).await
        } else {
            self.fetch_from_rfengine(last_target_ts)
                .await
                .map(|x| Some(x))
        }
    }

    // Always fetch from backup when `min_target_time_span` is zero for test
    // purpose.
    fn should_fetch_from_backup(&self, last_target_ts: TimeStamp) -> bool {
        if !self.config.fetch_target_from_backup {
            return false;
        }

        #[cfg(feature = "testexport")]
        if self.config.min_target_time_span.is_zero() {
            return true;
        }

        TimeStamp::now()
            .physical()
            .saturating_sub(last_target_ts.physical())
            >= self.config.max_target_time_span.as_millis()
    }

    async fn fetch_from_rfengine(self: &Arc<Self>, last_target_ts: TimeStamp) -> Result<WalTarget> {
        let ori_stores = get_all_stores_except_tiflash_async(self.pd.as_ref(), true).await?;
        let stores = filter_stores(
            &self.skip_store_addr_keywords,
            &self.include_only_store_addr_keywords,
            ori_stores,
        );
        let target_ts = self.pd.get_min_tso().await?;

        let mut errors = vec![];
        let mut progresses = HashMap::with_capacity(stores.len());
        let mut join_set = JoinSet::new();
        for store in stores {
            let fetcher = self.clone();
            join_set.spawn_on(
                async move {
                    let res = fetcher.track_store_wal_progress(&store, target_ts).await;
                    (store.id, res)
                },
                &self.thread_pool,
            );
        }

        while let Some(task) = join_set.join_next().await {
            let (store_id, res) = box_try_join!(task);
            match res {
                Ok(wal_progress) => {
                    progresses.insert(
                        store_id,
                        Some(StoreProgress {
                            store_id,
                            epoch: wal_progress.epoch,
                            offset: wal_progress.offset,
                        }),
                    );
                }
                Err(err) => {
                    warn!("WalTargetFetcher: fetch from rfengine failed: {:?}", err; "store" => store_id);
                    progresses.insert(store_id, None);
                    errors.push(err);
                }
            }
        }

        if errors.len() > self.tolerate_store_err() {
            return Err(errors.pop().unwrap());
        }

        debug!("WalTargetFetcher: fetched from rfengine";
            "target_ts" => ?target_ts,
            "progresses" => ?progresses,
            "last_target_ts" => ?last_target_ts,
        );
        REP_FEATCH_WAL_TARGET_COUNTER
            .with_label_values(&["ok", "from_rfengine"])
            .inc();
        Ok(WalTarget {
            ts: target_ts,
            store_targets: progresses,
        })
    }

    async fn track_store_wal_progress(
        &self,
        store: &metapb::Store,
        ts: TimeStamp,
    ) -> Result<StoreProgress> {
        let security_mgr = self.pd.get_security_mgr();
        let uri = security_mgr
            .build_uri(format!(
                "{}/rfengine/track_wal_progress",
                &store.status_address
            ))
            .unwrap();
        let track_req = cloud_server::TrackWalProgressRequest {
            ts: ts.into_inner(),
        };
        let req = http::Request::post(&uri)
            .body(serde_json::to_vec(&track_req).unwrap().into())
            .unwrap();
        let (status, data) = send_request_to_store(
            req,
            &self.http_client,
            self.config.track_wal_progress_timeout.0,
        )
        .await?;
        if !status.is_success() {
            let err_str = String::from_utf8_lossy(&data);
            return Err(box_err!(
                "track_store_wal_progress: send request error: {}, store {}",
                err_str,
                store.id
            ));
        }
        let progress: WalProgress = serde_json::from_slice(&data).map_err(|e| -> Error {
            box_err!(
                "track_store_wal_progress: decode error: {}, store {}",
                e,
                store.id
            )
        })?;
        debug_assert_ne!(progress.offset, 0); // 0 will be conflict with "read to end" during dump WAL.
        Ok(StoreProgress {
            store_id: store.id,
            epoch: progress.epoch,
            offset: progress.offset,
        })
    }

    async fn fetch_from_backup(
        &self,
        last_target_ts: TimeStamp,
        skipped_stores: &mut SkippedStores,
    ) -> Result<Option<WalTarget>> {
        let (backup_file, backup_meta) = self.search_for_valid_backup(last_target_ts).await?;

        let target_ts = TimeStamp::new(backup_meta.backup_ts);

        box_try!(skipped_stores.refresh().await);

        let stores = backup_meta.get_stores();
        let mut progresses =
            HashMap::with_capacity(stores.len() + backup_meta.tolerated_err as usize);

        if backup_meta.tolerated_err > 0 {
            let error_stores = backup_meta.get_tolerated_err_stores();
            if error_stores.len() == backup_meta.tolerated_err as usize {
                for &store_id in error_stores {
                    if skipped_stores.contains(store_id) {
                        continue;
                    }
                    progresses.insert(store_id, None);
                }
            } else {
                // Backward compatibility. Use fake store id.
                for i in 0..backup_meta.tolerated_err as u64 {
                    progresses.insert(u64::MAX - i, None);
                }
            }
        }

        for store in stores {
            if skipped_stores.contains(store.store_id) {
                continue;
            }
            progresses.insert(
                store.store_id,
                Some(StoreProgress {
                    store_id: store.store_id,
                    epoch: store.epoch,
                    offset: store.offset,
                }),
            );
        }

        info!("WalTargetFetcher: fetched from backup";
            "target_ts" => ?target_ts,
            "progresses" => ?progresses,
            "last_target_ts" => ?last_target_ts,
            "backup" => backup_file.name(),
            "safe_ts" => ?TimeStamp::new(backup_meta.safe_ts),
        );
        REP_FEATCH_WAL_TARGET_COUNTER
            .with_label_values(&["ok", "from_backup"])
            .inc();
        Ok(Some(WalTarget {
            ts: target_ts,
            store_targets: progresses,
        }))
    }

    async fn search_for_valid_backup(
        &self,
        last_target_ts: TimeStamp,
    ) -> Result<(IncrementalBackupFile, rfenginepb::ClusterBackupMeta)> {
        let timestamp_ms =
            last_target_ts.physical() as i64 + self.config.min_target_time_span.as_millis() as i64;
        let start_datetime = NaiveDateTime::from_timestamp_millis(timestamp_ms).unwrap();

        let max_backup_ts = TimeStamp::compose(
            last_target_ts.physical() + self.config.max_target_time_span.as_millis(),
            0,
        );
        let max_backup = IncrementalBackupFile::from_backup_ts(max_backup_ts.into_inner());

        let (files, _) = box_try!(
            get_all_incremental_backups(
                self.fs.as_ref(),
                &start_datetime.date(),
                Some(&start_datetime.time()),
                SEARCH_BACKUPS_LIMIT,
            )
            .await
        );
        debug!("WalTargetFetcher: search_for_valid_backup, candidates: {:?}", files;
            "start_datetime" => ?start_datetime, "max_backup" => ?max_backup);
        let mut candidates = 0;
        for backup_file in files {
            if backup_file.name() > max_backup.name() {
                break;
            }
            candidates += 1;

            let full_path = backup_file.full_path(&self.fs.get_prefix());
            let object = box_try!(
                self.fs
                    .get_object(
                        full_path.clone(),
                        full_path.clone(),
                        engine_traits::GetObjectOptions::default()
                    )
                    .await
            );
            let mut backup_meta = rfenginepb::ClusterBackupMeta::new();
            box_try!(backup_meta.merge_from_bytes(&object));

            if self.backup_is_valid(&backup_meta, last_target_ts) {
                return Ok((backup_file, backup_meta));
            }

            debug!("WalTargetFetcher: backup is invalid";
                "meta" => %backup_meta, "last_target_ts" => ?last_target_ts);
            REP_FEATCH_WAL_TARGET_COUNTER
                .with_label_values(&["info", "invalid_backup"])
                .inc();
        }

        // No valid backup, the replication will stop.
        // Should send an alarm.
        error!("WalTargetFetcher: no valid backup";
            "last_target_ts" => ?last_target_ts, "start_datetime" => ?start_datetime,
            "candidates" => candidates);
        REP_FEATCH_WAL_TARGET_COUNTER
            .with_label_values(&["fatal", "no_backup"])
            .inc();
        Err(Error::NoValidBackup)
    }

    fn backup_is_valid(
        &self,
        backup_meta: &rfenginepb::ClusterBackupMeta,
        last_target_ts: TimeStamp,
    ) -> bool {
        backup_meta.is_lightweight
            && backup_meta.backup_ts > last_target_ts.into_inner()
            && backup_meta.tolerated_err as usize <= self.tolerate_store_err()
            && backup_meta.safe_ts <= last_target_ts.into_inner()
    }
}

#[derive(Debug)]
pub(crate) struct FetchWalError {
    pub(crate) err: Error,
    pub(crate) target: Option<StoreProgress>,
}

impl FetchWalError {
    pub(crate) fn is_force_advance_eligible(&self) -> bool {
        error_is_force_advance_eligible(&self.err)
    }
}

fn filter_stores(
    skip_store_addr_keywords: &[String],
    include_only_store_addr_keywords: &[String],
    stores: Vec<metapb::Store>,
) -> Vec<metapb::Store> {
    if skip_store_addr_keywords.is_empty() && include_only_store_addr_keywords.is_empty() {
        return stores;
    }
    stores
        .into_iter()
        .filter(|store| {
            !should_skip_store(
                skip_store_addr_keywords,
                include_only_store_addr_keywords,
                store,
            )
        })
        .collect()
}

fn get_skipped_stores(
    skip_store_addr_keywords: &[String],
    include_only_store_addr_keywords: &[String],
    stores: &[metapb::Store],
) -> Option<HashSet<u64 /* store_id */>> {
    if skip_store_addr_keywords.is_empty() && include_only_store_addr_keywords.is_empty() {
        return None;
    }
    let skipped_store = stores
        .iter()
        .filter_map(|store| {
            should_skip_store(
                skip_store_addr_keywords,
                include_only_store_addr_keywords,
                store,
            )
            .then_some(store.id)
        })
        .collect::<HashSet<_>>();
    (!skipped_store.is_empty()).then_some(skipped_store)
}

fn should_skip_store(
    skip_store_addr_keywords: &[String],
    include_only_store_addr_keywords: &[String],
    store: &metapb::Store,
) -> bool {
    if !include_only_store_addr_keywords.is_empty() {
        return !store_addr_contains_keywords(store, include_only_store_addr_keywords);
    }
    store_addr_contains_keywords(store, skip_store_addr_keywords)
}

fn store_addr_contains_keywords(store: &metapb::Store, keywords: &[String]) -> bool {
    keywords
        .iter()
        .any(|keyword| store.get_address().contains(keyword))
}

pub(crate) struct SkippedStores {
    pd: Arc<dyn PdClient>,
    skip_store_addr_keywords: Vec<String>,
    include_only_store_addr_keywords: Vec<String>,
    refresh_interval: Duration,

    inner: Option<HashSet<u64 /* store_id */>>,
    update_time: Option<Instant>,
}

impl SkippedStores {
    pub(crate) fn new(
        pd: Arc<dyn PdClient>,
        skip_store_addr_keywords: Vec<String>,
        include_only_store_addr_keywords: Vec<String>,
        refresh_interval: Duration,
    ) -> Self {
        Self {
            pd,
            skip_store_addr_keywords,
            include_only_store_addr_keywords,
            refresh_interval,
            inner: None,
            update_time: None,
        }
    }
    pub(crate) async fn refresh(&mut self) -> Result<()> {
        if self.skip_store_addr_keywords.is_empty()
            && self.include_only_store_addr_keywords.is_empty()
        {
            return Ok(());
        }

        let now = Instant::now_coarse();
        if self
            .update_time
            .is_some_and(|x| now.saturating_duration_since(x) < self.refresh_interval)
        {
            return Ok(());
        }

        // Includes tombstone stores.
        let ori_stores = get_all_stores_except_tiflash_async(self.pd.as_ref(), false).await?;
        self.inner = get_skipped_stores(
            &self.skip_store_addr_keywords,
            &self.include_only_store_addr_keywords,
            &ori_stores,
        );
        self.update_time = Some(now);
        trace!("SkippedStores refresh";
            "ori_stores" => ?ori_stores, "skipped_stores" => ?self.inner);
        Ok(())
    }

    pub(crate) fn contains(&self, store_id: u64) -> bool {
        self.inner.as_ref().is_some_and(|s| s.contains(&store_id))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_filter_stores() {
        let store_addrs = vec![
            "tikv-p0-tikv-0", // id: 0
            "tikv-p1-tikv-0",
            "tikv-poc-p0-tikv-0", // id: 2
            "tikv-poc-p1-tikv-0",
            "tikv-poc-p2-tikv-0",
            "tikv-p2-tikv-0",
            "tikv-hot-p0-tikv-0", // id: 6
            "tikv-hot-p1-tikv-0",
            "tikv-hot-p2-tikv-0",
        ];
        let stores = store_addrs
            .into_iter()
            .enumerate()
            .map(|(i, addr)| {
                let mut store = metapb::Store::default();
                store.set_id(i as u64);
                store.set_address(addr.to_string());
                store
            })
            .collect::<Vec<_>>();

        let skip_keywords = vec!["poc".to_string(), "hot".to_string()];
        let filtered_stores = filter_stores(&skip_keywords, &[], stores.clone());
        let filtered_addrs: Vec<&str> = filtered_stores
            .iter()
            .map(|store| store.get_address())
            .collect();
        let expected_addrs = vec!["tikv-p0-tikv-0", "tikv-p1-tikv-0", "tikv-p2-tikv-0"];
        assert_eq!(filtered_addrs, expected_addrs);

        let skipped_stores = get_skipped_stores(&skip_keywords, &[], &stores).unwrap();
        assert_eq!(skipped_stores, HashSet::from_iter([2u64, 3, 4, 6, 7, 8]));
        assert_eq!(get_skipped_stores(&["nothing".into()], &[], &stores), None);

        let include_only_keywords = vec!["tikv-poc-p0".to_string()];
        let filtered_stores = filter_stores(&skip_keywords, &include_only_keywords, stores.clone());
        let filtered_addrs: Vec<&str> = filtered_stores
            .iter()
            .map(|store| store.get_address())
            .collect();
        assert_eq!(filtered_addrs, vec!["tikv-poc-p0-tikv-0"]);
        let skipped_stores =
            get_skipped_stores(&skip_keywords, &include_only_keywords, &stores).unwrap();
        assert_eq!(
            skipped_stores,
            HashSet::from_iter([0u64, 1, 3, 4, 5, 6, 7, 8])
        );
    }
}