// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use std::{
    collections::HashSet,
    fs,
    path::{Path, PathBuf},
    time::Duration,
};

use anyhow::{Context, Result, bail, ensure};
use serde::{Deserialize, Serialize};
use tikv_util::config::ReadableDuration;

const MAX_SLICE_PRECHARGE_US: u64 = 1_000_000;
const YATP_SIM_ROOT: &str = env!("CARGO_MANIFEST_DIR");
const GLOBAL_OVERRIDE_FIELDS: &str = concat!(
    "duration, warmup-duration, stats-print-interval, random-seed, drain-on-stop, ",
    "cooldown-duration, report-path, summary-path, read-pool.workers, ",
    "read-pool.keyspace-fairness.enabled, read-pool.keyspace-fairness.use-wall-time, ",
    "read-pool.keyspace-fairness.slice-precharge-us, ",
    "read-pool.fair-global-cpu-admission.enabled, ",
    "read-pool.fair-global-cpu-admission.max-cpu-cores, ",
    "read-pool.fair-global-cpu-admission.refill-interval, ",
    "read-pool.fair-global-cpu-admission.admit-interval-us, ",
    "read-pool.fair-global-cpu-admission.request-timeout"
);
const KEYSPACE_OVERRIDE_FIELDS: &str = "count";
const TASK_OVERRIDE_FIELDS: &str = concat!(
    "enabled, arrival-mode, qps, concurrency, task-cpu-us, task-cpu-jitter-us, slice-cpu-us, ",
    "slice-cpu-jitter-us, blocking-wall-us, blocking-wall-jitter-us, priority, ",
    "start-time, duration, slow-threshold"
);

/// A command-line assignment applied after the TOML file is deserialized.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ConfigOverride {
    /// Override a top-level or `read-pool.*` field.
    Global(String),
    /// Override a keyspace group selected by its configured name.
    Keyspace { group: String, assignment: String },
    /// Override a task selected by its keyspace group and configured task name.
    Task {
        group: String,
        task: String,
        assignment: String,
    },
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
pub struct SimulatorConfig {
    #[serde(default = "default_simulation_duration")]
    pub duration: ReadableDuration,
    #[serde(default = "zero_duration")]
    pub warmup_duration: ReadableDuration,
    #[serde(default = "default_stats_print_interval")]
    pub stats_print_interval: ReadableDuration,
    #[serde(default = "default_random_seed")]
    pub random_seed: u64,
    #[serde(default = "default_true")]
    pub drain_on_stop: bool,
    #[serde(default = "default_cooldown_duration")]
    pub cooldown_duration: ReadableDuration,
    #[serde(default = "default_report_path")]
    pub report_path: PathBuf,
    #[serde(default = "default_summary_path")]
    pub summary_path: PathBuf,
    #[serde(default)]
    pub read_pool: ReadPoolConfig,
    pub keyspace_groups: Vec<KeyspaceGroupConfig>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
pub struct ReadPoolConfig {
    #[serde(default = "default_workers")]
    pub workers: usize,
    #[serde(default)]
    pub keyspace_fairness: KeyspaceFairnessConfig,
    #[serde(default)]
    pub fair_global_cpu_admission: FairGlobalCpuAdmissionConfig,
}

impl Default for ReadPoolConfig {
    fn default() -> Self {
        Self {
            workers: default_workers(),
            keyspace_fairness: KeyspaceFairnessConfig::default(),
            fair_global_cpu_admission: FairGlobalCpuAdmissionConfig::default(),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
pub struct KeyspaceFairnessConfig {
    #[serde(default)]
    pub enabled: bool,
    #[serde(default = "default_true")]
    pub use_wall_time: bool,
    #[serde(default)]
    pub slice_precharge_us: u64,
}

impl Default for KeyspaceFairnessConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            use_wall_time: true,
            slice_precharge_us: 0,
        }
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
pub struct FairGlobalCpuAdmissionConfig {
    #[serde(default)]
    pub enabled: bool,
    #[serde(default = "default_max_cpu_cores")]
    pub max_cpu_cores: f64,
    #[serde(default = "default_cpu_refill_interval")]
    pub refill_interval: ReadableDuration,
    #[serde(default = "default_admit_interval_us")]
    pub admit_interval_us: u64,
    #[serde(default = "default_admission_request_timeout")]
    pub request_timeout: ReadableDuration,
}

impl Default for FairGlobalCpuAdmissionConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            max_cpu_cores: default_max_cpu_cores(),
            refill_interval: default_cpu_refill_interval(),
            admit_interval_us: default_admit_interval_us(),
            request_timeout: default_admission_request_timeout(),
        }
    }
}

impl FairGlobalCpuAdmissionConfig {
    /// Returns the global CPU token capacity for one refill interval.
    pub fn global_cpu_quota_us(&self) -> Result<u64> {
        ensure!(
            self.max_cpu_cores.is_finite() && self.max_cpu_cores > 0.0,
            "read-pool.fair-global-cpu-admission.max-cpu-cores must be finite and greater than zero"
        );
        let refill_us = u64::try_from(self.refill_interval.0.as_micros())
            .context("read-pool.fair-global-cpu-admission.refill-interval is too large")?;
        let quota_us = self.max_cpu_cores * refill_us as f64;
        ensure!(
            quota_us.is_finite() && quota_us >= 1.0 && quota_us <= u64::MAX as f64,
            "read-pool.fair-global-cpu-admission max CPU produces an invalid token quota"
        );
        Ok(quota_us as u64)
    }
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
pub struct KeyspaceGroupConfig {
    pub name: String,
    #[serde(default = "default_group_count")]
    pub count: usize,
    #[serde(default)]
    pub tasks: Vec<TaskConfig>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "kebab-case", deny_unknown_fields)]
