use std::collections::HashMap;
use std::time::Duration;

use crate::allocator::AllocatorConfig;

/// Strict-compatible default for legacy proxy stream admission limit.
pub const DEFAULT_MAX_CONCURRENT_TSO_PROXY_STREAMINGS: i32 = 5_000;
/// Strict-compatible default for legacy proxy receive timeout.
pub const DEFAULT_TSO_PROXY_RECV_FROM_CLIENT_TIMEOUT: Duration = Duration::from_secs(60 * 60);
/// Strict-compatible default for legacy proxy stream idle timeout.
pub const DEFAULT_TSO_PROXY_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60);

/// Runtime-applied configuration that can be updated via hot-reload admin flow.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RuntimeConfigState {
    /// Allocator-level lease/oracle runtime knobs.
    pub allocator: AllocatorConfig,
    /// Maximum accepted `count` for one `Tso` stream request (`0` keeps strict unlimited parity).
    pub max_tso_count_per_request: u32,
    /// Per-caller `GetMinTS` QPS limit (`0` disables the guardrail).
    pub get_min_ts_rate_limit_qps_per_caller: u32,
    /// Global `GetMinTS` QPS limit across callers (`0` disables the guardrail).
    pub get_min_ts_rate_limit_qps_global: u32,
    /// Extension-only `GetMinTS` authorization guardrail (`0` keeps strict-compat disabled).
    pub get_min_ts_authz_allow_sender_id: u64,
    /// Enables extension-mode async prefetch for allocator windows.
    pub enable_async_tso_window_prefetch: bool,
    /// Enables extension-mode stream overload hardening.
    pub enable_tso_overload_hardening: bool,
    /// Maximum in-flight requests per group when overload hardening is enabled.
    pub tso_overload_max_in_flight_per_group: u32,
    /// Legacy PD-proxy concurrent stream limit (`<0` unlimited, `0` disabled, `>0` bounded).
    pub max_concurrent_tso_proxy_streamings: i32,
    /// Legacy PD-proxy receive timeout while waiting for one client TSO request.
    pub tso_proxy_recv_from_client_timeout: Duration,
    /// Legacy PD-proxy idle timeout for one forwarding dispatcher stream.
    pub tso_proxy_stream_idle_timeout: Duration,
}

impl Default for RuntimeConfigState {
    fn default() -> Self {
        Self {
            allocator: AllocatorConfig::default(),
            max_tso_count_per_request: 0,
            get_min_ts_rate_limit_qps_per_caller: 0,
            get_min_ts_rate_limit_qps_global: 0,
            get_min_ts_authz_allow_sender_id: 0,
            enable_async_tso_window_prefetch: false,
            enable_tso_overload_hardening: false,
            tso_overload_max_in_flight_per_group: 64,
            max_concurrent_tso_proxy_streamings: DEFAULT_MAX_CONCURRENT_TSO_PROXY_STREAMINGS,
            tso_proxy_recv_from_client_timeout: DEFAULT_TSO_PROXY_RECV_FROM_CLIENT_TIMEOUT,
            tso_proxy_stream_idle_timeout: DEFAULT_TSO_PROXY_STREAM_IDLE_TIMEOUT,
        }
    }
}

