// Copyright 2022 PingCAP, Inc.

use std::path::Path;

use online_config::{ConfigChange, ConfigValue, OnlineConfig};
use tikv_util::{
    config::{AbsoluteOrPercentSize, ReadableDuration, ReadableSize},
    warn,
};

use crate::{
    DEFAULT_SNAPSHOT_RLOG_MULTIPART_UPLOAD_PART_SIZE,
    DEFAULT_SNAPSHOT_RLOG_MULTIPART_UPLOAD_THRESHOLD, MAX_EPOCH_ROTATE_LEN,
    MAX_SNAPSHOT_RLOG_MULTIPART_UPLOAD_PART_SIZE, MAX_SNAPSHOT_RLOG_OBJECT_SIZE,
    MIN_EPOCH_ROTATE_LEN,
};

/// Maximum uncompressed write batch accepted by the production WAL writer.
///
/// Tests can override the corresponding private config field with a smaller
/// value.
pub const WAL_BATCH_SIZE_LIMIT: ReadableSize = ReadableSize::mb(256);

#[derive(Clone, Serialize, Deserialize, PartialEq, Debug)]
#[serde(default)]
#[serde(rename_all = "kebab-case")]
pub struct Config {
    /// Compress a log batch if its size exceeds this value. Setting it to zero
    /// disables compression.
    ///
    /// Default: "8KB"
    pub batch_compression_threshold: ReadableSize,
    /// Target file size for rotating log files.
    ///
    /// Default: "512MB"
    pub target_file_size: ReadableSize,

    /// Limit the worker io
    ///
    /// Default: "125MB"
    #[deprecated = "The configuration has been discarded."]
    pub worker_rate_limit: ReadableSize,

    /// The directory to store the wal files for synchronous write.
    /// It's used to reduce the latency of writing wal.
    pub wal_sync_dir: String,

    /// The directory to support double write for low tail latency.
    pub wal_secondary_dir: String,

    /// One of the double writer becomes unhealthy when its fall behind size
    /// exceeds this value.
    pub wal_double_write_unhealthy_size: ReadableSize,

    /// Whether to enable the lightweight backup.
    ///
    /// Default: false
    pub lightweight_backup: bool,

    /// Target file size for wal chunk files.
    ///
    /// Default: "64MB"
    pub wal_chunk_target_file_size: ReadableSize,

    /// Minimum snapshot rlog object size to use DFS multipart upload.
    ///
    /// 0 disables multipart upload for snapshot rlog backups.
    ///
    /// Default: "128MB"
    pub snapshot_rlog_multipart_upload_threshold: ReadableSize,

    /// Part size for snapshot rlog multipart upload.
    ///
    /// Must be non-zero when snapshot rlog multipart upload is enabled.
    ///
    /// Default: "64MB"
    pub snapshot_rlog_multipart_upload_part_size: ReadableSize,

    /// Max concurrent part uploads for snapshot rlog multipart upload.
    ///
    /// Default: 4
    pub snapshot_rlog_multipart_upload_concurrency: usize,

    /// Maximum snapshot rlog object size.
    ///
    /// Default: "10GB"
    pub snapshot_rlog_max_object_size: ReadableSize,

    /// Open RfEngine in cli mode used by tools. Skip serde for this field to
    /// avoid misconfiguration. If cli_mode is set, the sync WAL writer will
    /// avoid sync to improve performance.
    ///
    /// Default: false
    #[serde(skip)]
    pub cli_mode: bool,

    /// Open RfEngine without compaction. This is used for test and debug.
    ///
    /// Default: false
    #[serde(skip)]
    pub disable_compaction: bool,

    /// Rlog file cache capacity. The cache is populated independently of
    /// `lightweight_backup`. Set to zero to disable the cache.
    ///
    /// Default: "128MB"
    pub rlog_cache_capacity: ReadableSize,