pub struct TaskConfig {
    pub name: String,
    #[serde(default = "default_true")]
    pub enabled: bool,
    pub arrival_mode: ArrivalMode,
    #[serde(default)]
    pub qps: Option<f64>,
    pub concurrency: usize,
    pub task_cpu_us: u64,
    #[serde(default)]
    pub task_cpu_jitter_us: u64,
    pub slice_cpu_us: u64,
    #[serde(default)]
    pub slice_cpu_jitter_us: u64,
    #[serde(default)]
    pub blocking_wall_us: u64,
    #[serde(default)]
    pub blocking_wall_jitter_us: u64,
    #[serde(default)]
    pub priority: Priority,
    #[serde(default = "zero_duration")]
    pub start_time: ReadableDuration,
    #[serde(default = "zero_duration")]
    pub duration: ReadableDuration,
    #[serde(default = "zero_duration")]
    pub slow_threshold: ReadableDuration,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum ArrivalMode {
    OpenLoop,
    ClosedLoop,
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum Priority {
    High,
    #[default]
    Normal,
    Low,
}

impl SimulatorConfig {
    /// Loads a TOML config, applies command-line overrides, and validates the
    /// resulting effective configuration.
    pub fn load(path: &Path, overrides: &[ConfigOverride]) -> Result<Self> {
        let path = resolve_config_path(path);
        let source = fs::read_to_string(&path)
            .with_context(|| format!("failed to read config {}", path.display()))?;
        let config: Self = toml::from_str(&source)
            .with_context(|| format!("failed to parse config {}", path.display()))?;
        config.apply_overrides_and_validate(overrides)
    }

    #[cfg(test)]
    fn parse(source: &str) -> Result<Self> {
        Self::parse_with_overrides(source, &[])
    }

    #[cfg(test)]
    fn parse_with_overrides(source: &str, overrides: &[ConfigOverride]) -> Result<Self> {
        let config: Self = toml::from_str(source)?;
        config.apply_overrides_and_validate(overrides)
    }

    fn apply_overrides_and_validate(mut self, overrides: &[ConfigOverride]) -> Result<Self> {
        for config_override in overrides {
            self.apply_override(config_override)?;
        }
        self.validate()?;
        Ok(self)
    }

    fn apply_override(&mut self, config_override: &ConfigOverride) -> Result<()> {
        match config_override {
            ConfigOverride::Global(assignment) => self.apply_global_override(assignment),
            ConfigOverride::Keyspace { group, assignment } => {
                self.apply_keyspace_override(group, assignment)
            }
            ConfigOverride::Task {
                group,
                task,
                assignment,
            } => self.apply_task_override(group, task, assignment),
        }
    }

    fn apply_global_override(&mut self, assignment: &str) -> Result<()> {
        let (field, value) = split_assignment("--set", assignment)?;
        match field {
            "duration" => self.duration = parse_override_value(value, field)?,
            "warmup-duration" => self.warmup_duration = parse_override_value(value, field)?,
            "stats-print-interval" => {
                self.stats_print_interval = parse_override_value(value, field)?
            }
            "random-seed" => self.random_seed = parse_override_value(value, field)?,
            "drain-on-stop" => self.drain_on_stop = parse_override_value(value, field)?,
            "cooldown-duration" => self.cooldown_duration = parse_override_value(value, field)?,
            "report-path" => self.report_path = PathBuf::from(value),
            "summary-path" => self.summary_path = PathBuf::from(value),
            "read-pool.workers" => self.read_pool.workers = parse_override_value(value, field)?,
            "read-pool.keyspace-fairness.enabled" => {
                self.read_pool.keyspace_fairness.enabled = parse_override_value(value, field)?
            }
            "read-pool.keyspace-fairness.use-wall-time" => {
                self.read_pool.keyspace_fairness.use_wall_time = parse_override_value(value, field)?
            }
            "read-pool.keyspace-fairness.slice-precharge-us" => {
                self.read_pool.keyspace_fairness.slice_precharge_us =
                    parse_override_value(value, field)?
            }
            "read-pool.fair-global-cpu-admission.enabled" => {
                self.read_pool.fair_global_cpu_admission.enabled =
                    parse_override_value(value, field)?
            }
            "read-pool.fair-global-cpu-admission.max-cpu-cores" => {
                self.read_pool.fair_global_cpu_admission.max_cpu_cores =
                    parse_override_value(value, field)?
            }
            "read-pool.fair-global-cpu-admission.refill-interval" => {
                self.read_pool.fair_global_cpu_admission.refill_interval =
                    parse_override_value(value, field)?
            }
            "read-pool.fair-global-cpu-admission.admit-interval-us" => {
                self.read_pool.fair_global_cpu_admission.admit_interval_us =
                    parse_override_value(value, field)?
            }
            "read-pool.fair-global-cpu-admission.request-timeout" => {
                self.read_pool.fair_global_cpu_admission.request_timeout =
                    parse_override_value(value, field)?
            }
            _ => bail!("unknown --set field {field:?}; supported fields: {GLOBAL_OVERRIDE_FIELDS}"),
        }
        Ok(())
    }

    fn apply_keyspace_override(&mut self, group_name: &str, assignment: &str) -> Result<()> {
        let (field, value) = split_assignment("--set-keyspace", assignment)?;
        let group = find_keyspace_group_mut(&mut self.keyspace_groups, group_name)?;
        apply_keyspace_field(group, field, value).with_context(|| {
            format!("failed to apply --set-keyspace {assignment:?} to group {group_name:?}")
        })
    }

    fn apply_task_override(
        &mut self,
        group_name: &str,
        task_name: &str,
        assignment: &str,
    ) -> Result<()> {
        let (field, value) = split_assignment("--set-task", assignment)?;
        let group = find_keyspace_group_mut(&mut self.keyspace_groups, group_name)?;
        let task = find_task_mut(group, task_name)?;
        apply_task_field(task, field, value).with_context(|| {
            format!(
                "failed to apply --set-task {assignment:?} to task {group_name:?}/{task_name:?}"
            )
        })
    }

    fn validate(&self) -> Result<()> {
        ensure!(
            self.duration.0 > Duration::ZERO,
            "duration must be greater than zero"
        );
        ensure!(
            self.warmup_duration.0 < self.duration.0,
            "warmup-duration must be shorter than duration"
        );
        ensure!(
            self.stats_print_interval.0 > Duration::ZERO,
            "stats-print-interval must be greater than zero"
        );
        ensure!(
            !self.report_path.as_os_str().is_empty(),
            "report-path must not be empty"
        );
        ensure!(
            !self.summary_path.as_os_str().is_empty(),
            "summary-path must not be empty"
        );
        ensure!(
            self.report_path != self.summary_path,
            "report-path and summary-path must be different"
        );
        ensure!(
            self.read_pool.workers > 0,
            "read-pool.workers must be greater than zero"
        );
        ensure!(
            self.read_pool.keyspace_fairness.slice_precharge_us <= MAX_SLICE_PRECHARGE_US,
            "read-pool.keyspace-fairness.slice-precharge-us must not exceed {MAX_SLICE_PRECHARGE_US}"
        );
        let admission = &self.read_pool.fair_global_cpu_admission;
        ensure!(
            admission.refill_interval.0 >= Duration::from_millis(1),
            "read-pool.fair-global-cpu-admission.refill-interval must be at least 1ms"
        );
        ensure!(
            admission.admit_interval_us > 0,
            "read-pool.fair-global-cpu-admission.admit-interval-us must be greater than zero"
        );
        ensure!(
            admission.request_timeout.0 > Duration::ZERO,
            "read-pool.fair-global-cpu-admission.request-timeout must be greater than zero"
        );
        admission.global_cpu_quota_us()?;
        if admission.enabled {
            ensure!(
                self.read_pool.keyspace_fairness.enabled,
                "read-pool.fair-global-cpu-admission requires keyspace fairness"
            );
        }
        ensure!(
            !self.keyspace_groups.is_empty(),
            "at least one keyspace group is required"
        );

        let mut group_names = HashSet::new();
        let mut total_keyspaces = 0usize;
        let mut enabled_task_classes = 0usize;
        for group in &self.keyspace_groups {
            ensure!(
                !group.name.trim().is_empty(),
                "keyspace group name must not be empty"
            );
            ensure!(
                group_names.insert(group.name.as_str()),
                "duplicate keyspace group name {:?}",
                group.name
            );
            ensure!(
                group.count > 0,
                "keyspace group {:?} count must be greater than zero",
                group.name
            );
            total_keyspaces = total_keyspaces
                .checked_add(group.count)
                .context("total keyspace count overflows usize")?;

            let mut task_names = HashSet::new();
            for task in &group.tasks {
                if task.enabled {
                    enabled_task_classes += 1;
                }
                ensure!(
                    !task.name.trim().is_empty(),
                    "task name in group {:?} must not be empty",
                    group.name
                );
                ensure!(
                    task_names.insert(task.name.as_str()),
                    "duplicate task name {:?} in group {:?}",
                    task.name,
                    group.name
                );
                self.validate_task(group, task)?;
            }
        }

        ensure!(
            total_keyspaces <= u32::MAX as usize,
            "total keyspace count must not exceed {}",
            u32::MAX
        );
        ensure!(
            enabled_task_classes > 0,
            "at least one enabled task class is required"
        );
        Ok(())
    }

    fn validate_task(&self, group: &KeyspaceGroupConfig, task: &TaskConfig) -> Result<()> {
        let field = |name: &str| {
            format!(
                "keyspace-groups[{}].tasks[{}].{name}",
                group.name, task.name
            )
        };

        ensure!(
            task.concurrency > 0,
            "{} must be greater than zero",
            field("concurrency")
        );
        ensure!(
            task.task_cpu_us > 0,
            "{} must be greater than zero",
            field("task-cpu-us")
        );
        ensure!(
            task.slice_cpu_us > 0,
            "{} must be greater than zero",
            field("slice-cpu-us")
        );
        checked_jitter_upper(
            task.task_cpu_us,
            task.task_cpu_jitter_us,
            &field("task-cpu-jitter-us"),
        )?;
        checked_jitter_upper(
            task.slice_cpu_us,
            task.slice_cpu_jitter_us,
            &field("slice-cpu-jitter-us"),
        )?;
        checked_jitter_upper(
            task.blocking_wall_us,
            task.blocking_wall_jitter_us,
            &field("blocking-wall-jitter-us"),
        )?;

        ensure!(
            task.start_time.0 < self.duration.0,
            "{} must be shorter than the global duration",
            field("start-time")
        );
        let active_duration = self.task_active_duration(task)?;
        ensure!(
            active_duration > Duration::ZERO,
            "{} has an empty active window",
            field("duration")
        );

        match task.arrival_mode {
            ArrivalMode::OpenLoop => {
                let qps = task
                    .qps
                    .context(format!("{} is required in open-loop mode", field("qps")))?;
                ensure!(
                    qps.is_finite() && qps > 0.0,
                    "{} must be finite and greater than zero",
                    field("qps")
                );
                let offered = qps * active_duration.as_secs_f64();
                ensure!(
                    offered.is_finite() && offered < u64::MAX as f64,
                    "{} produces too many arrivals for its active window",
                    field("qps")
                );
            }
            ArrivalMode::ClosedLoop => {
                ensure!(
                    task.qps.is_none(),
                    "{} is not valid in closed-loop mode",
                    field("qps")
                );
            }
        }
        Ok(())
    }

    pub fn task_end_time(&self, task: &TaskConfig) -> Result<Duration> {
        if task.duration.0 == Duration::ZERO {
            return Ok(self.duration.0);
        }
        Ok(task
            .start_time
            .0
            .checked_add(task.duration.0)
            .context("task start-time plus duration overflows")?
            .min(self.duration.0))
    }

    fn task_active_duration(&self, task: &TaskConfig) -> Result<Duration> {
        self.task_end_time(task)?
            .checked_sub(task.start_time.0)
            .context("task active window is invalid")
    }
}

fn apply_keyspace_field(group: &mut KeyspaceGroupConfig, field: &str, value: &str) -> Result<()> {
    match field {
        "count" => group.count = parse_override_value(value, field)?,
        _ => bail!(
            "unknown --set-keyspace field {field:?}; supported fields: {KEYSPACE_OVERRIDE_FIELDS}"
        ),
    }
    Ok(())
}

fn apply_task_field(task: &mut TaskConfig, field: &str, value: &str) -> Result<()> {
    match field {
        "enabled" => task.enabled = parse_override_value(value, field)?,
        "arrival-mode" => task.arrival_mode = parse_arrival_mode(value, field)?,
        "qps" => {
            task.qps = if value == "null" {
                None
            } else {
                Some(parse_override_value(value, field)?)
            }
        }
        "concurrency" => task.concurrency = parse_override_value(value, field)?,
        "task-cpu-us" => task.task_cpu_us = parse_override_value(value, field)?,
        "task-cpu-jitter-us" => task.task_cpu_jitter_us = parse_override_value(value, field)?,
        "slice-cpu-us" => task.slice_cpu_us = parse_override_value(value, field)?,
        "slice-cpu-jitter-us" => task.slice_cpu_jitter_us = parse_override_value(value, field)?,
        "blocking-wall-us" => task.blocking_wall_us = parse_override_value(value, field)?,
        "blocking-wall-jitter-us" => {
            task.blocking_wall_jitter_us = parse_override_value(value, field)?
        }
        "priority" => task.priority = parse_priority(value, field)?,
        "start-time" => task.start_time = parse_override_value(value, field)?,
        "duration" => task.duration = parse_override_value(value, field)?,
        "slow-threshold" => task.slow_threshold = parse_override_value(value, field)?,
        _ => bail!("unknown --set-task field {field:?}; supported fields: {TASK_OVERRIDE_FIELDS}"),
    }
    Ok(())
}

fn split_assignment<'a>(option: &str, assignment: &'a str) -> Result<(&'a str, &'a str)> {
    let (field, value) = assignment
        .split_once('=')
        .with_context(|| format!("{option} expects FIELD=VALUE, got {assignment:?}"))?;
    ensure!(
        !field.is_empty(),
        "{option} expects a non-empty field before '='"
    );
    Ok((field, value))
}

fn parse_override_value<T>(value: &str, field: &str) -> Result<T>
where
    T: std::str::FromStr,
    T::Err: std::fmt::Display,
{
    value.parse().map_err(|error| {
        anyhow::anyhow!("invalid value {value:?} for override field {field:?}: {error}")
    })
}

fn parse_arrival_mode(value: &str, field: &str) -> Result<ArrivalMode> {
    match value {
        "open-loop" => Ok(ArrivalMode::OpenLoop),
        "closed-loop" => Ok(ArrivalMode::ClosedLoop),
        _ => bail!(
            "invalid value {value:?} for override field {field:?}; expected open-loop or closed-loop"
        ),
    }
}

fn parse_priority(value: &str, field: &str) -> Result<Priority> {
    match value {
        "high" => Ok(Priority::High),
        "normal" => Ok(Priority::Normal),
        "low" => Ok(Priority::Low),
        _ => bail!(
            "invalid value {value:?} for override field {field:?}; expected high, normal, or low"
        ),
    }
}

fn find_keyspace_group_mut<'a>(
    groups: &'a mut [KeyspaceGroupConfig],
    name: &str,
) -> Result<&'a mut KeyspaceGroupConfig> {
    let mut matching_index = None;
    for (index, group) in groups.iter().enumerate() {
        if group.name == name {
            ensure!(
                matching_index.is_none(),
                "keyspace group selector {name:?} is ambiguous because the name is duplicated"
            );
            matching_index = Some(index);
        }
    }

    let index = matching_index.with_context(|| {
        format!(
            "keyspace group {name:?} not found; available groups: {}",
            format_group_names(groups)
        )
    })?;
    Ok(&mut groups[index])
}

