use serde::{Deserialize, Serialize};

pub const DEFAULT_NAME_SERVER: &str = "127.0.0.1:9876";
pub const DEFAULT_CONSUMER_GROUP: &str = "kenevent";
pub const DEFAULT_TOPIC_PREFIXES: [&str; 2] = ["gitdata", "kenshell"];
pub const DEFAULT_CONSUME_TIMEOUT_MS: u64 = 15 * 60 * 1_000;
pub const DEFAULT_MAX_RECONSUME_TIMES: u32 = 16;
pub const DEFAULT_PULL_BATCH_SIZE: u32 = 32;
pub const DEFAULT_PULL_BATCH_SIZE_IN_BYTES: u32 = 256 * 1024;
pub const DEFAULT_PULL_INTERVAL_MS: u64 = 0;
pub const MAX_TOPIC_LEN: usize = 127;
pub const MAX_AREA_LEN: usize = 64;

fn default_name_server() -> String {
    DEFAULT_NAME_SERVER.to_string()
}

fn default_consumer_group() -> String {
    DEFAULT_CONSUMER_GROUP.to_string()
}

fn default_topic_prefixes() -> Vec<String> {
    DEFAULT_TOPIC_PREFIXES
        .iter()
        .map(|prefix| (*prefix).to_string())
        .collect()
}

fn default_consume_timeout_ms() -> u64 {
    DEFAULT_CONSUME_TIMEOUT_MS
}

fn default_max_reconsume_times() -> u32 {
    DEFAULT_MAX_RECONSUME_TIMES
}

fn default_pull_batch_size() -> u32 {
    DEFAULT_PULL_BATCH_SIZE
}

fn default_pull_batch_size_in_bytes() -> u32 {
    DEFAULT_PULL_BATCH_SIZE_IN_BYTES
}

fn default_pull_interval_ms() -> u64 {
    DEFAULT_PULL_INTERVAL_MS
}

#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct RocketMqConsumerConfig {
    #[serde(default = "default_name_server")]
    pub name_server: String,
    #[serde(default = "default_consumer_group")]
    pub consumer_group: String,
    #[serde(default = "default_topic_prefixes")]
    pub topic_prefixes: Vec<String>,
    #[serde(default = "default_consume_timeout_ms")]
    pub consume_timeout_ms: u64,
    #[serde(default = "default_max_reconsume_times")]
    pub max_reconsume_times: u32,
    #[serde(default = "default_pull_batch_size")]
    pub pull_batch_size: u32,
    #[serde(default = "default_pull_batch_size_in_bytes")]
    pub pull_batch_size_in_bytes: u32,
    #[serde(default = "default_pull_interval_ms")]
    pub pull_interval_ms: u64,
}

impl Default for RocketMqConsumerConfig {
    fn default() -> Self {
        Self {
            name_server: default_name_server(),
            consumer_group: default_consumer_group(),
            topic_prefixes: default_topic_prefixes(),
            consume_timeout_ms: default_consume_timeout_ms(),
            max_reconsume_times: default_max_reconsume_times(),
            pull_batch_size: default_pull_batch_size(),
            pull_batch_size_in_bytes: default_pull_batch_size_in_bytes(),
            pull_interval_ms: default_pull_interval_ms(),
        }
    }
}

impl RocketMqConsumerConfig {
    pub fn name_servers(&self) -> Vec<&str> {
        self.name_server
            .split([';', ','])
            .map(str::trim)
            .filter(|server| !server.is_empty())
            .collect()
    }

    pub fn consume_timeout_minutes(&self) -> u64 {
        self.consume_timeout_ms.div_ceil(60_000).max(1)
    }

    pub fn validate(&self) -> Result<(), String> {
        let servers = self.name_servers();
        if servers.is_empty() {
            return Err("rocketmq.name_server must not be blank".to_string());
        }
        for server in servers {
            let (host, port) = server.rsplit_once(':').ok_or_else(|| {
                format!("rocketmq.name_server `{server}` must use the `host:port` form")
            })?;
            if host.trim().is_empty() {
                return Err(format!("rocketmq.name_server `{server}` has an empty host"));
            }
            let port = port
                .parse::<u16>()
                .map_err(|_| format!("rocketmq.name_server `{server}` has an invalid port"))?;
            if port == 0 {
                return Err(format!("rocketmq.name_server `{server}` has a zero port"));
            }
        }

        if !is_valid_identifier(&self.consumer_group) {
            return Err("rocketmq.consumer_group must use letters, digits, `_` or `-`".to_string());
        }

        if self.topic_prefixes.is_empty() {
            return Err("rocketmq.topic_prefixes must not be empty".to_string());
        }
        for (index, prefix) in self.topic_prefixes.iter().enumerate() {
            if !DEFAULT_TOPIC_PREFIXES.contains(&prefix.as_str()) {
                return Err(format!(
                    "unsupported rocketmq.topic_prefixes[{index}]: {prefix}"
                ));
            }
            if !is_valid_identifier(prefix) {
                return Err(format!(
                    "rocketmq.topic_prefixes[{index}] must use letters, digits, `_` or `-`"
                ));
            }
            if prefix.len() + 1 + MAX_AREA_LEN > MAX_TOPIC_LEN {
                return Err(format!(
                    "rocketmq.topic_prefixes[{index}] leaves no room for a topic under {MAX_TOPIC_LEN} characters"
                ));
            }
            if self.topic_prefixes[..index].contains(prefix) {
                return Err(format!(
                    "rocketmq.topic_prefixes[{index}] `{prefix}` is duplicated"
                ));
            }
        }

        if self.consume_timeout_ms == 0 {
            return Err("rocketmq.consume_timeout_ms must be greater than zero".to_string());
        }
        if self.max_reconsume_times == 0 {
            return Err("rocketmq.max_reconsume_times must be greater than zero".to_string());
        }
        if self.max_reconsume_times > i32::MAX as u32 {
            return Err("rocketmq.max_reconsume_times exceeds i32::MAX".to_string());
        }
        if self.pull_batch_size == 0 {
            return Err("rocketmq.pull_batch_size must be greater than zero".to_string());
        }
        if self.pull_batch_size_in_bytes == 0 {
            return Err("rocketmq.pull_batch_size_in_bytes must be greater than zero".to_string());
        }

        Ok(())
    }
}

fn is_valid_identifier(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_TOPIC_LEN
        && value
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-'))
}