    /// Size of rlog files not more that this threshold will be cached in memory
    /// for performance.
    ///
    /// Must be non-zero when `rlog_cache_capacity` is non-zero.
    ///
    /// Default: "4KB"
    pub rlog_cache_size_threshold: ReadableSize,

    /// Memory limit of DFS worker.
    ///
    /// Default: "10%"
    pub dfs_worker_memory_limit: AbsoluteOrPercentSize,

    /// throttle duration per write when WAL compaction is lagging.
    pub write_throttle_duration: ReadableDuration,

    /// Size limit for the rlog data section in a single rlog file.
    ///
    /// During rlog generation, a new file is started once the data
    /// section exceeds this limit.
    ///
    /// Valid range: [16MB, 4GB]. Default: "256MB"
    pub rlog_file_size: ReadableSize,

    /// The number of threads to compact WAL concurrently.
    ///
    /// Must be non-zero.
    pub compaction_concurrency: usize,

    /// The target peer count of each balanced subcompaction task when
    /// compacting WAL files.
    ///
    /// Must be non-zero.
    pub compaction_batch_size: usize,

    /// The number of bytes per second to compact WAL files.
    pub compact_bytes_per_sec: ReadableSize,

    /// Enable rate limiter for compact WAL files.
    pub enable_compact_rate_limiter: bool,

    /// WAL file will be rotated and overwritten on every `epoch_rotate_len`
    /// epoches.
    pub epoch_rotate_len: usize,

    /// Delay compaction with latest truncated index to reduce generated rlog
    /// file size.
    pub delay_compaction_epoches: usize,

    /// The max gap between now and last WAL chunk updated.
    /// If `None`, will only flush chunks to S3 when the size limit reached or
    /// shutting down.
    pub max_wal_chunk_gap_duration: Option<ReadableDuration>,

    /// modified by test only, skip serde to avoid misconfiguration.
    #[serde(skip)]
    pub(crate) max_batch_size: ReadableSize,

    /// Soft memory limit for raft log entries.
    ///
    /// Controls the rlog offloading mechanism, which tries to keep in-memory
    /// raft entries under this limit by offloading entries associated with
    /// older WAL files to disk. Due to factors such as WAL compression, this
    /// mechanism is best-effort and the limit should be considered soft.
    ///
    /// This value is converted into the maximum number of WAL epochs that can
    /// be kept in memory by dividing it by the WAL file size (target_file_size)
    /// and rounding up. In other words, the effective in-memory rlog limit
    /// is always an integer multiple of the WAL file size.
    ///
    /// It is recommended to set this value to at least 4× the WAL file size.
    /// Smaller values are allowed, but rlog offloading may be delayed until the
    /// corresponding rlog files are generated (i.e., after WAL compaction).
    ///
    /// Default: "4GB"
    pub rlog_soft_memory_limit: ReadableSize,

    /// When set, ensure at least one WAL rotation and snapshot upload happens
    /// within this interval.
    ///
    /// This helps lightweight backup avoid referencing snapshots that may have
    /// already expired.
    ///
    /// Default: None
    pub force_rotate_interval: Option<ReadableDuration>,
}

impl<'a> OnlineConfig<'a> for Config {
    type Encoder = &'a Self;

    fn diff(&self, incoming: &Self) -> ConfigChange {
        let mut diff = ConfigChange::new();
        if self.lightweight_backup != incoming.lightweight_backup {
            diff.insert(
                "lightweight_backup".to_owned(),
                ConfigValue::Bool(incoming.lightweight_backup),
            );
        }
        diff
    }

    fn update(&mut self, mut incoming: ConfigChange) -> online_config::Result<()> {
        if let Some(value) = incoming.remove("lightweight_backup") {
            self.lightweight_backup = value.into();
        }
        Ok(())
    }

    fn get_encoder(&'a self) -> Self::Encoder {
        self
    }

    fn typed(&self) -> ConfigChange {
        ConfigChange::from([(
            "lightweight_backup".to_owned(),
            ConfigValue::Bool(self.lightweight_backup),
        )])
    }
}