fn find_task_mut<'a>(group: &'a mut KeyspaceGroupConfig, name: &str) -> Result<&'a mut TaskConfig> {
    let mut matching_index = None;
    for (index, task) in group.tasks.iter().enumerate() {
        if task.name == name {
            ensure!(
                matching_index.is_none(),
                "task selector {}/{name:?} is ambiguous because the task name is duplicated",
                group.name
            );
            matching_index = Some(index);
        }
    }

    let index = matching_index.with_context(|| {
        format!(
            "task {name:?} not found in keyspace group {:?}; available tasks: {}",
            group.name,
            format_task_names(&group.tasks)
        )
    })?;
    Ok(&mut group.tasks[index])
}

fn format_group_names(groups: &[KeyspaceGroupConfig]) -> String {
    format_names(groups.iter().map(|group| group.name.as_str()))
}

fn format_task_names(tasks: &[TaskConfig]) -> String {
    format_names(tasks.iter().map(|task| task.name.as_str()))
}

fn format_names<'a>(names: impl Iterator<Item = &'a str>) -> String {
    let names = names.map(|name| format!("{name:?}")).collect::<Vec<_>>();
    if names.is_empty() {
        "<none>".to_owned()
    } else {
        names.join(", ")
    }
}

fn resolve_config_path(path: &Path) -> PathBuf {
    if path.is_absolute() || path.exists() {
        return path.to_owned();
    }
    Path::new(YATP_SIM_ROOT).join(path)
}

