use reqwest::{Client, Url};
use serde_json::{Map, Value};

pub const D1_ENGINE_NAME: &str = "cloudflare-d1";
pub const HEADER_D1_BOOKMARK: &str = "x-athena-d1-bookmark";
pub const HEADER_D1_SESSION_MODE: &str = "x-athena-d1-session-mode";

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct D1ConnectionInfo {
    pub worker_base_url: String,
    pub auth_token_env_var: String,
    pub database_binding: String,
    pub default_session_mode: Option<String>,
}

#[derive(Debug, Clone)]
pub struct D1ExecutionResult {
    pub rows: Vec<Value>,
    pub columns: Vec<String>,
    pub duration_ms: Option<u64>,
    pub bookmark: Option<String>,
    pub count: Option<u64>,
    pub meta: Value,
}

impl D1ConnectionInfo {
    pub fn from_metadata(metadata: &Value) -> Result<Option<Self>, String> {
        let Some(cloudflare_d1) = metadata.get("cloudflareD1") else {
            return Ok(None);
        };
        let Some(cloudflare_d1) = cloudflare_d1.as_object() else {
            return Err("cloudflareD1 metadata must be an object".to_string());
        };

        let engine: &str = metadata
            .get("dbEngine")
            .and_then(Value::as_str)
            .map(str::trim)
            .unwrap_or(D1_ENGINE_NAME);
        if !engine.eq_ignore_ascii_case(D1_ENGINE_NAME) {
            return Ok(None);
        }

        let worker_base_url = required_string(
            cloudflare_d1,
            "worker_base_url",
            "cloudflareD1.worker_base_url",
        )?;
        validate_worker_base_url(&worker_base_url)?;

        let auth_token_env_var = required_string(
            cloudflare_d1,
            "auth_token_env_var",
            "cloudflareD1.auth_token_env_var",
        )?;
        let database_binding = required_string(
            cloudflare_d1,
            "database_binding",
            "cloudflareD1.database_binding",
        )?;
        let default_session_mode = optional_string(
            cloudflare_d1,
            "default_session_mode",
            "cloudflareD1.default_session_mode",
        )?;

        Ok(Some(Self {
            worker_base_url,
            auth_token_env_var,
            database_binding,
            default_session_mode,
        }))
    }

    pub fn resolve_auth_token(&self) -> Result<String, String> {
        let value = std::env::var(&self.auth_token_env_var).map_err(|_| {
            format!(
                "missing Cloudflare D1 proxy auth token env var '{}'",
                self.auth_token_env_var
            )
        })?;
        let trimmed = value.trim();
        if trimmed.is_empty() {
            return Err(format!(
                "Cloudflare D1 proxy auth token env var '{}' is empty",
                self.auth_token_env_var
            ));
        }
        Ok(trimmed.to_string())
    }

    pub fn effective_session_mode(&self, requested: Option<&str>) -> Option<String> {
        requested
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_string)
            .or_else(|| self.default_session_mode.clone())
    }
}

pub async fn execute_query_via_proxy(
    _http: &Client,
    _info: &D1ConnectionInfo,
    _query: &str,
    _params: Vec<Value>,
    _requested_session_mode: Option<&str>,
    _bookmark: Option<&str>,
    _retry_writes: bool,
) -> Result<D1ExecutionResult, String> {
    Err("Cloudflare D1 support is disabled in this build. Enable the `d1-driver` feature to execute queries through the proxy.".to_string())
}

fn required_string(object: &Map<String, Value>, key: &str, path: &str) -> Result<String, String> {
    let Some(value) = object.get(key) else {
        return Err(format!("{path} is required"));
    };
    let Some(value) = value.as_str() else {
        return Err(format!("{path} must be a string"));
    };
    let trimmed = value.trim();
    if trimmed.is_empty() {
        return Err(format!("{path} must not be empty"));
    }
    Ok(trimmed.to_string())
}

fn optional_string(
    object: &Map<String, Value>,
    key: &str,
    path: &str,
) -> Result<Option<String>, String> {
    match object.get(key) {
        Some(value) if value.is_null() => Ok(None),
        Some(value) => {
            let Some(value) = value.as_str() else {
                return Err(format!("{path} must be a string"));
            };
            let trimmed = value.trim();
            if trimmed.is_empty() {
                Ok(None)
            } else {
                Ok(Some(trimmed.to_string()))
            }
        }
        None => Ok(None),
    }
}

fn validate_worker_base_url(value: &str) -> Result<(), String> {
    let url = Url::parse(value).map_err(|_| {
        format!("cloudflareD1.worker_base_url must be a valid absolute http or https URL")
    })?;
    match url.scheme() {
        "http" | "https" => Ok(()),
        _ => Err("cloudflareD1.worker_base_url must use http or https".to_string()),
    }
}

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

    #[test]
    fn parses_cloudflare_d1_metadata_when_present() {
        let metadata = json!({
            "dbEngine": "cloudflare-d1",
            "cloudflareD1": {
                "worker_base_url": "https://d1.example.com",
                "auth_token_env_var": "D1_TOKEN",
                "database_binding": "DB",
                "default_session_mode": "read-write"
            }
        });

        let info = D1ConnectionInfo::from_metadata(&metadata)
            .expect("metadata parses")
            .expect("d1 metadata present");
        assert_eq!(info.database_binding, "DB");
        assert_eq!(info.default_session_mode.as_deref(), Some("read-write"));
    }

    #[tokio::test]
    async fn execute_query_reports_feature_disabled() {
        let info = D1ConnectionInfo {
            worker_base_url: "https://d1.example.com".to_string(),
            auth_token_env_var: "D1_TOKEN".to_string(),
            database_binding: "DB".to_string(),
            default_session_mode: None,
        };
        let err = execute_query_via_proxy(
            &Client::new(),
            &info,
            "SELECT 1",
            Vec::new(),
            None,
            None,
            false,
        )
        .await
        .expect_err("feature is disabled in the default build");
        assert!(err.contains("d1-driver"));
    }
}