#[allow(deprecated)]
impl Default for Config {
    fn default() -> Self {
        Self {
            batch_compression_threshold: ReadableSize::kb(8),
            target_file_size: ReadableSize::mb(512),
            worker_rate_limit: ReadableSize::mb(125),
            wal_sync_dir: "".to_owned(),
            wal_secondary_dir: "".to_owned(),
            wal_double_write_unhealthy_size: ReadableSize::mb(256),
            lightweight_backup: false,
            wal_chunk_target_file_size: ReadableSize::mb(64),
            snapshot_rlog_multipart_upload_threshold: ReadableSize(
                DEFAULT_SNAPSHOT_RLOG_MULTIPART_UPLOAD_THRESHOLD,
            ),
            snapshot_rlog_multipart_upload_part_size: ReadableSize(
                DEFAULT_SNAPSHOT_RLOG_MULTIPART_UPLOAD_PART_SIZE as u64,
            ),
            snapshot_rlog_multipart_upload_concurrency: 4,
            snapshot_rlog_max_object_size: ReadableSize(MAX_SNAPSHOT_RLOG_OBJECT_SIZE),
            cli_mode: false,
            disable_compaction: false,
            rlog_cache_capacity: ReadableSize::mb(128),
            rlog_cache_size_threshold: ReadableSize::kb(4),
            dfs_worker_memory_limit: AbsoluteOrPercentSize::Percent(10.0),
            write_throttle_duration: ReadableDuration::millis(10),
            rlog_file_size: ReadableSize::mb(256),
            compaction_concurrency: 2,
            compaction_batch_size: 64,
            compact_bytes_per_sec: ReadableSize::mb(200),
            enable_compact_rate_limiter: false,
            epoch_rotate_len: MIN_EPOCH_ROTATE_LEN,
            delay_compaction_epoches: 0,
            max_wal_chunk_gap_duration: None,
            max_batch_size: WAL_BATCH_SIZE_LIMIT,
            rlog_soft_memory_limit: ReadableSize::gb(4),
            force_rotate_interval: None,
        }
    }
}

