use std::sync::Arc;

use tici_common::config::S3Config;

use super::BackendParseError;
use crate::storage::{self, BackendStorage};

// Some cloud vendors provide S3 compatible APIs, but they are not fully
// compatible with AWS S3. And tidb-server needs this information to determine
// whether to use the compatible API or not.
#[allow(clippy::upper_case_acronyms)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CloudVendor {
    Unknown,
    AWS,
    KingsoftCloud,
    AlibabaCloud,
}

#[derive(Debug)]
pub struct S3BackendOptions {
    pub endpoint: String,
    pub region: String,
    pub access_key: String,
    pub secret_key: String,
    pub force_path_style: bool,
    pub vendor: CloudVendor,
}

// AWS endpoint format:
//   - Standard: s3.<region>.amazonaws.com
//   - Dualstack: s3.dualstack.<region>.amazonaws.com
//   - FIPS: s3-fips.<region>.amazonaws.com
//   - FIPS && Dualstack: s3-fips.dualstack.<region>.amazonaws.com
// Reference: https://docs.aws.amazon.com/general/latest/gr/s3.html
// Alibaba Cloud endpoint format:
//   - Internal: oss-<region>-internal.aliyuncs.com
//   - External: oss-<region>.aliyuncs.com
// Reference: https://www.alibabacloud.com/help/en/oss/regions-and-endpoints
// Kingsoft Cloud endpoint format:
//   - Internal: ks3-<region>-internal.ksyuncs.com
//   - External: ks3-<region>.ksyuncs.com
// Reference: https://endocs.ksyun.com/documents/37088
fn detect_cloud_vendor_and_region(
    endpoint: &str,
) -> Result<(CloudVendor, String), BackendParseError> {
    let normalized_endpoint = if endpoint.contains("://") {
        endpoint.to_string()
    } else {
        // assume http if no scheme is provided
        format!("http://{endpoint}")
    };
    let url = url::Url::parse(&normalized_endpoint).map_err(|e| {
        BackendParseError::InvalidOption(format!(
            "failed to parse endpoint, endpoint={endpoint}, err={e:?}",
        ))
    })?;
    let host = url.host_str().ok_or_else(|| {
        BackendParseError::InvalidOption(format!("host not found in endpoint, endpoint={endpoint}"))
    })?;
    let (vendor, region) = if host.contains("amazonaws.com") {
        (CloudVendor::AWS, parse_aws_region(host))
    } else if host.contains("aliyuncs.com") {
        (
            CloudVendor::AlibabaCloud,
            parse_prefixed_region(host, "oss"),
        )
    } else if host.contains("ksyuncs.com") {
        (
            CloudVendor::KingsoftCloud,
            parse_prefixed_region(host, "ks3"),
        )
    } else {
        (CloudVendor::Unknown, String::new())
    };
    Ok((vendor, region))
}

fn parse_aws_region(host: &str) -> String {
    let labels: Vec<&str> = host.split('.').collect();
    if labels.len() < 4 {
        return String::new();
    }
    match (labels[0], labels[1]) {
        // AWS dualstack endpoint
        ("s3", "dualstack") | ("s3-fips", "dualstack") => labels
            .get(2)
            .map_or_else(String::new, |region| (*region).to_string()),
        // AWS standard endpoint or FIPS endpoint
        ("s3", _) | ("s3-fips", _) => labels
            .get(1)
            .map_or_else(String::new, |region| (*region).to_string()),
        // other cases
        _ => String::new(),
    }
}

fn parse_prefixed_region(host: &str, prefix: &str) -> String {
    let first_label = host.split('.').next().unwrap_or_default();
    let Some(without_prefix) = first_label.strip_prefix(&format!("{prefix}-")) else {
        return String::new();
    };
    without_prefix
        .strip_suffix("-internal")
        .unwrap_or(without_prefix)
        .to_string()
}