impl RuntimeConfigState {
    /// Builds runtime config from normalized admin config map and current fallback.
    ///
    /// Missing keys preserve fallback values so upgrades remain backward-compatible.
    #[must_use]
    pub fn from_map(
        normalized: &HashMap<String, String>,
        fallback: RuntimeConfigState,
        allocator_lease_ms: i64,
    ) -> Self {
        let mut next = fallback;

        next.allocator.lease_ms = allocator_lease_ms.max(1);
        if let Some(value) = parse_i64(normalized, "tso-save-interval-ms") {
            next.allocator.oracle.save_interval_ms = value.max(1);
        }
        if let Some(value) = parse_i64(normalized, "tso-update-physical-interval-ms") {
            next.allocator.oracle.update_physical_interval_ms = value.clamp(1, 10_000);
        }
        if let Some(value) =
            parse_i64_first(normalized, &["max-gap-reset-ts", "max-gap-reset-ts-ms"])
        {
            next.allocator.oracle.max_reset_ts_gap_ms = value.max(1);
        }
        if let Some(value) = parse_u32(normalized, "max-tso-count-per-request") {
            next.max_tso_count_per_request = value;
        }
        if let Some(value) = parse_u32(normalized, "get-min-ts-rate-limit-qps-per-caller") {
            next.get_min_ts_rate_limit_qps_per_caller = value;
        }
        if let Some(value) = parse_u32(normalized, "get-min-ts-rate-limit-qps-global") {
            next.get_min_ts_rate_limit_qps_global = value;
        }
        if let Some(value) = parse_u64(normalized, "get-min-ts-authz-allow-sender-id") {
            next.get_min_ts_authz_allow_sender_id = value;
        }
        if let Some(value) = parse_bool(normalized, "enable-async-tso-window-prefetch") {
            next.enable_async_tso_window_prefetch = value;
        }
        if let Some(value) = parse_bool(normalized, "enable-tso-overload-hardening") {
            next.enable_tso_overload_hardening = value;
        }
        if let Some(value) = parse_non_zero_u32(normalized, "tso-overload-max-in-flight-per-group")
        {
            next.tso_overload_max_in_flight_per_group = value;
        }
        if let Some(value) = parse_i32(normalized, "max-concurrent-tso-proxy-streamings") {
            next.max_concurrent_tso_proxy_streamings = value;
        }
        if let Some(value_ms) = parse_i64(normalized, "tso-proxy-recv-from-client-timeout") {
            let normalized_ms = if value_ms <= 0 {
                DEFAULT_TSO_PROXY_RECV_FROM_CLIENT_TIMEOUT.as_millis() as i64
            } else {
                value_ms
            };
            next.tso_proxy_recv_from_client_timeout = Duration::from_millis(normalized_ms as u64);
        }
        if let Some(value_ms) = parse_i64(normalized, "tso-proxy-stream-idle-timeout-ms") {
            let normalized_ms = if value_ms <= 0 {
                DEFAULT_TSO_PROXY_STREAM_IDLE_TIMEOUT.as_millis() as i64
            } else {
                value_ms
            };
            next.tso_proxy_stream_idle_timeout = Duration::from_millis(normalized_ms as u64);
        }

        next
    }
}

fn parse_i64(values: &HashMap<String, String>, key: &str) -> Option<i64> {
    values.get(key).and_then(|raw| raw.parse::<i64>().ok())
}

fn parse_i64_first(values: &HashMap<String, String>, keys: &[&str]) -> Option<i64> {
    keys.iter().find_map(|key| parse_i64(values, key))
}

fn parse_u32(values: &HashMap<String, String>, key: &str) -> Option<u32> {
    values.get(key).and_then(|raw| raw.parse::<u32>().ok())
}

fn parse_u64(values: &HashMap<String, String>, key: &str) -> Option<u64> {
    values.get(key).and_then(|raw| raw.parse::<u64>().ok())
}

fn parse_i32(values: &HashMap<String, String>, key: &str) -> Option<i32> {
    values.get(key).and_then(|raw| raw.parse::<i32>().ok())
}

fn parse_non_zero_u32(values: &HashMap<String, String>, key: &str) -> Option<u32> {
    parse_u32(values, key).filter(|value| *value > 0)
}

