//! Reference HTTP server for the Oxygen wire protocol (spec §15.1, §16).
//!
//! Serves every repository directory under a root at `/{repo}/oxy/v1/...`,
//! over TLS when a certificate and key are given, else plain HTTP for use
//! behind a TLS-terminating proxy or on a loopback interface (spec §15.1).
//!
//! Authentication: when bearer tokens are configured, `push` requires one;
//! reads are open. With no tokens configured everything is open (development).

use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{header, HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::Router;
use oxy_core::protocol::{wire, CONTENT_TYPE};
use oxy_core::repo::Repo;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

#[derive(Clone, Debug)]
pub struct ServerConfig {
    /// Directory containing one repository per subdirectory.
    pub root: PathBuf,
    /// Bearer tokens allowed to push; empty means no authentication.
    pub tokens: Vec<String>,
    /// Serve HTTPS with this certificate chain and private key (PEM files).
    pub tls: Option<TlsFiles>,
}

#[derive(Clone, Debug)]
pub struct TlsFiles {
    pub cert: PathBuf,
    pub key: PathBuf,
}

struct App {
    cfg: ServerConfig,
    /// One lock per repository so concurrent pushes serialize.
    locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
}

type Shared = Arc<App>;

struct Failure(StatusCode, String);

impl IntoResponse for Failure {
    fn into_response(self) -> Response {
        (self.0, self.1).into_response()
    }
}

impl From<oxy_core::repo::RepoError> for Failure {
    fn from(e: oxy_core::repo::RepoError) -> Failure {
        Failure(StatusCode::BAD_REQUEST, e.to_string())
    }
}

fn ok(body: Vec<u8>) -> Response {
    ([(header::CONTENT_TYPE, CONTENT_TYPE)], body).into_response()
}

fn valid_repo_name(name: &str) -> bool {
    !name.is_empty()
        && !name.starts_with('.')
        && name.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
}

impl App {
    fn open(&self, name: &str) -> Result<Repo, Failure> {
        if !valid_repo_name(name) {
            return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
        }
        let dir = self.cfg.root.join(name);
        if !dir.join(".oxy").is_dir() {
            return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
        }
        Repo::open(&dir).map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
    }

    fn lock_for(&self, name: &str) -> Arc<Mutex<()>> {
        self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()
    }

    fn authorize_push(&self, headers: &HeaderMap) -> Result<(), Failure> {
        if self.cfg.tokens.is_empty() {
            return Ok(());
        }
        let presented = headers
            .get(header::AUTHORIZATION)
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "))
            .map(str::trim);
        match presented {
            Some(t) if self.cfg.tokens.iter().any(|x| x == t) => Ok(()),
            _ => Err(Failure(StatusCode::UNAUTHORIZED, "a valid bearer token is required to push".into())),
        }
    }
}

async fn blocking<T: Send + 'static>(f: impl FnOnce() -> Result<T, Failure> + Send + 'static) -> Result<T, Failure> {
    tokio::task::spawn_blocking(f).await.map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
}

async fn capabilities(State(app): State<Shared>, Path(repo): Path<String>) -> Result<Response, Failure> {
    blocking(move || {
        let r = app.open(&repo)?;
        Ok(ok(wire::encode_capabilities(r.id()?)))
    })
    .await
}

async fn refs(State(app): State<Shared>, Path(repo): Path<String>) -> Result<Response, Failure> {
    blocking(move || {
        let r = app.open(&repo)?;
        Ok(ok(wire::encode_refs(r.id()?, &r.remote_refs()?)))
    })
    .await
}

async fn negotiate(State(app): State<Shared>, Path(repo): Path<String>, body: Bytes) -> Result<Response, Failure> {
    blocking(move || {
        let r = app.open(&repo)?;
        let (have, want) = wire::decode_negotiate(&body)?;
        Ok(ok(wire::encode_hashes(&r.negotiate(&have, &want)?)))
    })
    .await
}

async fn fetch(State(app): State<Shared>, Path(repo): Path<String>, body: Bytes) -> Result<Response, Failure> {
    blocking(move || {
        let r = app.open(&repo)?;
        let (changes, max_blob) = wire::decode_fetch(&body)?;
        Ok(ok(wire::encode_bundle(&r.bundle_for_filtered(&changes, max_blob)?)))
    })
    .await
}

async fn objects(State(app): State<Shared>, Path(repo): Path<String>, body: Bytes) -> Result<Response, Failure> {
    blocking(move || {
        let r = app.open(&repo)?;
        let refs = wire::decode_hashes(&body)?;
        let bundle = oxy_core::protocol::Bundle { objects: r.objects_for(&refs)?, ..Default::default() };
        Ok(ok(wire::encode_bundle(&bundle)))
    })
    .await
}

async fn push(State(app): State<Shared>, Path(repo): Path<String>, headers: HeaderMap, body: Bytes) -> Result<Response, Failure> {
    app.authorize_push(&headers)?;
    blocking(move || {
        let r = app.open(&repo)?;
        let (bundle, updates, policy) = wire::decode_push(&body)?;
        let lock = app.lock_for(&repo);
        let _guard = lock.lock().unwrap();
        Ok(ok(wire::encode_push_response(&r.receive_push(bundle, updates, policy)?)))
    })
    .await
}

pub fn router(cfg: ServerConfig) -> Router {
    let app: Shared = Arc::new(App { cfg, locks: Mutex::new(HashMap::new()) });
    Router::new()
        .route("/{repo}/oxy/v1/capabilities", get(capabilities))
        .route("/{repo}/oxy/v1/refs", get(refs))
        .route("/{repo}/oxy/v1/negotiate", post(negotiate))
        .route("/{repo}/oxy/v1/fetch", post(fetch))
        .route("/{repo}/oxy/v1/objects", post(objects))
        .route("/{repo}/oxy/v1/push", post(push))
        .with_state(app)
}

/// Serve on an already-bound listener until the process ends; HTTPS when
/// `cfg.tls` is set.
pub async fn serve(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
    listener.set_nonblocking(true)?;
    match cfg.tls.clone() {
        Some(tls) => {
            let _ = rustls::crypto::ring::default_provider().install_default();
            let rustls_cfg = axum_server::tls_rustls::RustlsConfig::from_pem_file(&tls.cert, &tls.key)
                .await
                .map_err(|e| std::io::Error::other(format!("loading TLS files: {e}")))?;
            axum_server::from_tcp_rustls(listener, rustls_cfg)?.serve(router(cfg).into_make_service()).await
        }
        None => {
            let listener = tokio::net::TcpListener::from_std(listener)?;
            axum::serve(listener, router(cfg)).await
        }
    }
}

/// Build a runtime and serve; for the CLI and for tests on a background thread.
pub fn run_blocking(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {
    let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build()?;
    rt.block_on(serve(listener, cfg))
}