impl S3BackendOptions {
    pub fn from_query(u: &url::Url) -> Result<Self, BackendParseError> {
        let mut opt = Self {
            endpoint: String::new(),
            region: String::new(),
            access_key: String::new(),
            secret_key: String::new(),
            force_path_style: true,
            vendor: CloudVendor::Unknown,
        };
        u.query_pairs().for_each(|(k, v)| {
            let normalized_key = k.replace('_', "-");
            match normalized_key.as_str() {
                "endpoint" => {
                    opt.endpoint = v.trim_end_matches("/").to_string();
                }
                "region" => {
                    opt.region = v.to_string();
                }
                "access-key" => {
                    opt.access_key = v.to_string();
                }
                "secret-key" | "secret-access-key" => {
                    opt.secret_key = v.to_string();
                }
                "force-path-style" => {
                    opt.force_path_style = v.to_lowercase() == "true";
                }
                _ => {
                    // TODO: maybe log a warning for unknown query param
                }
            }
        });
        if !opt.endpoint.is_empty() {
            let (vendor, detected_region) = detect_cloud_vendor_and_region(&opt.endpoint)?;
            opt.vendor = vendor;
            // if region is not set, use the detected region
            if opt.region.is_empty() {
                opt.region = detected_region;
            }
        }
        opt.validate()?;
        Ok(opt)
    }

    pub fn from_config(config: &S3Config) -> Result<Self, BackendParseError> {
        let mut opt = Self {
            endpoint: config.endpoint.trim_end_matches("/").to_string(),
            region: config.region.clone(),
            access_key: config.access_key.clone(),
            secret_key: config.secret_key.clone(),
            force_path_style: config.use_path_style,
            vendor: CloudVendor::Unknown,
        };
        if !opt.endpoint.is_empty() {
            let (vendor, detected_region) = detect_cloud_vendor_and_region(&opt.endpoint)?;
            opt.vendor = vendor;
            // if region is not set, use the detected region
            if opt.region.is_empty() {
                opt.region = detected_region;
            }
        }
        opt.validate()?;
        Ok(opt)
    }

    pub fn validate(&self) -> Result<&Self, BackendParseError> {
        if !self.endpoint.is_empty() {
            // validate endpoint is a valid url
            let endpoint = url::Url::parse(&self.endpoint).map_err(|e| {
                BackendParseError::InvalidOption(format!("endpoint={} err={}", self.endpoint, e))
            })?;
            if endpoint.scheme().is_empty() {
                return Err(BackendParseError::InvalidOption(format!(
                    "scheme not found in endpoint, endpoint={}",
                    self.endpoint
                )));
            }
            if endpoint.host().is_none() {
                return Err(BackendParseError::InvalidOption(format!(
                    "host not found in endpoint, endpoint={}",
                    self.endpoint
                )));
            }
        }
        if self.access_key.is_empty() && !self.secret_key.is_empty() {
            return Err(BackendParseError::InvalidOption(
                "access_key not found while secret_key is set".to_string(),
            ));
        }
        if !self.access_key.is_empty() && self.secret_key.is_empty() {
            return Err(BackendParseError::InvalidOption(
                "secret_key not found while access_key is set".to_string(),
            ));
        }
        Ok(self)
    }

    pub fn to_query_iter(&self) -> impl Iterator<Item = (&'static str, &str)> {
        let mut params = Vec::new();
        if !self.endpoint.is_empty() {
            params.push(("endpoint", self.endpoint.as_str()));
        }
        if !self.region.is_empty() {
            params.push(("region", self.region.as_str()));
        }
        if !self.access_key.is_empty() {
            params.push(("access-key", self.access_key.as_str()));
        }
        if !self.secret_key.is_empty() {
            params.push(("secret-access-key", self.secret_key.as_str()));
        }
        // default is true, only include it in query when it's false to reduce query
        // length
        if !self.force_path_style {
            params.push(("force-path-style", "false"));
        }
        // vendor is set using "provider" query param
        match self.vendor {
            CloudVendor::AWS | CloudVendor::Unknown => {
                // no need to set provider for AWS and unknown
            }
            CloudVendor::AlibabaCloud => {
                params.push(("provider", "alibaba"));
            }
            CloudVendor::KingsoftCloud => {
                params.push(("provider", "ks3"));
            }
        }
        params.into_iter()
    }
}

#[derive(Debug)]
pub struct S3Backend {
    pub bucket: String,
    pub prefix: String,
    pub options: S3BackendOptions,
}

impl S3Backend {
    pub fn from_url(u: url::Url, raw_url: &str) -> Result<Self, BackendParseError> {
        if u.host().is_none() {
            return Err(BackendParseError::InvalidUrl(format!(
                "Missing bucket in S3 Backend URL, url={}",
                raw_url
            )));
        }
        let prefix = u.path().trim_start_matches('/').to_string();
        let options = S3BackendOptions::from_query(&u)?;
        Ok(Self {
            bucket: u.host_str().unwrap().to_string(),
            prefix,
            options,
        })
    }

