//! Reference HTTP server for the Oxygen wire protocol (spec §15.1, §16).

//!

//! Serves every repository directory under a root at `/{repo}/oxy/v1/...`.

//! TLS is expected to be terminated in front of this server for now; the

//! protocol's integrity does not depend on the channel (spec §15).

//!

//! 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>,

}

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 = wire::decode_hashes(&body)?;

        Ok(ok(wire::encode_bundle(&r.bundle_for(&changes)?)))

    })

    .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) = 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)?)))

    })

    .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/push", post(push))

        .with_state(app)

}

/// Serve on an already-bound listener until the process ends.

pub async fn serve(listener: std::net::TcpListener, cfg: ServerConfig) -> std::io::Result<()> {

    listener.set_nonblocking(true)?;

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

}