impl Config {
    pub fn validate(&mut self, dir: &str) -> Result<(), String> {
        if self.rlog_cache_capacity.0 > 0 && self.rlog_cache_size_threshold.0 == 0 {
            return Err(
                "rfengine rlog_cache_size_threshold must be non-zero when rlog_cache_capacity is non-zero"
                    .to_owned(),
            );
        }
        if self.compaction_concurrency == 0 {
            return Err(
                "rfengine compaction_concurrency is invalid, it must be at least 1".to_owned(),
            );
        }
        if self.compaction_batch_size == 0 {
            return Err(
                "rfengine compaction_batch_size is invalid, it must be at least 1".to_owned(),
            );
        }
        if self.snapshot_rlog_multipart_upload_threshold.0 > 0 {
            if self.snapshot_rlog_multipart_upload_part_size.0 == 0 {
                return Err(
                    "rfengine snapshot_rlog_multipart_upload_part_size is invalid, it must be non-zero when snapshot_rlog_multipart_upload_threshold is non-zero"
                    .to_owned(),
                );
            }
            if self.snapshot_rlog_multipart_upload_part_size.0
                > MAX_SNAPSHOT_RLOG_MULTIPART_UPLOAD_PART_SIZE
            {
                return Err(format!(
                    "rfengine snapshot_rlog_multipart_upload_part_size {} is invalid, it must be at most {} bytes",
                    self.snapshot_rlog_multipart_upload_part_size.0,
                    MAX_SNAPSHOT_RLOG_MULTIPART_UPLOAD_PART_SIZE
                ));
            }
            if self.snapshot_rlog_multipart_upload_concurrency == 0 {
                return Err(
                    "rfengine snapshot_rlog_multipart_upload_concurrency is invalid, it must be at least 1 when snapshot_rlog_multipart_upload_threshold is non-zero"
                        .to_owned(),
                );
            }
        }
        if self.snapshot_rlog_max_object_size.0 == 0 {
            return Err(
                "rfengine snapshot_rlog_max_object_size is invalid, it must be non-zero".to_owned(),
            );
        }
        // make sure epoch_rotate_len is valid.
        if self.epoch_rotate_len < MIN_EPOCH_ROTATE_LEN
            || self.epoch_rotate_len > MAX_EPOCH_ROTATE_LEN
        {
            return Err(format!(
                "rfengine epoch_rotate_len {} is invalid, it must be between {} and {}",
                self.epoch_rotate_len, MIN_EPOCH_ROTATE_LEN, MAX_EPOCH_ROTATE_LEN
            ));
        };
        // do not modify epoch_rotate_len if there are existing wal files.
        let wal_file_count = wal_file_count(Path::new(dir));
        if wal_file_count > 0 && wal_file_count != self.epoch_rotate_len {
            if wal_file_count < MIN_EPOCH_ROTATE_LEN {
                return Err(format!(
                    "rfengine wal_file_count {} is less than {}, maybe interrupted on initialization",
                    wal_file_count, MIN_EPOCH_ROTATE_LEN
                ));
            }
            // The existing wal files count is different from the configured
            // epoch_rotate_len, we keep the existing value.
            warn!(
                "rfengine epoch_rotate_len is modified from {} to {}, as there are existing wal files",
                self.epoch_rotate_len, wal_file_count
            );
            self.epoch_rotate_len = wal_file_count;
            if self.delay_compaction_epoches + MIN_EPOCH_ROTATE_LEN > self.epoch_rotate_len {
                // adjust delay_compaction_epoches as well.
                self.delay_compaction_epoches = self.epoch_rotate_len - MIN_EPOCH_ROTATE_LEN;
            }
        }
        if self.delay_compaction_epoches + MIN_EPOCH_ROTATE_LEN > self.epoch_rotate_len {
            return Err(format!(
                "invalid delay_compaction_epoches {} with epoch_rotate_len {}",
                self.delay_compaction_epoches, self.epoch_rotate_len
            ));
        }
        Ok(())
    }

    pub fn disable_rlog_offloading(&mut self) {
        self.rlog_soft_memory_limit = ReadableSize::mb(0);
    }

    pub fn rlog_offloading_enabled(&self) -> bool {
        self.rlog_soft_memory_limit > ReadableSize::mb(0)
    }
}

pub(crate) fn wal_file_count(dir: &Path) -> usize {
    let mut count = 0;
    loop {
        let file_path = crate::engine::wal_file_path(dir, count);
        if !file_path.exists() {
            break;
        }
        count += 1;
    }
    count
}

