//! 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).
//!
//! Hosts embed the protocol by implementing [`Backend`] (name resolution
//! and authorization) and routing to [`Server::handle`]; `router` is the
//! standalone server built on [`DirBackend`].

use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{header, HeaderMap, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::any;
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,
}

/// Which kind of access a request needs (spec §15.1: reads may be open;
/// a push must be authenticated when any credential is configured).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Access {
    Read,
    Push,
}

/// What a host provides to the protocol server: how a repository name maps
/// to a repository on disk and who may do what. `oxy serve` uses
/// [`DirBackend`]; a hub supplies its own with accounts and roles.
pub trait Backend: Send + Sync + 'static {
    /// Open the repository called `name`, or fail with the status to
    /// answer. An unknown name is `404`; a name the caller may not see
    /// SHOULD also be `404`, so private repositories are not enumerable.
    fn open(&self, name: &str) -> Result<Repo, Failure>;

    /// Decide whether this request may perform `access` on `name`.
    fn authorize(&self, name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure>;

    /// Called after a push was processed, with its per-reference results
    /// (accepted or refused), for logs and cache invalidation.
    fn after_push(&self, _name: &str, _results: &[oxy_core::protocol::RefResult]) {}
}

/// A protocol server over any [`Backend`], serialising the pushes to each
/// repository. Hosts route requests to [`Server::handle`].
pub struct Server {
    backend: Box<dyn Backend>,
    locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
}

/// An error response: a status and a plain-text body.
#[derive(Debug)]
pub struct Failure(pub StatusCode, pub 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())
    }
}

/// The wire protocol's endpoints (spec §16.1), named as in the URL.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Endpoint {
    Capabilities,
    Refs,
    Negotiate,
    Fetch,
    Objects,
    Checkpoint,
    Shallow,
    Push,
}

impl Endpoint {
    pub fn parse(name: &str) -> Option<Endpoint> {
        Some(match name {
            "capabilities" => Endpoint::Capabilities,
            "refs" => Endpoint::Refs,
            "negotiate" => Endpoint::Negotiate,
            "fetch" => Endpoint::Fetch,
            "objects" => Endpoint::Objects,
            "checkpoint" => Endpoint::Checkpoint,
            "shallow" => Endpoint::Shallow,
            "push" => Endpoint::Push,
            _ => return None,
        })
    }

    pub fn name(self) -> &'static str {
        match self {
            Endpoint::Capabilities => "capabilities",
            Endpoint::Refs => "refs",
            Endpoint::Negotiate => "negotiate",
            Endpoint::Fetch => "fetch",
            Endpoint::Objects => "objects",
            Endpoint::Checkpoint => "checkpoint",
            Endpoint::Shallow => "shallow",
            Endpoint::Push => "push",
        }
    }

    /// The HTTP method the endpoint answers to.
    pub fn method(self) -> Method {
        match self {
            Endpoint::Capabilities | Endpoint::Refs => Method::GET,
            _ => Method::POST,
        }
    }

    fn access(self) -> Access {
        match self {
            Endpoint::Push => Access::Push,
            _ => Access::Read,
        }
    }
}

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

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

impl Server {
    pub fn new(backend: impl Backend) -> Arc<Server> {
        Arc::new(Server {
            backend: Box::new(backend),
            locks: Mutex::new(HashMap::new()),
        })
    }

    pub fn backend(&self) -> &dyn Backend {
        self.backend.as_ref()
    }

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

    /// Answer one protocol request for repository `name`. `method` is
    /// checked against the endpoint; the body is the raw CBOR request.
    pub async fn handle(
        self: &Arc<Self>,
        name: String,
        endpoint: Endpoint,
        method: &Method,
        headers: &HeaderMap,
        body: Bytes,
    ) -> Response {
        if *method != endpoint.method() {
            return Failure(
                StatusCode::METHOD_NOT_ALLOWED,
                format!("{} is {}", endpoint.name(), endpoint.method()),
            )
            .into_response();
        }
        if let Err(f) = self.backend.authorize(&name, endpoint.access(), headers) {
            return f.into_response();
        }
        let claimed = headers
            .get(oxy_core::protocol::REPO_ID_HEADER)
            .and_then(|v| v.to_str().ok())
            .and_then(hex_id);
        let server = Arc::clone(self);
        let result =
            tokio::task::spawn_blocking(move || server.respond(&name, endpoint, body, claimed))
                .await;
        match result {
            Ok(Ok(r)) => r,
            Ok(Err(f)) => f.into_response(),
            Err(e) => Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
        }
    }

