//! 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), the states each accepted reference moved

    /// between, and the changes stored, for logs and cache invalidation.

    fn after_push(&self, _name: &str, _report: &oxy_core::repo::PushReport, _headers: &HeaderMap) {}

    /// Add the host's transparency log to a refs answer (spec §14.8): its

    /// checkpoint, and each attestation's entry and inclusion proof. A host

    /// without a log leaves it as it is.

    fn add_log(&self, _name: &str, _refs: &mut oxy_core::protocol::RefsResponse) {}

    /// The bookmark of `name` that pushes may never delete (its default),

    /// when the host keeps one apart from the repository's own config.

    fn default_bookmark(&self, _name: &str) -> Option<String> {

        None

    }

}

/// 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 {

        use oxy_core::repo::RepoError;

        // The server's own storage failing is not the client's fault.

        let status = match &e {

            RepoError::Io(_) | RepoError::Store(_) => StatusCode::INTERNAL_SERVER_ERROR,

            _ => StatusCode::BAD_REQUEST,

        };

        Failure(status, 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 request_headers = headers.clone();

        let repo = name.clone();

        let result = tokio::task::spawn_blocking(move || {

            server.respond(&repo, endpoint, body, claimed, &request_headers)

        })

        .await;

        let failure = match result {

            Ok(Ok(r)) => return r,

            Ok(Err(f)) => f,

            Err(e) => Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()),

        };

        // Name the operation and repository, in the log and for the client.

        let f = Failure(

            failure.0,

            format!("{} of {name}: {}", endpoint.name(), failure.1),

        );

        eprintln!("oxy-server: {}: {}", f.0.as_u16(), f.1);

        f.into_response()

    }

    fn respond(

        &self,

        name: &str,

        endpoint: Endpoint,

        body: Bytes,

        claimed: Option<[u8; 16]>,

        headers: &HeaderMap,

    ) -> Result<Response, Failure> {

        let r = self.backend.open(name)?;

        Ok(match endpoint {

            Endpoint::Capabilities => ok(wire::encode_capabilities(r.id()?)),

            Endpoint::Refs => {

                let mut refs = r.remote_refs()?;

                if let Some(d) = self.backend.default_bookmark(name) {

                    refs.default_bookmark = Some(d);

                }

                self.backend.add_log(name, &mut refs);

                ok(wire::encode_refs(r.id()?, &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 protected = self.backend.default_bookmark(name);

                let (refused, updates): (Vec<_>, Vec<_>) = updates.into_iter().partition(|u| {

                    u.new_state.is_zero() && protected.as_deref() == Some(u.name.as_str())

                });

                let mut report = r.receive_push_report(bundle, updates, policy, tags)?;

                // The request named references, all refused here: it was no

                // push of objects alone, whatever the repository saw.

                if !refused.is_empty() {

                    report

                        .results

                        .retain(|r| r.name != oxy_core::repo::OBJECTS_RESULT);

                }

                for u in refused {

                    report.results.push(oxy_core::protocol::RefResult {

                        name: u.name.clone(),

                        ok: false,

                        reason: format!(

                            "{:?} is this repository's default bookmark and cannot be deleted",

                            u.name

                        ),

                    });

                }

                for r in report.results.iter().filter(|r| !r.ok) {

                    eprintln!(

                        "oxy-server: push to {name}: {} refused: {}",

                        r.name, r.reason

                    );

                }

                self.backend.after_push(name, &report, headers);

                ok(wire::encode_push_response(&report.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 default_bookmark(&self, name: &str) -> Option<String> {

        self.open(name).ok().map(|r| r.config.bookmark.clone())

    }

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

}