fn checked_jitter_upper(base: u64, jitter: u64, field: &str) -> Result<u64> {
    base.checked_add(jitter)
        .with_context(|| format!("{field} makes the sampled value overflow u64"))
}

fn default_simulation_duration() -> ReadableDuration {
    ReadableDuration::minutes(1)
}

fn default_random_seed() -> u64 {
    1
}

fn default_stats_print_interval() -> ReadableDuration {
    ReadableDuration::secs(1)
}

fn default_cooldown_duration() -> ReadableDuration {
    ReadableDuration::secs(5)
}

fn default_report_path() -> PathBuf {
    PathBuf::from("target/yatp-sim/report.html")
}

fn default_summary_path() -> PathBuf {
    PathBuf::from("target/yatp-sim/summary.json")
}

fn default_workers() -> usize {
    4
}

fn default_max_cpu_cores() -> f64 {
    4.0
}

fn default_cpu_refill_interval() -> ReadableDuration {
    ReadableDuration::millis(100)
}

fn default_admit_interval_us() -> u64 {
    1_000
}

fn default_admission_request_timeout() -> ReadableDuration {
    ReadableDuration::hours(1)
}

fn default_group_count() -> usize {
    1
}

fn default_true() -> bool {
    true
}

fn zero_duration() -> ReadableDuration {
    ReadableDuration::ZERO
}

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

    const MINIMAL_CLOSED_LOOP: &str = r#"
