// Copyright 2026 PingCAP, Inc.

use std::{convert::TryFrom, sync::Arc};

use online_config::{ConfigChange, ConfigManager, ConfigValue, OnlineConfig};
use tikv_util::{
    config::{ReadableDuration, ReadableSize, VersionTrack},
    mpsc::Sender,
    yatp_pool::FuturePool,
};

use crate::store::{
    ColdRegionPolicy, Config, IoWorkerTask, TxnFileLockResolveLimiter,
    config::validate_raft_batch_sizes,
};

pub struct RfstoreConfigManager {
    config: Arc<VersionTrack<Config>>,
    initial_local_file_gc_timeout: ReadableDuration,
    io_worker_sender: Sender<Option<IoWorkerTask>>,
    apply_pool: FuturePool,
    txn_file_limiter: TxnFileLockResolveLimiter,
}

impl RfstoreConfigManager {
    pub(crate) fn new(
        config: Arc<VersionTrack<Config>>,
        io_worker_sender: Sender<Option<IoWorkerTask>>,
        apply_pool: FuturePool,
        txn_file_limiter: TxnFileLockResolveLimiter,
    ) -> Self {
        let initial_local_file_gc_timeout = config.value().local_file_gc_timeout;
        Self {
            config,
            initial_local_file_gc_timeout,
            io_worker_sender,
            apply_pool,
            txn_file_limiter,
        }
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn get_config(&self) -> Config {
        self.config.value().clone()
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn apply_pool(&self) -> FuturePool {
        self.apply_pool.clone()
    }
}

impl ConfigManager for RfstoreConfigManager {
    fn dispatch(&mut self, change: ConfigChange) -> online_config::Result<()> {
        let cold_region_policy = match change.get("cold_region") {
            Some(ConfigValue::Module(cold_region_change)) => cold_region_change.get("policy"),
            _ => None,
        };
        if let Some(policy) = cold_region_policy {
            let new_policy = ColdRegionPolicy::try_from(policy.clone())
                .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?;
            let current_policy = self.config.value().cold_region.policy;
            let valid_transition = matches!(
                (current_policy, new_policy),
                (ColdRegionPolicy::Disabled, ColdRegionPolicy::EnableFollower)
                    | (ColdRegionPolicy::EnableFollower, ColdRegionPolicy::Enabled)
                    | (ColdRegionPolicy::Enabled, ColdRegionPolicy::EnableFollower)
            );
            if new_policy != current_policy && !valid_transition {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!(
                        "raftstore.cold-region.policy allows online transitions \
                         disabled -> enable-follower and enable-follower <-> enabled, got \
                         {current_policy} to {new_policy}"
                    ),
                )
                .into());
            }
        }
        if cold_region_policy.is_some()
            || change.contains_key("split_region_check_tick_interval")
            || change.contains_key("max_leader_missing_duration")
        {
            let mut new_config = self.config.value().clone();
            new_config.update(change.clone())?;
            new_config.validate_cold_region()?;
        }

        let new_disk_hang_timeout = match change.get("disk_hang_timeout") {
            Some(ConfigValue::None) => Some(None),
            Some(v) => Some(Some(ReadableDuration::from(v.clone()))),
            None => None,
        };
        let new_concurrency = match change.get("txn_file_lock_active_resolve") {
            Some(ConfigValue::Module(active_resolve_change)) => {
                active_resolve_change.get("concurrency").map(usize::from)
            }
            _ => None,
        };
        if let Some(new_timeout) = new_disk_hang_timeout {
            Config::validate_disk_hang_timeout(new_timeout)?;
        }
        if new_concurrency == Some(0) {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "raftstore.txn-file-lock-active-resolve.concurrency should be >= 1",
            )
            .into());
        }

        let (current_entry_size, current_message_size, current_worker_batch_size) = {
            let current = self.config.value();
            (
                current.raft_entry_max_size,
                current.raft_max_size_per_msg,
                current.raft_worker_max_batch_size,
            )
        };
        let raft_entry_max_size = change
            .get("raft_entry_max_size")
            .map(|value| ReadableSize::from(value.clone()))
            .unwrap_or(current_entry_size);
        let raft_max_size_per_msg = change
            .get("raft_max_size_per_msg")
            .map(|value| ReadableSize::from(value.clone()))
            .unwrap_or(current_message_size);
        let raft_worker_max_batch_size = change
            .get("raft_worker_max_batch_size")
            .map(|value| ReadableSize::from(value.clone()))
            .unwrap_or(current_worker_batch_size);
        if let Err(error) = validate_raft_batch_sizes(
            raft_entry_max_size,
            raft_max_size_per_msg,
            raft_worker_max_batch_size,
        ) {
            return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, error).into());
        }

        if let Some(v) = change.get("local_file_gc_timeout") {
            let timeout = ReadableDuration::from(v.clone());
            if timeout < self.initial_local_file_gc_timeout {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    format!(
                        "raftstore.local-file-gc-timeout must not be less than its startup value {}",
                        self.initial_local_file_gc_timeout
                    ),
                )
                .into());
            }
        }

        // Mirror the startup floor in `Config::validate`: a non-zero interval
        // below `raft_base_tick_interval` would be silently rounded up to one
        // base tick, so reject it online too instead of letting it round up.
        // `raft_base_tick_interval` is `online_config(hidden)`, so it cannot
        // appear in `change`; compare against the current base interval.
        if let Some(v) = change.get("file_gc_compaction_check_tick_interval") {
            let interval_ms = ReadableDuration::from(v.clone()).as_millis();
            let base_ms = self.config.value().raft_base_tick_interval.as_millis();
            if interval_ms > 0 && interval_ms < base_ms {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidInput,
                    "raftstore.file-gc-compaction-check-tick-interval should be 0 (disabled) \
                     or at least raft-base-tick-interval",
                )
                .into());
            }
        }

        if change.contains_key("raft_worker_max_batch_size")
            || change.contains_key("io_worker_min_write_duration")
        {
            let max_batch_size = change
                .get("raft_worker_max_batch_size")
                .map(|v| ReadableSize::from(v.clone()).0 as usize);
            let min_write_duration = change
                .get("io_worker_min_write_duration")
                .map(|v| ReadableDuration::from(v.clone()).0);
            self.io_worker_sender
                .send(Some(IoWorkerTask::UpdateConfig {
                    max_batch_size,
                    min_write_duration,
                }))
                .unwrap();
        }
        if let Some(ConfigValue::Module(apply_change)) = change.get("apply_batch_system") {
            if let Some(v) = apply_change.get("pool_size") {
                let new_size: usize = v.into();
                let current_size = self.apply_pool.get_pool_size();
                self.apply_pool.scale_pool_size(new_size);
                tikv_util::info!("apply pool thread change"; "current" => current_size, "new" => new_size);
            }
        }

        self.config.update(|c| c.update(change.clone()))?;
        if let Some(new_concurrency) = new_concurrency {
            self.txn_file_limiter.update_concurrency(new_concurrency);
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use online_config::{ConfigChange, ConfigValue, OnlineConfig};
    use raftstore::store::config::MAX_RAFT_ENTRY_SIZE;
    use tikv_util::{
        config::ReadableDuration,
        mpsc,
        yatp_pool::{Config as YatpPoolConfig, DefaultTicker, YatpPoolBuilder},
    };

    use super::*;

    fn build_apply_pool(name: &str) -> FuturePool {
        let mut builder = YatpPoolBuilder::new(DefaultTicker);
        builder
            .name_prefix(name)
            .config(YatpPoolConfig::default_for_test());
        builder.build_future_pool()
    }

    fn build_manager(
        initial_concurrency: usize,
    ) -> (RfstoreConfigManager, TxnFileLockResolveLimiter, FuturePool) {
        let mut config = Config::default();
        config.txn_file_lock_active_resolve.concurrency = initial_concurrency;
        let (io_worker_sender, _io_worker_receiver) = mpsc::unbounded();
        let apply_pool = build_apply_pool("rfstore-config-manager-test");
        let limiter = TxnFileLockResolveLimiter::new(initial_concurrency);
        let manager = RfstoreConfigManager::new(
            Arc::new(VersionTrack::new(config)),
            io_worker_sender,
            apply_pool.clone(),
            limiter.clone(),
        );
        (manager, limiter, apply_pool)
    }

    #[test]
    fn test_dispatch_rejects_zero_active_resolve_concurrency() {
        let (mut manager, limiter, apply_pool) = build_manager(1);
        let mut active_resolve_change = ConfigChange::new();
        active_resolve_change.insert("concurrency".to_owned(), ConfigValue::Usize(0));
        let mut change = ConfigChange::new();
        change.insert(
            "txn_file_lock_active_resolve".to_owned(),
            ConfigValue::Module(active_resolve_change),
        );

        let err = manager.dispatch(change).unwrap_err().to_string();
        assert!(err.contains("txn-file-lock-active-resolve.concurrency"));
        assert_eq!(
            manager
                .get_config()
                .txn_file_lock_active_resolve
                .concurrency,
            1
        );
        assert_eq!(limiter.configured_concurrency(), 1);

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_rejects_subbase_file_gc_tick_interval() {
        let (mut manager, _limiter, apply_pool) = build_manager(1);
        // Default raft_base_tick_interval is 1s; a 500ms online value is below
        // the floor and must be rejected, not silently rounded up to one tick.
        let mut change = ConfigChange::new();
        change.insert(
            "file_gc_compaction_check_tick_interval".to_owned(),
            ConfigValue::Duration(500),
        );

        let err = manager.dispatch(change).unwrap_err().to_string();
        assert!(err.contains("file-gc-compaction-check-tick-interval"));
        assert_eq!(
            manager
                .get_config()
                .file_gc_compaction_check_tick_interval
                .as_millis(),
            10_000,
        );

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_validates_raft_entry_batch_limits() {
        let (mut manager, _limiter, apply_pool) = build_manager(1);
        let update_entry_size = |size| {
            let mut change = ConfigChange::new();
            change.insert("raft_entry_max_size".to_owned(), ConfigValue::Size(size));
            change
        };

        manager
            .dispatch(update_entry_size(MAX_RAFT_ENTRY_SIZE.0))
            .unwrap();
        assert_eq!(
            manager.get_config().raft_entry_max_size,
            MAX_RAFT_ENTRY_SIZE
        );

        let error = manager
            .dispatch(update_entry_size(MAX_RAFT_ENTRY_SIZE.0 + 1))
            .unwrap_err()
            .to_string();
        assert!(error.contains("less than or equal to 128MiB"));
        assert_eq!(
            manager.get_config().raft_entry_max_size,
            MAX_RAFT_ENTRY_SIZE
        );

        let mut change = ConfigChange::new();
        change.insert(
            "raft_worker_max_batch_size".to_owned(),
            ConfigValue::Size(ReadableSize::mb(38).0),
        );
        let error = manager.dispatch(change).unwrap_err().to_string();
        assert!(error.contains("plus three times"));
        assert_eq!(
            manager.get_config().raft_worker_max_batch_size,
            ReadableSize::mb(1)
        );

        let mut change = ConfigChange::new();
        change.insert(
            "raft_max_size_per_msg".to_owned(),
            ConfigValue::Size(ReadableSize::mb(239).0),
        );
        let error = manager.dispatch(change).unwrap_err().to_string();
        assert!(error.contains("the larger of"));
        assert_eq!(
            manager.get_config().raft_max_size_per_msg,
            ReadableSize::mb(1)
        );

        let mut change = ConfigChange::new();
        change.insert("raft_max_size_per_msg".to_owned(), ConfigValue::Size(0));
        let error = manager.dispatch(change).unwrap_err().to_string();
        assert!(error.contains("greater than 0"));
        assert_eq!(
            manager.get_config().raft_max_size_per_msg,
            ReadableSize::mb(1)
        );

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_limits_local_file_gc_timeout_decrease() {
        let (mut manager, _limiter, apply_pool) = build_manager(1);

        let update_timeout = |minutes| {
            let mut change = ConfigChange::new();
            change.insert(
                "local_file_gc_timeout".to_owned(),
                ConfigValue::Duration(ReadableDuration::minutes(minutes).as_millis()),
            );
            change
        };

        manager.dispatch(update_timeout(60)).unwrap();
        manager.dispatch(update_timeout(40)).unwrap();
        assert_eq!(
            manager.get_config().local_file_gc_timeout,
            ReadableDuration::minutes(40)
        );

        let err = manager
            .dispatch(update_timeout(20))
            .unwrap_err()
            .to_string();
        assert!(err.contains("must not be less than its startup value 30m"));
        assert_eq!(
            manager.get_config().local_file_gc_timeout,
            ReadableDuration::minutes(40)
        );

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_updates_config_and_limiter_on_success() {
        let (mut manager, limiter, apply_pool) = build_manager(1);
        let mut active_resolve_change = ConfigChange::new();
        active_resolve_change.insert("concurrency".to_owned(), ConfigValue::Usize(4));
        let mut change = ConfigChange::new();
        change.insert(
            "txn_file_lock_active_resolve".to_owned(),
            ConfigValue::Module(active_resolve_change),
        );

        manager.dispatch(change).unwrap();

        assert_eq!(
            manager
                .get_config()
                .txn_file_lock_active_resolve
                .concurrency,
            4
        );
        assert_eq!(limiter.configured_concurrency(), 4);
        assert_eq!(limiter.available_permits(), 4);

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_cold_region_policy_transitions() {
        fn policy_change(policy: ColdRegionPolicy) -> ConfigChange {
            let mut cold_region_change = ConfigChange::new();
            cold_region_change.insert("policy".to_owned(), policy.into());
            let mut change = ConfigChange::new();
            change.insert(
                "cold_region".to_owned(),
                ConfigValue::Module(cold_region_change),
            );
            change
        }

        let (mut manager, _limiter, apply_pool) = build_manager(1);
        let err = manager
            .dispatch(policy_change(ColdRegionPolicy::Enabled))
            .unwrap_err()
            .to_string();
        assert!(err.contains("disabled -> enable-follower and enable-follower <-> enabled"));
        assert_eq!(
            manager.get_config().cold_region.policy,
            ColdRegionPolicy::Disabled
        );

        manager
            .config
            .update(|cfg| {
                cfg.cold_region.region_heartbeat_interval = ReadableDuration::millis(0);
                Ok::<(), std::convert::Infallible>(())
            })
            .unwrap();
        let err = manager
            .dispatch(policy_change(ColdRegionPolicy::EnableFollower))
            .unwrap_err()
            .to_string();
        assert!(err.contains("region-heartbeat-interval"));
        assert_eq!(
            manager.get_config().cold_region.policy,
            ColdRegionPolicy::Disabled
        );
        manager
            .config
            .update(|cfg| {
                cfg.cold_region.region_heartbeat_interval = ReadableDuration::minutes(5);
                Ok::<(), std::convert::Infallible>(())
            })
            .unwrap();

        manager
            .dispatch(policy_change(ColdRegionPolicy::EnableFollower))
            .unwrap();
        assert_eq!(
            manager.get_config().cold_region.policy,
            ColdRegionPolicy::EnableFollower
        );

        let split_check_change = [(
            "split_region_check_tick_interval".to_owned(),
            ConfigValue::Duration(0),
        )]
        .into();
        let err = manager
            .dispatch(split_check_change)
            .unwrap_err()
            .to_string();
        assert!(err.contains("split-region-check-tick-interval"));

        manager
            .dispatch(policy_change(ColdRegionPolicy::Enabled))
            .unwrap();
        assert_eq!(
            manager.get_config().cold_region.policy,
            ColdRegionPolicy::Enabled
        );

        manager
            .dispatch(policy_change(ColdRegionPolicy::EnableFollower))
            .unwrap();
        assert_eq!(
            manager.get_config().cold_region.policy,
            ColdRegionPolicy::EnableFollower
        );

        let err = manager
            .dispatch(policy_change(ColdRegionPolicy::Disabled))
            .unwrap_err()
            .to_string();
        assert!(err.contains("disabled -> enable-follower and enable-follower <-> enabled"));
        assert_eq!(
            manager.get_config().cold_region.policy,
            ColdRegionPolicy::EnableFollower
        );

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_preserves_unrelated_apply_pool_updates() {
        let (mut manager, limiter, apply_pool) = build_manager(3);

        let mut apply_change = ConfigChange::new();
        apply_change.insert("pool_size".to_owned(), ConfigValue::Usize(1));
        let mut change = ConfigChange::new();
        change.insert(
            "apply_batch_system".to_owned(),
            ConfigValue::Module(apply_change),
        );

        manager.dispatch(change).unwrap();

        assert_eq!(manager.get_config().apply_batch_system.pool_size, 1);
        assert_eq!(apply_pool.get_pool_size(), 1);
        assert_eq!(limiter.configured_concurrency(), 3);

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_updates_disk_hang_timeout() {
        let (mut manager, limiter, apply_pool) = build_manager(1);
        let old_cfg = manager.get_config();
        let mut new_cfg = old_cfg.clone();
        new_cfg.disk_hang_timeout = Some(ReadableDuration::secs(30));

        let change: ConfigChange = old_cfg.diff(&new_cfg);
        assert!(change.contains_key("disk_hang_timeout"));

        manager.dispatch(change).unwrap();

        assert_eq!(
            manager.get_config().disk_hang_timeout,
            Some(ReadableDuration::secs(30))
        );
        assert_eq!(limiter.configured_concurrency(), 1);

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_allows_zero_disk_hang_timeout() {
        let (mut manager, limiter, apply_pool) = build_manager(1);
        let old_cfg = manager.get_config();
        let mut new_cfg = old_cfg.clone();
        new_cfg.disk_hang_timeout = Some(ReadableDuration::secs(0));

        let change: ConfigChange = old_cfg.diff(&new_cfg);
        assert!(change.contains_key("disk_hang_timeout"));

        manager.dispatch(change).unwrap();

        assert_eq!(
            manager.get_config().disk_hang_timeout,
            Some(ReadableDuration::secs(0))
        );
        assert_eq!(limiter.configured_concurrency(), 1);

        apply_pool.shutdown();
    }

    #[test]
    fn test_dispatch_rejects_invalid_disk_hang_timeout() {
        let (mut manager, limiter, apply_pool) = build_manager(1);
        let old_cfg = manager.get_config();
        let mut new_cfg = old_cfg.clone();
        new_cfg.disk_hang_timeout = Some(ReadableDuration::secs(29));

        let change: ConfigChange = old_cfg.diff(&new_cfg);
        assert!(change.contains_key("disk_hang_timeout"));

        let err = manager.dispatch(change).unwrap_err().to_string();
        assert!(err.contains("raftstore.disk-hang-timeout must be at least"));
        assert_eq!(
            manager.get_config().disk_hang_timeout,
            old_cfg.disk_hang_timeout
        );
        assert_eq!(limiter.configured_concurrency(), 1);

        apply_pool.shutdown();
    }
}