    fn respond(
        &self,
        name: &str,
        endpoint: Endpoint,
        body: Bytes,
        claimed: Option<[u8; 16]>,
    ) -> Result<Response, Failure> {
        let r = self.backend.open(name)?;
        Ok(match endpoint {
            Endpoint::Capabilities => ok(wire::encode_capabilities(r.id()?)),
            Endpoint::Refs => ok(wire::encode_refs(r.id()?, &r.remote_refs()?)),
            Endpoint::Negotiate => {
                let (have, want) = wire::decode_negotiate(&body)?;
                ok(wire::encode_hashes(&r.negotiate(&have, &want)?))
            }
            Endpoint::Fetch => {
                let (changes, max_blob) = wire::decode_fetch(&body)?;
                ok(wire::encode_bundle(
                    &r.bundle_for_filtered(&changes, max_blob)?,
                ))
            }
            Endpoint::Objects => {
                let refs = wire::decode_hashes(&body)?;
                let bundle = oxy_core::protocol::Bundle {
                    objects: r.objects_for(&refs)?,
                    ..Default::default()
                };
                ok(wire::encode_bundle(&bundle))
            }
            Endpoint::Checkpoint => {
                let heads = wire::decode_hashes(&body)?;
                let lock = self.lock_for(name);
                let _guard = lock.lock().unwrap();
                ok(wire::encode_bundle(&r.checkpoint_bundle_for(&heads)?))
            }
            Endpoint::Shallow => {
                let (heads, depth) = wire::decode_shallow(&body)?;
                ok(wire::encode_hashes(&r.shallow_changes(&heads, depth)?))
            }
            Endpoint::Push => {
                let (bundle, updates, policy, tags) = wire::decode_push(&body)?;
                let lock = self.lock_for(name);
                let _guard = lock.lock().unwrap();
                if r.is_unbound() {
                    // The first push binds the repository to the pusher's id (spec §16.1).
                    match claimed {
                        Some(id) => r.bind_id(id)?,
                        None => {
                            return Err(Failure(
                                StatusCode::BAD_REQUEST,
                                format!(
                                    "this repository is unbound; push with a client that sends {}",
                                    oxy_core::protocol::REPO_ID_HEADER
                                ),
                            ))
                        }
                    }
                }
                let results = r.receive_push(bundle, updates, policy, tags)?;
                self.backend.after_push(name, &results);
                ok(wire::encode_push_response(&results))
            }
        })
    }
}

/// The backend of `oxy serve`: one repository per subdirectory of a root,
/// open reads, and pushes gated by a list of bearer tokens.
pub struct DirBackend {
    pub root: PathBuf,
    pub tokens: Vec<String>,
}

impl Backend for DirBackend {
    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.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 authorize(&self, _name: &str, access: Access, headers: &HeaderMap) -> Result<(), Failure> {
        if access == Access::Read || self.tokens.is_empty() {
            return Ok(());
        }
        match bearer_token(headers) {
            Some(t) if self.tokens.iter().any(|x| x == t) => Ok(()),
            _ => Err(Failure(
                StatusCode::UNAUTHORIZED,
                "a valid bearer token is required to push".into(),
            )),
        }
    }
}

fn hex_id(h: &str) -> Option<[u8; 16]> {
    let h = h.trim();
    if h.len() != 32 {
        return None;
    }
    let mut out = [0u8; 16];
    for (i, byte) in out.iter_mut().enumerate() {
        *byte = u8::from_str_radix(&h[2 * i..2 * i + 2], 16).ok()?;
    }
    Some(out)
}

/// The token of an `Authorization: Bearer` header, if any.
pub fn bearer_token(headers: &HeaderMap) -> Option<&str> {
    headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(str::trim)
}

async fn dispatch(
    State(server): State<Arc<Server>>,
    Path((repo, endpoint)): Path<(String, String)>,
    method: Method,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    match Endpoint::parse(&endpoint) {
        Some(e) => server.handle(repo, e, &method, &headers, body).await,
        None => Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response(),
    }
}

/// The router of `oxy serve`: `/{repo}/oxy/v1/{endpoint}` over a [`DirBackend`].
pub fn router(cfg: ServerConfig) -> Router {
    let server = Server::new(DirBackend {
        root: cfg.root,
        tokens: cfg.tokens,
    });
    Router::new()
        .route("/{repo}/oxy/v1/{endpoint}", any(dispatch))
        .with_state(server)
}

/// 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))
}