    fn to_storage_config(&self) -> S3Config {
        S3Config {
            endpoint: self.options.endpoint.clone(),
            region: self.options.region.clone(),
            access_key: self.options.access_key.clone(),
            secret_key: self.options.secret_key.clone(),
            use_path_style: self.options.force_path_style,
            bucket: self.bucket.clone(),
            prefix: self.prefix.clone(),
        }
    }

    pub async fn create_storage(
        &self,
        service: &'static str,
    ) -> storage::Result<Arc<dyn BackendStorage>> {
        let config = self.to_storage_config();
        storage::create_backend_storage(&config, service).await
    }
}

#[cfg(test)]
mod tests {

    use super::{super::Backend, *};

    #[test]
    fn test_s3_backend_parse_no_options() {
        let url = "s3://test-bucket";
        let backend = Backend::parse(url).unwrap();
        match backend {
            Backend::S3(s3) => {
                assert_eq!(s3.bucket, "test-bucket");
                assert_eq!(s3.prefix, "");
                assert_eq!(s3.options.endpoint, "");
                assert_eq!(s3.options.region, "");
                assert_eq!(s3.options.access_key, "");
                assert_eq!(s3.options.secret_key, "");
                assert!(s3.options.force_path_style);
            }
            _ => unreachable!("Expected S3 backend"),
        }

        let url = "s3://test-bucket/prefix";
        let backend = Backend::parse(url).unwrap();
        match backend {
            Backend::S3(s3) => {
                assert_eq!(s3.bucket, "test-bucket");
                assert_eq!(s3.prefix, "prefix");
                assert_eq!(s3.options.endpoint, "");
                assert_eq!(s3.options.region, "");
                assert_eq!(s3.options.access_key, "");
                assert_eq!(s3.options.secret_key, "");
                assert!(s3.options.force_path_style);
            }
            _ => unreachable!("Expected S3 backend"),
        }
    }

    #[test]
    fn test_s3_backend_parse_with_plus_in_secret() {
        let url = "s3://bucket?access_key=key&secret_key=secret+with+plus";
        let backend = Backend::parse(url).unwrap();

        match backend {
            Backend::S3(s3) => {
                // The plus signs are replaced with %2B during parsing, so they remain as +
                assert_eq!(s3.options.secret_key, "secret+with+plus");
            }
            _ => unreachable!("Expected S3 backend"),
        }
    }

    #[test]
    fn test_s3_backend_parse_with_options() {
        let url = "s3://my-bucket/path/to/prefix?endpoint=https://127.0.0.1&region=us-west-2&access_key=FAKEaccessKEY&secret_key=FAKEsecret_KEY&force_path_style=false";
        let backend = Backend::parse(url).unwrap();

        match backend {
            Backend::S3(s3) => {
                assert_eq!(s3.bucket, "my-bucket");
                assert_eq!(s3.prefix, "path/to/prefix");
                assert_eq!(s3.options.endpoint, "https://127.0.0.1");
                assert_eq!(s3.options.region, "us-west-2");
                assert_eq!(s3.options.access_key, "FAKEaccessKEY");
                assert_eq!(s3.options.secret_key, "FAKEsecret_KEY");
                assert!(!s3.options.force_path_style);
            }
            _ => unreachable!("Expected S3 backend"),
        }

        // endpoint with trailing '/'
        let url = "s3://bucket?endpoint=https://example.com/";
        let backend = Backend::parse(url).unwrap();

        match backend {
            Backend::S3(s3) => {
                assert_eq!(s3.options.endpoint, "https://example.com");
            }
            _ => unreachable!("Expected S3 backend"),
        }

        // hyphenated query keys are also accepted
        let url = "s3://bucket?access-key=ak&secret-key=sk&force-path-style=false";
        let backend = Backend::parse(url).unwrap();
        match backend {
            Backend::S3(s3) => {
                assert_eq!(s3.options.access_key, "ak");
                assert_eq!(s3.options.secret_key, "sk");
                assert!(!s3.options.force_path_style);
            }
            _ => unreachable!("Expected S3 backend"),
        }
    }

    #[test]
    fn test_backend_parse_invalid() {
        for u in [
            "",
            "not-a-url",
            "s3:///path/without/bucket",
            // invalid endpoint
            "s3://bucket?endpoint=invalid-url",
            // endpoint without scheme
            "s3://bucket?endpoint=example.com",
            // access_key, secret_key
            "s3://bucket?access_key=key",
            "s3://bucket?secret_key=secret",
        ] {
            let result = Backend::parse(u);
            assert!(result.is_err());
        }
    }

