use std::net::TcpStream;

use anyhow::Context;
use std::io::Write;
use std::net::TcpListener;
use std::panic::{self, AssertUnwindSafe};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use crate::s3_listener::constants::{S3StorageMode, MAX_ACTIVE_CONNECTIONS};
use crate::s3_listener::dispatch::ListenerSurface;
use crate::s3_listener::errors::write_http_error;
use crate::s3_listener::parser::read_http_request;
use crate::s3_listener::S3ServeOptions;
use bucketwarden_server::{BucketWarden, OperatorRole};

macro_rules! log_listener {
    ($($arg:tt)*) => {{
        let _ = writeln!(std::io::stderr(), $($arg)*);
    }};
}

pub(crate) fn serve(options: S3ServeOptions) -> anyhow::Result<()> {
    let mut runtime = match &options.storage {
        S3StorageMode::Filesystem { data_dir } if data_dir.join("manifest.json").exists() => {
            BucketWarden::restore_filesystem_store(options.config.clone(), data_dir)?
        }
        _ => BucketWarden::new(options.config.clone())?,
    };
    configure_serve_identity(&mut runtime, &options)?;

    let storage = Arc::new(options.storage);
    let runtime = Arc::new(Mutex::new(runtime));
    let s3_listener = TcpListener::bind(&options.bind)
        .with_context(|| format!("failed to bind S3 listener on `{}`", options.bind))?;
    let console_listener = TcpListener::bind(&options.console_bind).with_context(|| {
        format!(
            "failed to bind browser console listener on `{}`",
            options.console_bind
        )
    })?;
    log_listener!("bucketwarden s3 listening on http://{}", options.bind);
    log_listener!(
        "bucketwarden browser console listening on http://{}/ui",
        options.console_bind
    );
    match storage.as_ref() {
        S3StorageMode::Filesystem { data_dir } => {
            log_listener!(
                "bucketwarden s3 storage mode: filesystem data_dir={}",
                data_dir.display()
            );
        }
        S3StorageMode::InMemory => {
            log_listener!("bucketwarden s3 storage mode: in-memory ephemeral=true");
        }
    }
    let console_runtime = Arc::clone(&runtime);
    let console_storage = Arc::clone(&storage);
    thread::Builder::new()
        .name("bucketwarden-console-listener".to_string())
        .spawn(move || {
            if let Err(error) = serve_listener(
                console_listener,
                console_runtime,
                console_storage,
                ListenerSurface::Console,
                "console",
            ) {
                log_listener!("bucketwarden console listener error: {error}");
            }
        })
        .context("failed to spawn browser console listener")?;

    serve_listener(s3_listener, runtime, storage, ListenerSurface::S3, "s3")
}

fn serve_listener(
    listener: TcpListener,
    runtime: Arc<Mutex<BucketWarden>>,
    storage: Arc<S3StorageMode>,
    surface: ListenerSurface,
    label: &'static str,
) -> anyhow::Result<()> {
    let active_connections = Arc::new(AtomicUsize::new(0));
    for stream in listener.incoming() {
        let stream = stream?;
        let active_connections = Arc::clone(&active_connections);
        let runtime = Arc::clone(&runtime);
        let storage = Arc::clone(&storage);
        if let Err(error) = stream.set_read_timeout(Some(Duration::from_secs(15))) {
            log_listener!("bucketwarden {label} read-timeout init error: {error}");
            continue;
        }
        if let Err(error) = stream.set_write_timeout(Some(Duration::from_secs(15))) {
            log_listener!("bucketwarden {label} write-timeout init error: {error}");
            continue;
        }
        let active = active_connections.fetch_add(1, Ordering::AcqRel) + 1;
        if active > MAX_ACTIVE_CONNECTIONS {
            active_connections.fetch_sub(1, Ordering::AcqRel);
            if let Err(error) = crate::s3_listener::errors::reject_overloaded_connection(stream) {
                log_listener!("bucketwarden {label} overload reject write error: {error}");
            }
            continue;
        }
        let active_connections_for_thread = Arc::clone(&active_connections);
        if let Err(error) = thread::Builder::new()
            .name(format!("bucketwarden-{label}-connection"))
            .spawn(move || {
                let result = panic::catch_unwind(AssertUnwindSafe(|| {
                    handle_connection(stream, runtime, storage, surface)
                }));
                if let Err(error) = result {
                    log_listener!("bucketwarden {label} connection panic: {:?}", error);
                } else if let Ok(Err(error)) = result {
                    log_listener!("bucketwarden {label} connection error: {error}");
                }
                active_connections_for_thread.fetch_sub(1, Ordering::AcqRel);
            })
        {
            log_listener!("bucketwarden {label} connection handler spawn error: {error}");
            active_connections.fetch_sub(1, Ordering::AcqRel);
        }
    }
    Ok(())
}