fn parse_bool(values: &HashMap<String, String>, key: &str) -> Option<bool> {
    values
        .get(key)
        .and_then(|raw| match raw.trim().to_ascii_lowercase().as_str() {
            "1" | "true" | "yes" | "on" => Some(true),
            "0" | "false" | "no" | "off" => Some(false),
            _ => None,
        })
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::time::Duration;

    use super::RuntimeConfigState;

    #[test]
    fn from_map_applies_known_hot_reload_keys() {
        let mut map = HashMap::new();
        map.insert("tso-save-interval-ms".to_string(), "8000".to_string());
        map.insert(
            "tso-update-physical-interval-ms".to_string(),
            "100".to_string(),
        );
        map.insert("max-tso-count-per-request".to_string(), "123".to_string());
        map.insert(
            "get-min-ts-rate-limit-qps-global".to_string(),
            "456".to_string(),
        );
        map.insert(
            "enable-tso-overload-hardening".to_string(),
            "true".to_string(),
        );
        map.insert(
            "get-min-ts-authz-allow-sender-id".to_string(),
            "789".to_string(),
        );
        map.insert(
            "tso-overload-max-in-flight-per-group".to_string(),
            "17".to_string(),
        );
        map.insert(
            "max-concurrent-tso-proxy-streamings".to_string(),
            "-1".to_string(),
        );
        map.insert(
            "tso-proxy-recv-from-client-timeout".to_string(),
            "2500".to_string(),
        );
        map.insert(
            "tso-proxy-stream-idle-timeout-ms".to_string(),
            "3500".to_string(),
        );
        map.insert("max-gap-reset-ts".to_string(), "9000".to_string());

        let state = RuntimeConfigState::from_map(&map, RuntimeConfigState::default(), 9_000);
        assert_eq!(state.allocator.lease_ms, 9_000);
        assert_eq!(state.allocator.oracle.save_interval_ms, 8_000);
        assert_eq!(state.allocator.oracle.update_physical_interval_ms, 100);
        assert_eq!(state.allocator.oracle.max_reset_ts_gap_ms, 9_000);
        assert_eq!(state.max_tso_count_per_request, 123);
        assert_eq!(state.get_min_ts_rate_limit_qps_global, 456);
        assert_eq!(state.get_min_ts_authz_allow_sender_id, 789);
        assert!(state.enable_tso_overload_hardening);
        assert_eq!(state.tso_overload_max_in_flight_per_group, 17);
        assert_eq!(state.max_concurrent_tso_proxy_streamings, -1);
        assert_eq!(
            state.tso_proxy_recv_from_client_timeout,
            Duration::from_millis(2_500)
        );
        assert_eq!(
            state.tso_proxy_stream_idle_timeout,
            Duration::from_millis(3_500)
        );
    }

    #[test]
    fn from_map_normalizes_non_positive_proxy_recv_timeout_to_default() {
        let mut map = HashMap::new();
        map.insert(
            "tso-proxy-recv-from-client-timeout".to_string(),
            "0".to_string(),
        );

        let state = RuntimeConfigState::from_map(&map, RuntimeConfigState::default(), 9_000);
        assert_eq!(
            state.tso_proxy_recv_from_client_timeout,
            super::DEFAULT_TSO_PROXY_RECV_FROM_CLIENT_TIMEOUT
        );
    }

    #[test]
    fn from_map_normalizes_non_positive_proxy_idle_timeout_to_default() {
        let mut map = HashMap::new();
        map.insert(
            "tso-proxy-stream-idle-timeout-ms".to_string(),
            "0".to_string(),
        );

        let state = RuntimeConfigState::from_map(&map, RuntimeConfigState::default(), 9_000);
        assert_eq!(
            state.tso_proxy_stream_idle_timeout,
            super::DEFAULT_TSO_PROXY_STREAM_IDLE_TIMEOUT
        );
    }

    #[test]
    fn from_map_accepts_legacy_max_gap_reset_ts_ms_alias() {
        let mut map = HashMap::new();
        map.insert("max-gap-reset-ts-ms".to_string(), "7777".to_string());

        let state = RuntimeConfigState::from_map(&map, RuntimeConfigState::default(), 9_000);
        assert_eq!(state.allocator.oracle.max_reset_ts_gap_ms, 7_777);
    }

    #[test]
    fn from_map_prefers_canonical_max_gap_reset_ts_key_when_both_present() {
        let mut map = HashMap::new();
        map.insert("max-gap-reset-ts".to_string(), "8888".to_string());
        map.insert("max-gap-reset-ts-ms".to_string(), "9999".to_string());

        let state = RuntimeConfigState::from_map(&map, RuntimeConfigState::default(), 9_000);
        assert_eq!(state.allocator.oracle.max_reset_ts_gap_ms, 8_888);
    }
}