    #[test]
    fn test_s3_options_unknown_query_param() {
        // This test ensures unknown query parameters are ignored without error
        let url =
            url::Url::parse("s3://bucket?unknown_param=value&access_key=key&secret_key=secret")
                .unwrap();
        let result = S3BackendOptions::from_query(&url);
        assert!(result.is_ok());
        let options = result.unwrap();
        assert_eq!(options.access_key, "key");
        assert_eq!(options.secret_key, "secret");
    }

    #[test]
    fn test_s3_backend_to_storage_config() {
        let backend = Backend::parse("s3://bucket/prefix").unwrap();
        let s3 = match backend {
            Backend::S3(s3) => s3,
            _ => unreachable!("Expected S3 backend"),
        };
        let config = s3.to_storage_config();
        assert_eq!(config.bucket, "bucket");
        assert_eq!(config.prefix, "prefix");
        assert_eq!(config.endpoint, "");
        assert_eq!(config.region, "");
        assert_eq!(config.access_key, "");
        assert_eq!(config.secret_key, "");
        assert!(config.use_path_style);
    }

    #[test]
    fn test_s3_backend_to_storage_config_with_options() {
        let backend = Backend::parse(
            "s3://bucket/path?endpoint=https://127.0.0.1&region=us-west-2&access_key=ak&secret_key=sk&force_path_style=false",
        )
        .unwrap();
        let s3 = match backend {
            Backend::S3(s3) => s3,
            _ => unreachable!("Expected S3 backend"),
        };
        let config = s3.to_storage_config();
        assert_eq!(config.bucket, "bucket");
        assert_eq!(config.prefix, "path");
        assert_eq!(config.endpoint, "https://127.0.0.1");
        assert_eq!(config.region, "us-west-2");
        assert_eq!(config.access_key, "ak");
        assert_eq!(config.secret_key, "sk");
        assert!(!config.use_path_style);
    }

    #[test]
    fn test_detect_cloud_vendor_and_region() {
        struct TestCase {
            endpoint: &'static str,
            expected_region: &'static str,
            expected_vendor: CloudVendor,
        }

        let cases = vec![
            // AWS endpoint
            TestCase {
                endpoint: "http://s3.us-east-1.amazonaws.com",
                expected_region: "us-east-1",
                expected_vendor: CloudVendor::AWS,
            },
            TestCase {
                endpoint: "https://s3.us-west-1.amazonaws.com",
                expected_region: "us-west-1",
                expected_vendor: CloudVendor::AWS,
            },
            // AWS dualstack endpoint
            TestCase {
                endpoint: "http://s3.dualstack.us-east-1.amazonaws.com",
                expected_region: "us-east-1",
                expected_vendor: CloudVendor::AWS,
            },
            // AWS FIPS endpoint
            TestCase {
                endpoint: "https://s3-fips.us-east-1.amazonaws.com",
                expected_region: "us-east-1",
                expected_vendor: CloudVendor::AWS,
            },
            // AWS FIPS dualstack endpoint
            TestCase {
                endpoint: "https://s3-fips.dualstack.us-east-1.amazonaws.com",
                expected_region: "us-east-1",
                expected_vendor: CloudVendor::AWS,
            },
            // Alibaba Cloud endpoint (internal)
            TestCase {
                endpoint: "http://oss-ap-southeast-1-internal.aliyuncs.com",
                expected_region: "ap-southeast-1",
                expected_vendor: CloudVendor::AlibabaCloud,
            },
            TestCase {
                endpoint: "https://oss-eu-central-1-internal.aliyuncs.com",
                expected_region: "eu-central-1",
                expected_vendor: CloudVendor::AlibabaCloud,
            },
            // Alibaba Cloud endpoint (external)
            TestCase {
                endpoint: "http://oss-ap-southeast-1.aliyuncs.com",
                expected_region: "ap-southeast-1",
                expected_vendor: CloudVendor::AlibabaCloud,
            },
            TestCase {
                endpoint: "https://oss-na-south-1.aliyuncs.com",
                expected_region: "na-south-1",
                expected_vendor: CloudVendor::AlibabaCloud,
            },
            // Kingsoft Cloud endpoint (internal)
            TestCase {
                endpoint: "http://ks3-cn-beijing-internal.ksyuncs.com",
                expected_region: "cn-beijing",
                expected_vendor: CloudVendor::KingsoftCloud,
            },
            TestCase {
                endpoint: "http://ks3-sgp-internal.ksyuncs.com",
                expected_region: "sgp",
                expected_vendor: CloudVendor::KingsoftCloud,
            },
            TestCase {
                endpoint: "http://ks3-jr-beijing-internal.ksyuncs.com",
                expected_region: "jr-beijing",
                expected_vendor: CloudVendor::KingsoftCloud,
            },
            // Kingsoft Cloud endpoint (external)
            TestCase {
                endpoint: "http://ks3-cn-beijing.ksyuncs.com",
                expected_region: "cn-beijing",
                expected_vendor: CloudVendor::KingsoftCloud,
            },
            TestCase {
                endpoint: "http://ks3-sgp.ksyuncs.com",
                expected_region: "sgp",
                expected_vendor: CloudVendor::KingsoftCloud,
            },
            TestCase {
                endpoint: "http://ks3-jr-shanghai.ksyuncs.com",
                expected_region: "jr-shanghai",
                expected_vendor: CloudVendor::KingsoftCloud,
            },
            // non-AWS endpoint
            TestCase {
                endpoint: "minio.mydomain.com",
                expected_region: "",
                expected_vendor: CloudVendor::Unknown,
            },
            TestCase {
                endpoint: "http://minio.mydomain.com",
                expected_region: "",
                expected_vendor: CloudVendor::Unknown,
            },
            TestCase {
                endpoint: "http://minio.mydomain.com:80",
                expected_region: "",
                expected_vendor: CloudVendor::Unknown,
            },
            TestCase {
                endpoint: "https://minio.mydomain.com",
                expected_region: "",
                expected_vendor: CloudVendor::Unknown,
            },
            TestCase {
                endpoint: "https://minio.mydomain.com:443",
                expected_region: "",
                expected_vendor: CloudVendor::Unknown,
            },
            TestCase {
                endpoint: "10.0.0.1:9000",
                expected_region: "",
                expected_vendor: CloudVendor::Unknown,
            },
        ];

        for case in cases {
            let (vendor, region) = detect_cloud_vendor_and_region(case.endpoint).unwrap();
            assert_eq!(vendor, case.expected_vendor, "{}", case.endpoint);
            assert_eq!(region, case.expected_region, "{}", case.endpoint);
        }
    }

