use anyhow::anyhow;
use bucketwarden_s3::S3HttpRequest;
use bucketwarden_server::{ConsoleApiRequest, ConsoleApiResponse};
use std::collections::BTreeMap;
use std::io::Write;
use std::sync::{Arc, Mutex};

use crate::s3_listener::errors::reason_phrase;
use crate::s3_listener::parser::{parse_request_line, split_target};
use crate::s3_listener::S3StorageMode;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ListenerSurface {
    S3,
    Console,
}

pub(crate) fn dispatch(
    stream: &mut std::net::TcpStream,
    headers_text: &str,
    body: Vec<u8>,
    runtime: Arc<Mutex<bucketwarden_server::BucketWarden>>,
    storage: Arc<S3StorageMode>,
    surface: ListenerSurface,
) -> anyhow::Result<()> {
    let mut lines = headers_text.lines();
    let request_line = lines
        .next()
        .ok_or_else(|| anyhow!("missing request line"))?;
    let (method, target) = parse_request_line(request_line)?;
    let (path, query) = split_target(target)?;
    let headers = header_map(lines.clone());

    if surface == ListenerSurface::Console {
        return dispatch_browser_ui(
            stream, &method, &path, query, headers, body, runtime, storage,
        );
    }

    let head_only = method == "HEAD";
    let mut request = S3HttpRequest::new("anonymous", &method, path).with_body(body);
    for (key, value) in query {
        request = request.with_query(key, value);
    }

    let mut saw_host = false;
    for line in lines {
        let Some((raw_key, value)) = line.split_once(':') else {
            continue;
        };
        let key = raw_key.trim().to_ascii_lowercase();
        let value = value.trim().to_string();
        if key == "host" {
            saw_host = true;
        }
        request = request.with_header(key, value);
    }
    if !saw_host {
        request = request.with_header("host", "127.0.0.1");
    }

    let response = runtime
        .lock()
        .map_err(|error| anyhow!("runtime lock poisoned: {error}"))?
        .handle_s3_http(request)?;
    if is_mutating_method(&method) && response.status < 400 {
        if let S3StorageMode::Filesystem { data_dir } = storage.as_ref() {
            runtime
                .lock()
                .map_err(|error| anyhow!("runtime lock poisoned: {error}"))?
                .save_filesystem_store(data_dir)
                .map_err(|error| anyhow!("filesystem persistence failed: {error}"))?;
        }
    }
    let response = response.with_header("connection", "close");
    let mut http = format!(
        "HTTP/1.1 {} {}\r\n",
        response.status,
        reason_phrase(response.status)
    );
    let mut headers = BTreeMap::new();
    headers.insert(
        "content-length".to_string(),
        response.body.len().to_string(),
    );
    for (key, value) in response.headers {
        headers.insert(key, value);
    }
    for (key, value) in headers {
        http.push_str(&format!("{key}: {value}\r\n"));
    }
    http.push_str("\r\n");

    stream.write_all(http.as_bytes())?;
    if !head_only {
        stream.write_all(&response.body)?;
    }
    stream.flush()?;
    Ok(())
}

fn is_mutating_method(method: &str) -> bool {
    matches!(method, "DELETE" | "POST" | "PUT")
}

fn dispatch_browser_ui(
    stream: &mut std::net::TcpStream,
    method: &str,
    path: &str,
    query: Vec<(String, String)>,
    headers: BTreeMap<String, String>,
    body: Vec<u8>,
    runtime: Arc<Mutex<bucketwarden_server::BucketWarden>>,
    storage: Arc<S3StorageMode>,
) -> anyhow::Result<()> {
    if path.starts_with("/ui/api") || path.starts_with("/api/v1") {
        let api_path = normalize_console_api_path(path);
        let body = if body.is_empty() {
            None
        } else {
            Some(serde_json::from_slice(&body)?)
        };
        let request = ConsoleApiRequest {
            method: method.to_string(),
            path: api_path,
            access_key_id: ui_access_key(&headers),
            query: query.into_iter().collect(),
            body,
        };
        let response = runtime
            .lock()
            .map_err(|error| anyhow!("runtime lock poisoned: {error}"))?
            .handle_console_api(request)?;
        if is_mutating_method(method) && response.status < 400 {
            persist_runtime_if_needed(&runtime, storage.as_ref())?;
        }
        return write_console_api_response(stream, response);
    }

    if !matches!(method, "GET" | "HEAD") {
        let body =
            b"{\"code\":\"MethodNotAllowed\",\"message\":\"UI assets support GET and HEAD\"}";
        return write_raw_response(
            stream,
            405,
            "application/json",
            BTreeMap::new(),
            body,
            false,
        );
    }
    let response = runtime
        .lock()
        .map_err(|error| anyhow!("runtime lock poisoned: {error}"))?
        .browser_ui_asset_response(path)?;
    write_raw_response(
        stream,
        response.status,
        &response.content_type,
        response.headers,
        &response.body,
        method == "HEAD",
    )
}

fn normalize_console_api_path(path: &str) -> String {
    path.strip_prefix("/api/v1")
        .map(|suffix| format!("/ui/api{suffix}"))
        .unwrap_or_else(|| path.to_string())
}

