use std::{cmp::Ordering, fmt::Display};

use anyhow::{Context, Result, ensure};
use lazy_static::lazy_static;
use regex::Regex;
use serde::{Deserialize, Deserializer, Serialize, Serializer};

lazy_static! {
    pub static ref CDC_FILENAME_MATCH: Regex = Regex::new(r"CDC\d{20}\.json$").unwrap();

    // This matches: prefix/table_id/table_version/CDCsequence.json
    // - (.+?/\\d+): (greedy, ends with table_id)
    //   - group 1: prefix (including table_id)
    // - (\d+): (one or more digits)
    //   - group 2: table_version
    // - CDC(\d{20}): (20 digits)
    //   - group 3: sequence
    pub static ref CDC_FILENAME_CAPTURE: Regex = Regex::new(
        r"^(.+?/\d+)/(\d+)/CDC(\d{20})\.json$"
    ).unwrap();
}

#[derive(Debug, PartialEq, Eq, Clone, Default)]
pub struct CDCFileName {
    prefix: String,
    table_version: u64,
    sequence: String,
}

impl CDCFileName {
    pub fn empty() -> Self {
        Self::default()
    }

    pub fn new(prefix: String, table_version: u64, sequence: String) -> Result<Self> {
        if prefix.is_empty() {
            ensure!(
                table_version == 0 && sequence.is_empty(),
                "Invalid empty CDC file name components"
            );
            return Ok(Self::empty());
        }

        let s3_key = format!("{prefix}/{table_version}/CDC{sequence}.json");
        Self::parse(&s3_key)
    }

    pub fn parse(s3_key: &str) -> Result<Self> {
        // During initialization, the last_cdc_file is empty.
        if s3_key.is_empty() {
            return Ok(CDCFileName::empty());
        }
        let caps = CDC_FILENAME_CAPTURE
            .captures(s3_key)
            .ok_or_else(|| anyhow::anyhow!("Invalid CDC file name: {}", s3_key))?;

        let prefix = caps
            .get(1)
            .ok_or_else(|| anyhow::anyhow!("Missing prefix in CDC file name: {}", s3_key))?
            .as_str()
            .to_string();
        let table_version = caps
            .get(2)
            .ok_or_else(|| anyhow::anyhow!("Missing table version in CDC file name: {}", s3_key))?
            .as_str()
            .parse::<u64>()
            .with_context(|| format!("Invalid table version in CDC file name: {}", s3_key))?;
        let sequence = caps
            .get(3)
            .ok_or_else(|| anyhow::anyhow!("Missing sequence in CDC file name: {}", s3_key))?
            .as_str()
            .to_string();

        Ok(CDCFileName {
            prefix,
            table_version,
            sequence,
        })
    }

    pub fn is_empty(&self) -> bool {
        self.prefix.is_empty()
    }

    pub fn prefix(&self) -> &str {
        &self.prefix
    }

    pub fn table_version(&self) -> u64 {
        self.table_version
    }

    pub fn sequence(&self) -> &str {
        &self.sequence
    }

    pub fn same_directory(&self, other: &Self) -> bool {
        self.prefix == other.prefix && self.table_version == other.table_version
    }

    pub fn sequence_number(&self) -> Result<u64> {
        ensure!(!self.is_empty(), "CDC file name is empty");
        self.sequence
            .parse::<u64>()
            .with_context(|| format!("Invalid sequence in CDC file name: {}", self))
    }

    pub fn data_dir_path(&self) -> Result<String> {
        ensure!(!self.is_empty(), "CDC file name is empty");
        Ok(format!("{}/{}", self.prefix, self.table_version))
    }

    pub fn index_file_path(&self) -> Result<String> {
        Ok(format!("{}/meta/CDC.index", self.data_dir_path()?))
    }

    pub fn file_name(&self) -> Result<String> {
        ensure!(!self.is_empty(), "CDC file name is empty");
        Ok(format!("CDC{}.json", self.sequence))
    }
}

impl Serialize for CDCFileName {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&self.to_string())
    }
}

impl<'de> Deserialize<'de> for CDCFileName {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let s3_key = String::deserialize(deserializer)?;
        CDCFileName::parse(&s3_key).map_err(serde::de::Error::custom)
    }
}

impl Display for CDCFileName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        if self.prefix.is_empty() {
            write!(f, "")
        } else {
            write!(
                f,
                "{}/{}/CDC{}.json",
                self.prefix, self.table_version, self.sequence
            )
        }
    }
}

impl PartialOrd for CDCFileName {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for CDCFileName {
    fn cmp(&self, other: &Self) -> Ordering {
        self.table_version
            .cmp(&other.table_version) // compare table_version first
            .then(self.sequence.cmp(&other.sequence)) // finally compare sequence
    }
}

pub fn older_cdc_file(lhs: &str, rhs: &str) -> String {
    let lhs_file = CDCFileName::parse(lhs);
    let rhs_file = CDCFileName::parse(rhs);
    match (lhs_file, rhs_file) {
        (Ok(lhs_file), Ok(rhs_file)) => {
            if lhs_file < rhs_file {
                lhs.to_string()
            } else {
                rhs.to_string()
            }
        }
        (Ok(_), Err(_)) => lhs.to_string(),
        (Err(_), Ok(_)) => rhs.to_string(),
        (Err(_), Err(_)) => String::new(),
    }
}

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

    #[test]
    fn test_cdc_file_name_parse_valid() {
        let s3_key =
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json";
        let cdc_file = CDCFileName::parse(s3_key).unwrap();
        assert_eq!(cdc_file.prefix(), "tici_default_prefix/cdc/10313");
        assert_eq!(cdc_file.table_version(), 461869219056451591);
        assert_eq!(cdc_file.sequence(), "00000000000000000001");
        assert_eq!(cdc_file.to_string(), s3_key);
    }