    #[test]
    fn test_s3_options_from_query_detect_vendor_and_region() {
        struct TestCase {
            url: &'static str,
            expected_region: &'static str,
            expected_vendor: CloudVendor,
        }

        let cases = vec![
            TestCase {
                url: "s3://bucket/prefix?endpoint=https://s3.us-east-1.amazonaws.com",
                expected_region: "us-east-1",
                expected_vendor: CloudVendor::AWS,
            },
            TestCase {
                url: "s3://bucket/prefix?endpoint=https://s3.us-east-1.amazonaws.com&region=us-west-2",
                expected_region: "us-west-2", // region is set explicitly, so it should be used
                expected_vendor: CloudVendor::AWS,
            },
        ];
        for case in cases {
            let url = url::Url::parse(case.url).unwrap();
            let options = S3BackendOptions::from_query(&url).unwrap();
            assert_eq!(options.vendor, case.expected_vendor);
            assert_eq!(options.region, case.expected_region);
        }
    }

    #[test]
    fn test_s3_options_from_config_detect_vendor_and_region() {
        struct TestCase {
            endpoint: &'static str,
            input_region: &'static str,
            expected_region: &'static str,
            expected_vendor: CloudVendor,
        }

        let cases = vec![
            TestCase {
                endpoint: "https://oss-ap-southeast-1.aliyuncs.com",
                input_region: "",
                expected_region: "ap-southeast-1",
                expected_vendor: CloudVendor::AlibabaCloud,
            },
            TestCase {
                endpoint: "https://oss-ap-southeast-1.aliyuncs.com",
                input_region: "ap-north-1",
                expected_region: "ap-north-1", // region is set explicitly, so it should be used
                expected_vendor: CloudVendor::AlibabaCloud,
            },
            TestCase {
                endpoint: "https://s3.us-east-1.amazonaws.com",
                input_region: "",
                expected_region: "us-east-1",
                expected_vendor: CloudVendor::AWS,
            },
        ];
        for case in cases {
            let config = S3Config {
                endpoint: case.endpoint.to_string(),
                region: case.input_region.to_string(),
                ..Default::default()
            };
            let options = S3BackendOptions::from_config(&config).unwrap();
            assert_eq!(options.vendor, case.expected_vendor);
            assert_eq!(options.region, case.expected_region);
        }
    }
}