[[keyspace-groups]]
name = "test"

[[keyspace-groups.tasks]]
name = "read"
arrival-mode = "closed-loop"
concurrency = 2
task-cpu-us = 100
slice-cpu-us = 10
"#;

    fn global_override(assignment: &str) -> ConfigOverride {
        ConfigOverride::Global(assignment.to_owned())
    }

    fn keyspace_override(group: &str, assignment: &str) -> ConfigOverride {
        ConfigOverride::Keyspace {
            group: group.to_owned(),
            assignment: assignment.to_owned(),
        }
    }

    fn task_override(group: &str, task: &str, assignment: &str) -> ConfigOverride {
        ConfigOverride::Task {
            group: group.to_owned(),
            task: task.to_owned(),
            assignment: assignment.to_owned(),
        }
    }

    #[test]
    fn test_global_and_read_pool_overrides() {
        let overrides = [
            global_override("duration=2m"),
            global_override("warmup-duration=5s"),
            global_override("stats-print-interval=250ms"),
            global_override("random-seed=42"),
            global_override("drain-on-stop=false"),
            global_override("cooldown-duration=0s"),
            global_override("report-path=target/report=override.html"),
            global_override("summary-path=target/summary.json"),
            global_override("read-pool.workers=8"),
            global_override("read-pool.keyspace-fairness.enabled=true"),
            global_override("read-pool.keyspace-fairness.use-wall-time=false"),
            global_override("read-pool.keyspace-fairness.slice-precharge-us=500"),
            global_override("read-pool.fair-global-cpu-admission.enabled=true"),
            global_override("read-pool.fair-global-cpu-admission.max-cpu-cores=2.5"),
            global_override("read-pool.fair-global-cpu-admission.refill-interval=20ms"),
            global_override("read-pool.fair-global-cpu-admission.admit-interval-us=250"),
            global_override("read-pool.fair-global-cpu-admission.request-timeout=5s"),
        ];
        let config =
            SimulatorConfig::parse_with_overrides(MINIMAL_CLOSED_LOOP, &overrides).unwrap();

        assert_eq!(config.duration, ReadableDuration::minutes(2));
        assert_eq!(config.warmup_duration, ReadableDuration::secs(5));
        assert_eq!(config.stats_print_interval, ReadableDuration::millis(250));
        assert_eq!(config.random_seed, 42);
        assert!(!config.drain_on_stop);
        assert_eq!(config.cooldown_duration, ReadableDuration::ZERO);
        assert_eq!(
            config.report_path,
            PathBuf::from("target/report=override.html")
        );
        assert_eq!(config.summary_path, PathBuf::from("target/summary.json"));
        assert_eq!(config.read_pool.workers, 8);
        assert!(config.read_pool.keyspace_fairness.enabled);
        assert!(!config.read_pool.keyspace_fairness.use_wall_time);
        assert_eq!(config.read_pool.keyspace_fairness.slice_precharge_us, 500);
        let admission = &config.read_pool.fair_global_cpu_admission;
        assert!(admission.enabled);
        assert_eq!(admission.max_cpu_cores, 2.5);
        assert_eq!(admission.refill_interval, ReadableDuration::millis(20));
        assert_eq!(admission.admit_interval_us, 250);
        assert_eq!(admission.request_timeout, ReadableDuration::secs(5));
    }

    #[test]
    fn test_keyspace_and_task_overrides() {
        let overrides = [
            keyspace_override("test", "count=3"),
            task_override("test", "read", "arrival-mode=open-loop"),
            task_override("test", "read", "qps=12.5"),
            task_override("test", "read", "concurrency=4"),
            task_override("test", "read", "task-cpu-us=200"),
            task_override("test", "read", "task-cpu-jitter-us=10"),
            task_override("test", "read", "slice-cpu-us=20"),
            task_override("test", "read", "slice-cpu-jitter-us=2"),
            task_override("test", "read", "blocking-wall-us=30"),
            task_override("test", "read", "blocking-wall-jitter-us=3"),
            task_override("test", "read", "priority=high"),
            task_override("test", "read", "start-time=5s"),
            task_override("test", "read", "duration=30s"),
            task_override("test", "read", "slow-threshold=25ms"),
        ];
        let config =
            SimulatorConfig::parse_with_overrides(MINIMAL_CLOSED_LOOP, &overrides).unwrap();

        let group = &config.keyspace_groups[0];
        assert_eq!(group.count, 3);
        let task = &group.tasks[0];
        assert_eq!(task.arrival_mode, ArrivalMode::OpenLoop);
        assert_eq!(task.qps, Some(12.5));
        assert_eq!(task.concurrency, 4);
        assert_eq!(task.task_cpu_us, 200);
        assert_eq!(task.task_cpu_jitter_us, 10);
        assert_eq!(task.slice_cpu_us, 20);
        assert_eq!(task.slice_cpu_jitter_us, 2);
        assert_eq!(task.blocking_wall_us, 30);
        assert_eq!(task.blocking_wall_jitter_us, 3);
        assert_eq!(task.priority, Priority::High);
        assert_eq!(task.start_time, ReadableDuration::secs(5));
        assert_eq!(task.duration, ReadableDuration::secs(30));
        assert_eq!(task.slow_threshold, ReadableDuration::millis(25));
    }

    #[test]
    fn test_task_overrides_can_clear_qps_before_final_validation() {
        let source = MINIMAL_CLOSED_LOOP
            .replace("closed-loop", "open-loop")
            .replace("concurrency = 2", "qps = 10\nconcurrency = 2");
        let overrides = [
            task_override("test", "read", "arrival-mode=closed-loop"),
            task_override("test", "read", "qps=null"),
        ];
        let config = SimulatorConfig::parse_with_overrides(&source, &overrides).unwrap();

        let task = &config.keyspace_groups[0].tasks[0];
        assert_eq!(task.arrival_mode, ArrivalMode::ClosedLoop);
        assert_eq!(task.qps, None);
    }

    #[test]
    fn test_task_enabled_can_be_overridden() {
        let overrides = [task_override("B", "burst", "enabled=false")];
        let config = SimulatorConfig::parse_with_overrides(
            include_str!("../configs/five-tenants.toml"),
            &overrides,
        )
        .unwrap();

        assert!(!config.keyspace_groups[1].tasks[1].enabled);
    }

    #[test]
    fn test_at_least_one_task_must_be_enabled() {
        let source =
            MINIMAL_CLOSED_LOOP.replace("name = \"read\"", "name = \"read\"\nenabled = false");
        let error = SimulatorConfig::parse(&source).unwrap_err().to_string();

        assert!(
            error.contains("at least one enabled task class is required"),
            "{error}"
        );
    }

    #[test]
    fn test_disabled_tasks_are_still_validated() {
        let source = format!(
            r#"{MINIMAL_CLOSED_LOOP}
[[keyspace-groups.tasks]]
name = "disabled"
enabled = false
arrival-mode = "closed-loop"
concurrency = 0
task-cpu-us = 100
slice-cpu-us = 10
"#
        );
        let error = SimulatorConfig::parse(&source).unwrap_err().to_string();

        assert!(error.contains("concurrency"), "{error}");
    }

    #[test]
    fn test_last_override_for_the_same_field_wins() {
        let overrides = [
            global_override("duration=30s"),
            global_override("duration=45s"),
            task_override("test", "read", "concurrency=4"),
            task_override("test", "read", "concurrency=8"),
        ];
        let config =
            SimulatorConfig::parse_with_overrides(MINIMAL_CLOSED_LOOP, &overrides).unwrap();

        assert_eq!(config.duration, ReadableDuration::secs(45));
        assert_eq!(config.keyspace_groups[0].tasks[0].concurrency, 8);
    }

    #[test]
    fn test_override_selectors_accept_spaces_and_punctuation() {
        let source = MINIMAL_CLOSED_LOOP
            .replace("name = \"test\"", "name = \"noisy = tenant\"")
            .replace("name = \"read\"", "name = \"normal / read\"");
        let overrides = [
            keyspace_override("noisy = tenant", "count=2"),
            task_override("noisy = tenant", "normal / read", "concurrency=7"),
        ];
        let config = SimulatorConfig::parse_with_overrides(&source, &overrides).unwrap();

        assert_eq!(config.keyspace_groups[0].count, 2);
        assert_eq!(config.keyspace_groups[0].tasks[0].concurrency, 7);
    }

    #[test]
    fn test_malformed_and_unknown_overrides_are_rejected() {
        for (config_override, expected) in [
            (global_override("duration"), "expects FIELD=VALUE"),
            (global_override("=30s"), "non-empty field"),
            (global_override("workers=8"), "unknown --set field"),
            (
                keyspace_override("missing", "count=2"),
                "keyspace group \"missing\" not found",
            ),
            (
                task_override("test", "missing", "concurrency=4"),
                "task \"missing\" not found",
            ),
            (
                keyspace_override("test", "name=renamed"),
                "unknown --set-keyspace field",
            ),
            (
                task_override("test", "read", "name=renamed"),
                "unknown --set-task field",
            ),
            (
                global_override("read-pool.workers=many"),
                "invalid value \"many\"",
            ),
            (
                task_override("test", "read", "priority=urgent"),
                "expected high, normal, or low",
            ),
            (
                task_override("test", "read", "enabled=maybe"),
                "invalid value \"maybe\"",
            ),
        ] {
            let error = format!(
                "{:#}",
                SimulatorConfig::parse_with_overrides(MINIMAL_CLOSED_LOOP, &[config_override])
                    .unwrap_err()
            );
            assert!(error.contains(expected), "{error}");
        }
    }

    #[test]
    fn test_scoped_override_value_error_includes_selector() {
        let error = format!(
            "{:#}",
            SimulatorConfig::parse_with_overrides(
                MINIMAL_CLOSED_LOOP,
                &[task_override("test", "read", "concurrency=many")],
            )
            .unwrap_err()
        );

        assert!(error.contains("task \"test\"/\"read\""), "{error}");
        assert!(error.contains("invalid value \"many\""), "{error}");
    }

    #[test]
    fn test_overrides_still_validate_the_effective_config() {
        for config_override in [
            global_override("duration=0s"),
            global_override("read-pool.workers=0"),
            keyspace_override("test", "count=0"),
            task_override("test", "read", "concurrency=0"),
            task_override("test", "read", "qps=10"),
        ] {
            SimulatorConfig::parse_with_overrides(MINIMAL_CLOSED_LOOP, &[config_override])
                .unwrap_err();
        }
    }

    #[test]
    fn test_defaults() {
        let config = SimulatorConfig::parse(MINIMAL_CLOSED_LOOP).unwrap();
        assert_eq!(config.duration, ReadableDuration::minutes(1));
        assert_eq!(config.warmup_duration, ReadableDuration::ZERO);
        assert_eq!(config.stats_print_interval, ReadableDuration::secs(1));
        assert_eq!(config.random_seed, 1);
        assert!(config.drain_on_stop);
        assert_eq!(config.cooldown_duration, ReadableDuration::secs(5));
        assert_eq!(
            config.report_path,
            PathBuf::from("target/yatp-sim/report.html")
        );
        assert_eq!(
            config.summary_path,
            PathBuf::from("target/yatp-sim/summary.json")
        );
        assert_eq!(config.read_pool.workers, 4);
        assert!(!config.read_pool.keyspace_fairness.enabled);
        assert!(config.read_pool.keyspace_fairness.use_wall_time);
        let admission = &config.read_pool.fair_global_cpu_admission;
        assert!(!admission.enabled);
        assert_eq!(admission.max_cpu_cores, 4.0);
        assert_eq!(admission.refill_interval, ReadableDuration::millis(100));
        assert_eq!(admission.admit_interval_us, 1_000);
        assert_eq!(admission.request_timeout, ReadableDuration::hours(1));
        assert_eq!(admission.global_cpu_quota_us().unwrap(), 400_000);

        let task = &config.keyspace_groups[0].tasks[0];
        assert!(task.enabled);
        assert_eq!(task.task_cpu_jitter_us, 0);
        assert_eq!(task.slice_cpu_jitter_us, 0);
        assert_eq!(task.blocking_wall_us, 0);
        assert_eq!(task.blocking_wall_jitter_us, 0);
        assert_eq!(task.priority, Priority::Normal);
        assert_eq!(task.start_time, ReadableDuration::ZERO);
        assert_eq!(task.duration, ReadableDuration::ZERO);
        assert_eq!(task.slow_threshold, ReadableDuration::ZERO);
    }

    #[test]
    fn test_task_slow_threshold_can_be_configured() {
        let source = MINIMAL_CLOSED_LOOP.replace(
            "slice-cpu-us = 10",
            "slice-cpu-us = 10\nslow-threshold = \"25ms\"",
        );
        let config = SimulatorConfig::parse(&source).unwrap();

        assert_eq!(
            config.keyspace_groups[0].tasks[0].slow_threshold,
            ReadableDuration::millis(25)
        );
    }

    #[test]
    fn test_open_loop_requires_qps() {
        let source = MINIMAL_CLOSED_LOOP.replace("closed-loop", "open-loop");
        let error = SimulatorConfig::parse(&source).unwrap_err().to_string();
        assert!(
            error.contains("qps is required in open-loop mode"),
            "{error}"
        );
    }

    #[test]
    fn test_closed_loop_rejects_qps() {
        let source = MINIMAL_CLOSED_LOOP.replace("concurrency = 2", "qps = 10\nconcurrency = 2");
        let error = SimulatorConfig::parse(&source).unwrap_err().to_string();
        assert!(
            error.contains("qps is not valid in closed-loop mode"),
            "{error}"
        );
    }

    #[test]
    fn test_unknown_fields_are_rejected() {
        let source = format!("unknown-field = true\n{MINIMAL_CLOSED_LOOP}");
        let error = format!("{:#}", SimulatorConfig::parse(&source).unwrap_err());
        assert!(error.contains("unknown field"), "{error}");
    }

    #[test]
    fn test_cooldown_duration_can_be_disabled() {
        let source = format!("cooldown-duration = \"0s\"\n{MINIMAL_CLOSED_LOOP}");
        let config = SimulatorConfig::parse(&source).unwrap();
        assert_eq!(config.cooldown_duration, ReadableDuration::ZERO);
    }

    #[test]
    fn test_fair_global_cpu_admission_requires_keyspace_fairness() {
        let source = format!(
            "[read-pool.fair-global-cpu-admission]\nenabled = true\n\n{MINIMAL_CLOSED_LOOP}"
        );
        let error = SimulatorConfig::parse(&source).unwrap_err().to_string();
        assert!(error.contains("requires keyspace fairness"), "{error}");
    }

    #[test]
    fn test_fair_global_cpu_admission_computes_global_quota() {
        let source = format!(
            "[read-pool.keyspace-fairness]\nenabled = true\n\n[read-pool.fair-global-cpu-admission]\nenabled = true\nmax-cpu-cores = 2.5\nrefill-interval = \"20ms\"\nadmit-interval-us = 250\nrequest-timeout = \"5s\"\n\n{MINIMAL_CLOSED_LOOP}"
        );
        let config = SimulatorConfig::parse(&source).unwrap();
        let admission = &config.read_pool.fair_global_cpu_admission;
        assert!(admission.enabled);
        assert_eq!(admission.global_cpu_quota_us().unwrap(), 50_000);
        assert_eq!(admission.admit_interval_us, 250);
        assert_eq!(admission.request_timeout, ReadableDuration::secs(5));
    }

    #[test]
    fn test_fair_global_cpu_admission_validates_limits_and_intervals() {
        for field in [
            "max-cpu-cores = 0",
            "refill-interval = \"0us\"",
            "admit-interval-us = 0",
            "request-timeout = \"0s\"",
        ] {
            let source =
                format!("[read-pool.fair-global-cpu-admission]\n{field}\n\n{MINIMAL_CLOSED_LOOP}");
            assert!(SimulatorConfig::parse(&source).is_err(), "accepted {field}");
        }
    }

    #[test]
    fn test_report_and_summary_paths_must_differ() {
        let source =
            format!("report-path = \"result\"\nsummary-path = \"result\"\n{MINIMAL_CLOSED_LOOP}");
        let error = SimulatorConfig::parse(&source).unwrap_err().to_string();
        assert!(
            error.contains("report-path and summary-path must be different"),
            "{error}"
        );
    }

    #[test]
    fn test_task_duration_is_clamped_to_global_duration() {
        let source = format!(
            "duration = \"10s\"\n{}",
            MINIMAL_CLOSED_LOOP.replace(
                "slice-cpu-us = 10",
                "slice-cpu-us = 10\nstart-time = \"8s\"\nduration = \"5s\"",
            )
        );
        let config = SimulatorConfig::parse(&source).unwrap();
        let task = &config.keyspace_groups[0].tasks[0];
        assert_eq!(config.task_end_time(task).unwrap(), Duration::from_secs(10));
    }

    #[test]
    fn test_example_config_supports_fairness_comparison_overrides() {
        let source = include_str!("../configs/five-tenants.toml");
        let fairness_on = SimulatorConfig::parse(source).unwrap();
        let fairness_off = SimulatorConfig::parse_with_overrides(
            source,
            &[
                global_override("read-pool.keyspace-fairness.enabled=false"),
                global_override("read-pool.fair-global-cpu-admission.enabled=false"),
                global_override("report-path=target/yatp-sim/five-tenants-fairness-off.html"),
                global_override("summary-path=target/yatp-sim/five-tenants-fairness-off.json"),
            ],
        )
        .unwrap();

        assert!(fairness_on.read_pool.keyspace_fairness.enabled);
        assert!(!fairness_off.read_pool.keyspace_fairness.enabled);
        assert_ne!(fairness_on.report_path, fairness_off.report_path);
        assert_ne!(fairness_on.summary_path, fairness_off.summary_path);
    }

    #[test]
    fn test_config_path_falls_back_to_yatp_sim_root() {
        let path = resolve_config_path(Path::new("configs/five-tenants.toml"));
        let expected = Path::new(YATP_SIM_ROOT).join("configs/five-tenants.toml");
        assert_eq!(
            fs::canonicalize(path).unwrap(),
            fs::canonicalize(expected).unwrap()
        );

        let missing = Path::new("configs/config-that-does-not-exist.toml");
        assert_eq!(
            resolve_config_path(missing),
            Path::new(YATP_SIM_ROOT).join(missing)
        );
    }

    #[test]
    fn test_absolute_config_path_is_unchanged() {
        let path = Path::new(YATP_SIM_ROOT).join("configs/five-tenants.toml");
        assert_eq!(resolve_config_path(&path), path);
    }
}