#[cfg(test)]
impl Config {
    pub(crate) fn new(target_file_size: usize) -> Self {
        let mut cfg = Self::default();
        cfg.target_file_size = ReadableSize(target_file_size as u64);
        cfg
    }
}

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

    #[test]
    fn test_default_snapshot_rlog_multipart_upload_threshold() {
        let cfg = Config::default();
        assert_eq!(
            cfg.snapshot_rlog_multipart_upload_threshold.0,
            DEFAULT_SNAPSHOT_RLOG_MULTIPART_UPLOAD_THRESHOLD
        );
        assert_eq!(
            cfg.snapshot_rlog_multipart_upload_threshold,
            ReadableSize::mb(128)
        );
        assert_eq!(
            cfg.snapshot_rlog_multipart_upload_part_size,
            ReadableSize::mb(64)
        );
        assert_eq!(cfg.snapshot_rlog_multipart_upload_concurrency, 4);
        assert_eq!(cfg.snapshot_rlog_max_object_size, ReadableSize::gb(10));
    }

    #[test]
    fn test_config_validate() {
        let tmp_dir = tempfile::Builder::new().tempdir().unwrap();
        let base_path = tmp_dir.path();

        // Test 1: Valid default configuration
        let mut cfg = Config::default();
        cfg.validate(base_path.to_str().unwrap()).unwrap();

        // Test 2: Invalid epoch_rotate_len - too low
        let mut cfg = Config::default();
        cfg.epoch_rotate_len = MIN_EPOCH_ROTATE_LEN - 1;
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Test 3: Invalid epoch_rotate_len - too high
        let mut cfg = Config::default();
        cfg.epoch_rotate_len = MAX_EPOCH_ROTATE_LEN + 1;
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Test 4: Invalid compaction_concurrency - zero
        let mut cfg = Config::default();
        cfg.compaction_concurrency = 0;
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Test 5: Invalid compaction_batch_size - zero
        let mut cfg = Config::default();
        cfg.compaction_batch_size = 0;
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // An enabled rlog cache needs a non-zero size threshold.
        let mut cfg = Config::default();
        cfg.rlog_cache_size_threshold = ReadableSize(0);
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Zero capacity disables the rlog cache and does not need a threshold.
        let mut cfg = Config::default();
        cfg.rlog_cache_capacity = ReadableSize(0);
        cfg.rlog_cache_size_threshold = ReadableSize(0);
        cfg.validate(base_path.to_str().unwrap()).unwrap();

        // Invalid snapshot rlog max object size - zero
        let mut cfg = Config::default();
        cfg.snapshot_rlog_max_object_size = ReadableSize(0);
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Invalid snapshot rlog multipart part size - zero when multipart is enabled
        let mut cfg = Config::default();
        cfg.snapshot_rlog_multipart_upload_part_size = ReadableSize(0);
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Invalid snapshot rlog multipart part size - above the local buffer limit
        let mut cfg = Config::default();
        cfg.snapshot_rlog_multipart_upload_part_size =
            ReadableSize(MAX_SNAPSHOT_RLOG_MULTIPART_UPLOAD_PART_SIZE + 1);
        assert!(cfg.validate(base_path.to_str().unwrap()).is_err());

        // Test 6: Existing WAL files - should adjust epoch_rotate_len
        let existing_files_path = base_path.join("existing_files");
        std::fs::create_dir(&existing_files_path).unwrap();
        create_wal_files(&existing_files_path, 6);

        let mut cfg = Config::default();
        cfg.epoch_rotate_len = 4; // Different from existing file count
        cfg.validate(existing_files_path.to_str().unwrap()).unwrap();
        assert_eq!(cfg.epoch_rotate_len, 6); // Should be adjusted to match existing files

        // Test interruption scenarios - WAL file count less than MIN_EPOCH_ROTATE_LEN
        test_interruption_scenario(base_path, "interruption_0_files", 0, true);
        test_interruption_scenario(base_path, "interruption_1_file", 1, false);
        test_interruption_scenario(base_path, "interruption_3_files", 3, false);
        test_interruption_scenario(base_path, "interruption_2_files", 2, false);
    }

    fn create_wal_files(dir: &Path, count: usize) {
        for i in 0..count {
            let wal_path = crate::engine::wal_file_path(dir, i);
            std::fs::write(&wal_path, b"test data").unwrap();
        }
    }

    fn test_interruption_scenario(
        base_path: &Path,
        subdir_name: &str,
        existing_files: usize,
        should_pass: bool,
    ) {
        let test_path = base_path.join(subdir_name);
        std::fs::create_dir(&test_path).unwrap();
        create_wal_files(&test_path, existing_files);

        let mut cfg = Config::default();
        let result = cfg.validate(test_path.to_str().unwrap());

        if should_pass {
            assert!(
                result.is_ok(),
                "Expected validation to pass for {} files",
                existing_files
            );
        } else {
            assert!(
                result.is_err(),
                "Expected validation to fail for {} files (less than MIN_EPOCH_ROTATE_LEN)",
                existing_files
            );
        }
    }
}