use std::sync::Arc;

use thiserror::Error;
use tici_common::config::S3Config;

use super::{GCSBackend, S3Backend};
use crate::{
    backend::{gcs::GCSBackendOptions, s3::S3BackendOptions},
    storage::{self, BackendStorage, gcs_storage},
};

#[derive(Debug, Error)]
pub enum BackendParseError {
    #[error("Invalid Backend URL: {0}")]
    InvalidUrl(String),
    #[error("Invalid Backend Option: {0}")]
    InvalidOption(String),
}

pub enum BackendOptions {
    S3(S3BackendOptions),
    GCS(GCSBackendOptions),
}

impl BackendOptions {
    pub fn from_config(config: &S3Config) -> Result<Self, BackendParseError> {
        if config.endpoint.is_empty() {
            return Err(BackendParseError::InvalidOption(
                "endpoint is empty when trying to create BackendOptions".to_string(),
            ));
        }
        if gcs_storage::is_gcs_endpoint(&config.endpoint) {
            GCSBackendOptions::from_config(config).map(Self::GCS)
        } else {
            S3BackendOptions::from_config(config).map(Self::S3)
        }
    }

    pub fn to_query_iter(&self) -> Box<dyn Iterator<Item = (&'static str, &str)> + '_> {
        match self {
            Self::S3(opts) => Box::new(opts.to_query_iter()),
            Self::GCS(opts) => Box::new(opts.to_query_iter()),
        }
    }
}

// Backend represent an abstraction that files read/write under a specified URI
// with authentication, storage class and other options. The URI can be S3/GCS
// or other cloud storage backends or even local file system.
//
// The basic format of the URI
// is as follows: [scheme]://[host]/[path]?[parameters]
// Docs: https://docs.pingcap.com/tidb/v8.5/external-storage-uri/
#[derive(Debug)]
pub enum Backend {
    S3(S3Backend),
    GCS(GCSBackend),
}

fn parse_raw_url(raw_url: &str) -> Result<url::Url, url::ParseError> {
    // https://github.com/pingcap/br/issues/603
    // In aws the secret key may contain '/+=' and '+' has a special meaning in
    // URL. Replace "+" by "%2B" here to avoid this problem.
    let u = raw_url.replace('+', "%2B");
    url::Url::parse(&u)
}

impl Backend {
    pub fn parse(raw_url: &str) -> Result<Self, BackendParseError> {
        if raw_url.is_empty() {
            return Err(BackendParseError::InvalidUrl(
                "Backend URI cannot be empty".to_string(),
            ));
        }
        let u = parse_raw_url(raw_url).map_err(|e| {
            BackendParseError::InvalidUrl(format!(
                "Failed to parse Backend URL, url={} err={}",
                raw_url, e
            ))
        })?;
        match u.scheme() {
            "s3" => S3Backend::from_url(u, raw_url).map(Self::S3),
            "gcs" | "gs" => GCSBackend::from_url(u, raw_url).map(Self::GCS),
            _ => Err(BackendParseError::InvalidUrl(format!(
                "Unsupported scheme in Backend URL, scheme={} url={}",
                u.scheme(),
                raw_url
            ))),
        }
    }

    pub async fn create_storage(
        &self,
        service: &'static str,
    ) -> storage::Result<Arc<dyn BackendStorage>> {
        match self {
            Self::S3(s3) => s3.create_storage(service).await,
            Self::GCS(gcs) => gcs.create_storage(service).await,
        }
    }
}

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

    #[test]
    fn test_parse_raw_url_with_plus() {
        let raw_url = "s3://bucket?secret_key=secret+key+with+plus";
        let parsed = parse_raw_url(raw_url).unwrap();

        // Check that the URL parsing actually works and the query is preserved
        assert_eq!(parsed.scheme(), "s3");
        assert_eq!(parsed.host_str(), Some("bucket"));
        assert!(parsed.query().is_some());

        // Since the function replaces + with %2B, the decoded query should have spaces
        let query_pairs: Vec<_> = parsed.query_pairs().collect();
        let secret_key_pair = query_pairs.iter().find(|(k, _)| k == "secret_key");
        assert!(secret_key_pair.is_some());

        let (_, value) = secret_key_pair.unwrap();
        // Should keep the '+' in secret_key
        assert_eq!(value, "secret+key+with+plus");
    }

    #[test]
    fn test_non_support_backend_parse() {
        let url = "file:///path/to/file";
        let result = Backend::parse(url);
        assert!(result.is_err());
        match result.unwrap_err() {
            BackendParseError::InvalidUrl(msg) => {
                assert!(msg.contains("Unsupported scheme"));
                assert!(msg.contains("file"));
            }
            _ => panic!("Expected InvalidUrl error"),
        }
    }
}