    #[test]
    fn test_cdc_file_name_parse_empty() {
        let s3_key = "";
        let cdc_file = CDCFileName::parse(s3_key).unwrap();
        assert_eq!(cdc_file.prefix(), "");
        assert_eq!(cdc_file.table_version(), 0);
        assert_eq!(cdc_file.sequence(), "");
    }

    #[test]
    fn test_cdc_file_name_parse_rejects_date() {
        let s3_key = "tici_default_prefix/cdc/10313/461869219056451591/2025-10-31/CDC00000000000000000001.json";
        let result = CDCFileName::parse(s3_key);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(err.contains("Invalid CDC file name"));
        assert!(err.contains(s3_key));
    }

    #[test]
    fn test_cdc_file_name_parse_wrong_sequence_length() {
        let s3_key = "tici_default_prefix/cdc/10313/461869219056451591/CDC000000000000000001.json";
        let result = CDCFileName::parse(s3_key);
        assert!(result.is_err());
    }

    #[test]
    fn test_cdc_file_name_comparison_sequence() {
        let a_str = "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json";
        let b_str = "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000002.json";
        let a = CDCFileName::parse(a_str).unwrap();
        let b = CDCFileName::parse(b_str).unwrap();
        assert!(a < b);
        assert!(b > a);
    }

    #[test]
    fn test_cdc_file_name_helpers() {
        let file = CDCFileName::parse(
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json",
        )
        .unwrap();
        let same_dir = CDCFileName::parse(
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000002.json",
        )
        .unwrap();
        let different_dir = CDCFileName::parse(
            "tici_default_prefix/cdc/10313/461869219056451592/CDC00000000000000000001.json",
        )
        .unwrap();

        assert!(!file.is_empty());
        assert!(file.same_directory(&same_dir));
        assert!(!file.same_directory(&different_dir));
        assert_eq!(file.sequence_number().unwrap(), 1);
        assert_eq!(
            file.data_dir_path().unwrap(),
            "tici_default_prefix/cdc/10313/461869219056451591"
        );
        assert_eq!(
            file.index_file_path().unwrap(),
            "tici_default_prefix/cdc/10313/461869219056451591/meta/CDC.index"
        );
        assert_eq!(file.file_name().unwrap(), "CDC00000000000000000001.json");
    }

    #[test]
    fn test_cdc_file_name_comparison_table_version() {
        let a_str = "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000002.json";
        let b_str = "tici_default_prefix/cdc/10313/461869219056451592/CDC00000000000000000001.json";
        let a = CDCFileName::parse(a_str).unwrap();
        let b = CDCFileName::parse(b_str).unwrap();
        assert!(a < b);
        assert!(b > a);
    }

    #[test]
    fn test_cdc_file_name_comparison_equal() {
        let a_str = "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json";
        let b_str = "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json";
        let a = CDCFileName::parse(a_str).unwrap();
        let b = CDCFileName::parse(b_str).unwrap();
        assert_eq!(a, b);
        assert_eq!(a.cmp(&b), std::cmp::Ordering::Equal);
    }

    #[test]
    fn test_cdc_file_name_comparison_empty_file_name() {
        let a_str = "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json";
        let b_str = "";
        let a = CDCFileName::parse(a_str).unwrap();
        let b = CDCFileName::parse(b_str).unwrap();
        assert!(a > b);
        assert!(b < a);
    }

    #[test]
    fn test_cdc_file_name_comparison_both_empty() {
        let a_str = "";
        let b_str = "";
        let a = CDCFileName::parse(a_str).unwrap();
        let b = CDCFileName::parse(b_str).unwrap();
        assert_eq!(a, b);
        assert_eq!(a.cmp(&b), std::cmp::Ordering::Equal);
    }

    #[test]
    fn test_cdc_file_name_missing_fields_error() {
        // Missing prefix
        let s3_key = "461869219056451591/CDC00000000000000000001.json";
        let result = CDCFileName::parse(s3_key);
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Invalid CDC file name")
        );

        // Missing table version
        let s3_key = "tici_default_prefix/cdc/10313/CDC00000000000000000001.json";
        let result = CDCFileName::parse(s3_key);
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Invalid CDC file name")
        );

        // Missing sequence
        let s3_key = "tici_default_prefix/cdc/10313/461869219056451591/";
        let result = CDCFileName::parse(s3_key);
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Invalid CDC file name")
        );
    }

    #[test]
    fn test_cdc_file_name_new_round_trip() {
        let cdc_file = CDCFileName::new(
            "tici_default_prefix/cdc/10313".to_string(),
            461869219056451591,
            "00000000000000000001".to_string(),
        )
        .unwrap();

        assert_eq!(
            cdc_file.to_string(),
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json"
        );
    }

    #[test]
    fn test_cdc_file_name_serde_as_string() {
        let cdc_file = CDCFileName::parse(
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json",
        )
        .unwrap();

        let serialized = serde_json::to_string(&cdc_file).unwrap();
        assert_eq!(
            serialized,
            "\"tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json\""
        );

        let deserialized: CDCFileName = serde_json::from_str(&serialized).unwrap();
        assert_eq!(deserialized, cdc_file);
    }

    #[test]
    fn test_empty_cdc_file_name_serde_as_empty_string() {
        let serialized = serde_json::to_string(&CDCFileName::empty()).unwrap();
        assert_eq!(serialized, "\"\"");

        let deserialized: CDCFileName = serde_json::from_str(&serialized).unwrap();
        assert!(deserialized.is_empty());
    }
}