fn write_console_api_response(
    stream: &mut std::net::TcpStream,
    response: ConsoleApiResponse,
) -> anyhow::Result<()> {
    let body = serde_json::to_vec(&response.body)?;
    let headers = BTreeMap::from([("x-bucketwarden-request-id".to_string(), response.request_id)]);
    write_raw_response(
        stream,
        response.status,
        &response.content_type,
        headers,
        &body,
        false,
    )
}

fn write_raw_response(
    stream: &mut std::net::TcpStream,
    status: u16,
    content_type: &str,
    extra_headers: BTreeMap<String, String>,
    body: &[u8],
    head_only: bool,
) -> anyhow::Result<()> {
    let mut http = format!("HTTP/1.1 {} {}\r\n", status, reason_phrase(status));
    let mut headers = BTreeMap::from([
        ("connection".to_string(), "close".to_string()),
        ("content-length".to_string(), body.len().to_string()),
        ("content-type".to_string(), content_type.to_string()),
    ]);
    headers.extend(extra_headers);
    for (key, value) in headers {
        http.push_str(&format!("{key}: {value}\r\n"));
    }
    http.push_str("\r\n");
    stream.write_all(http.as_bytes())?;
    if !head_only {
        stream.write_all(body)?;
    }
    stream.flush()?;
    Ok(())
}

fn persist_runtime_if_needed(
    runtime: &Arc<Mutex<bucketwarden_server::BucketWarden>>,
    storage: &S3StorageMode,
) -> anyhow::Result<()> {
    if let S3StorageMode::Filesystem { data_dir } = storage {
        runtime
            .lock()
            .map_err(|error| anyhow!("runtime lock poisoned: {error}"))?
            .save_filesystem_store(data_dir)
            .map_err(|error| anyhow!("filesystem persistence failed: {error}"))?;
    }
    Ok(())
}

fn header_map<'a>(lines: impl Iterator<Item = &'a str>) -> BTreeMap<String, String> {
    lines
        .filter_map(|line| {
            let (key, value) = line.split_once(':')?;
            Some((key.trim().to_ascii_lowercase(), value.trim().to_string()))
        })
        .collect()
}

fn ui_access_key(headers: &BTreeMap<String, String>) -> Option<String> {
    headers
        .get("x-bucketwarden-ui-access-key")
        .cloned()
        .or_else(|| {
            headers
                .get("authorization")
                .and_then(|value| value.strip_prefix("Bearer "))
                .map(str::to_string)
        })
}

#[cfg(test)]
mod tests {
    use super::*;
    use bucketwarden_server::{BucketWarden, RuntimeConfig};
    use std::io::Read;
    use std::net::{TcpListener, TcpStream};

    fn stream_pair() -> anyhow::Result<(std::net::TcpStream, TcpStream)> {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let addr = listener.local_addr()?;
        let client = TcpStream::connect(addr)?;
        let (server, _) = listener.accept()?;
        Ok((server, client))
    }

    fn dispatch_to_string(surface: ListenerSurface, request: &str) -> anyhow::Result<String> {
        let (mut server, mut client) = stream_pair()?;
        let runtime = Arc::new(Mutex::new(BucketWarden::new(RuntimeConfig::development())?));
        let storage = Arc::new(S3StorageMode::InMemory);
        dispatch(&mut server, request, Vec::new(), runtime, storage, surface)?;
        drop(server);
        let mut output = String::new();
        client.read_to_string(&mut output)?;
        Ok(output)
    }

    #[test]
    fn s3_surface_does_not_serve_browser_ui_routes() -> anyhow::Result<()> {
        let output = dispatch_to_string(ListenerSurface::S3, "GET /ui HTTP/1.1\r\n\r\n")?;
        assert!(!output.starts_with("HTTP/1.1 200 "));
        assert!(!output.contains("<!doctype html>"));
        assert!(!output.contains("content-type: text/html"));
        Ok(())
    }

    #[test]
    fn console_surface_serves_browser_ui_routes() -> anyhow::Result<()> {
        let output = dispatch_to_string(ListenerSurface::Console, "GET /ui HTTP/1.1\r\n\r\n")?;
        assert!(output.starts_with("HTTP/1.1 200 "));
        assert!(output.contains("<!doctype html>"));
        Ok(())
    }

    #[test]
    fn console_surface_serves_versioned_browser_ui_routes() -> anyhow::Result<()> {
        let output = dispatch_to_string(
            ListenerSurface::Console,
            "GET /ui/v1/buckets HTTP/1.1\r\n\r\n",
        )?;
        assert!(output.starts_with("HTTP/1.1 200 "));
        assert!(output.contains("<!doctype html>"));
        assert!(output.contains("BucketWarden"));
        Ok(())
    }

    #[test]
    fn versioned_console_api_routes_dispatch_to_console_api() -> anyhow::Result<()> {
        let output = dispatch_to_string(
            ListenerSurface::Console,
            "GET /api/v1/identity-providers HTTP/1.1\r\n\r\n",
        )?;
        assert!(output.starts_with("HTTP/1.1 200 "));
        assert!(output.contains("custom-shared-secret"));
        Ok(())
    }
}