fn configure_serve_identity(
    runtime: &mut BucketWarden,
    options: &S3ServeOptions,
) -> anyhow::Result<()> {
    runtime.allow(&options.principal, "s3:*", "*");
    runtime.register_sigv4_credentials(
        &options.principal,
        options.access_key_id.clone(),
        options.secret_access_key.clone(),
    )?;
    runtime.create_custom_identity(&options.principal, &options.secret_access_key)?;
    runtime.assign_operator_role(
        &options.principal,
        &options.principal,
        OperatorRole::ClusterAdmin,
        "*",
    )?;
    Ok(())
}

pub(crate) fn handle_connection(
    mut stream: TcpStream,
    runtime: Arc<Mutex<BucketWarden>>,
    storage: Arc<S3StorageMode>,
    surface: ListenerSurface,
) -> anyhow::Result<()> {
    let mut buffer = Vec::new();
    let mut chunk = [0_u8; 4096];

    let request = match read_http_request(&mut stream, &mut buffer, &mut chunk) {
        Ok((headers, body)) => {
            let result = crate::s3_listener::dispatch::dispatch(
                &mut stream,
                &headers,
                body,
                runtime,
                storage,
                surface,
            );
            match result {
                Ok(()) => Ok(()),
                Err(error) => {
                    let message = error.to_string();
                    if message.contains("runtime lock poisoned")
                        || message.contains("filesystem persistence failed")
                    {
                        write_http_error(&mut stream, 500, &message)
                    } else {
                        write_http_error(&mut stream, 400, &message)
                    }
                }
            }
        }
        Err(error) => write_http_error(&mut stream, error.status, &error.message),
    };
    request?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use bucketwarden_server::{
        BrowserUiLoginRequest, ConsoleApiRequest, RuntimeConfig, RuntimeError,
    };

    #[test]
    fn s3_serve_identity_supports_browser_ui_login_and_reports() -> Result<(), RuntimeError> {
        let options = S3ServeOptions {
            bind: "127.0.0.1:0".to_string(),
            console_bind: "127.0.0.1:0".to_string(),
            config: RuntimeConfig::development(),
            principal: "demo-root".to_string(),
            access_key_id: "demo-access".to_string(),
            secret_access_key: "demo-shared-secret".to_string(),
            storage: S3StorageMode::InMemory,
        };
        let mut runtime = BucketWarden::new(RuntimeConfig::development())?;

        configure_serve_identity(&mut runtime, &options).expect("serve identity should configure");
        let session = runtime.browser_ui_login(BrowserUiLoginRequest {
            principal_id: "demo-root".to_string(),
            shared_secret: "demo-shared-secret".to_string(),
        })?;
        let overview = runtime.handle_console_api(ConsoleApiRequest {
            method: "GET".to_string(),
            path: "/ui/api/overview".to_string(),
            access_key_id: Some(session.access_key_id),
            ..ConsoleApiRequest::default()
        })?;

        assert_eq!(overview.status, 200);
        assert_eq!(overview.body["session"]["principal_id"], "demo-root");
        Ok(())
    }
}
