lavina/src/http.rs

use std::convert::Infallible;
use std::net::SocketAddr;

use futures_util::FutureExt;
use http_body_util::{BodyExt, Full};
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response, StatusCode};
use hyper_util::rt::TokioIo;
use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;

use lavina_core::auth::UpdatePasswordResult;
use lavina_core::player::{PlayerId, SendMessageResult};
use lavina_core::prelude::*;
use lavina_core::room::RoomId;
use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore;
use mgmt_api::*;

mod clustering;

type HttpResult<T> = std::result::Result<T, Infallible>;

#[derive(Deserialize, Debug)]
pub struct ServerConfig {
    pub listen_on: SocketAddr,
}

pub async fn launch(config: ServerConfig, metrics: MetricsRegistry, core: LavinaCore) -> Result<Terminator> {
    log::info!("Starting the http service");
    let listener = TcpListener::bind(config.listen_on).await?;
    log::debug!("Listener started");
    let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, core, rx.map(|_| ())));
    Ok(terminator)
}

async fn main_loop(
    listener: TcpListener,
    metrics: MetricsRegistry,
    core: LavinaCore,
    termination: impl Future<Output = ()>,
) -> Result<()> {
    pin!(termination);
    loop {
        select! {
            biased;
            _ = &mut termination => break,
            result = listener.accept() => {
                let (stream, _) = result?;
                let stream = TokioIo::new(stream);
                let metrics = metrics.clone();
                let core = core.clone();
                tokio::task::spawn(async move {
                    let svc_fn = service_fn(|r| route(&metrics, &core, r));
                    let server = http1::Builder::new().serve_connection(stream, svc_fn);
                    if let Err(err) = server.await {
                        tracing::error!("Error serving connection: {:?}", err);
                    }
                });
            },
        }
    }
    log::info!("Terminating the http service");
    Ok(())
}

#[tracing::instrument(skip_all)]
async fn route(
    registry: &MetricsRegistry,
    core: &LavinaCore,
    request: Request<hyper::body::Incoming>,
) -> HttpResult<Response<Full<Bytes>>> {
    propagade_span_from_headers(&request);
    let res = match (request.method(), request.uri().path()) {
        (&Method::GET, "/metrics") => endpoint_metrics(registry),
        (&Method::GET, "/rooms") => endpoint_rooms(core).await,
        (&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, core).await.or5xx(),
        (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core).await.or5xx(),
        (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, core).await.or5xx(),
        (&Method::POST, rooms::paths::SEND_MESSAGE) => endpoint_send_room_message(request, core).await.or5xx(),
        (&Method::POST, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(),
        _ => clustering::route(core, request).await.unwrap_or_else(endpoint_not_found),
    };
    Ok(res)
}

fn endpoint_metrics(registry: &MetricsRegistry) -> Response<Full<Bytes>> {
    let mf = registry.gather();
    let mut buffer = vec![];
    TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail");
    Response::new(Full::new(Bytes::from(buffer)))
}

#[tracing::instrument(skip_all)]
async fn endpoint_rooms(core: &LavinaCore) -> Response<Full<Bytes>> {
    // TODO introduce management API types independent from core-domain types
    // TODO remove `Serialize` implementations from all core-domain types
    let room_list = core.get_all_rooms().await.to_body();
    Response::new(room_list)
}

#[tracing::instrument(skip_all)]
async fn endpoint_create_player(
    request: Request<hyper::body::Incoming>,
    core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> {
    let str = request.collect().await?.to_bytes();
    let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
        return Ok(malformed_request());
    };
    core.create_player(&PlayerId::from(res.name)?).await?;
    log::info!("Player {} created", res.name);
    let mut response = Response::new(Full::<Bytes>::default());
    *response.status_mut() = StatusCode::CREATED;
    Ok(response)
}

#[tracing::instrument(skip_all)]
async fn endpoint_stop_player(
    request: Request<hyper::body::Incoming>,
    core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> {
    let str = request.collect().await?.to_bytes();
    let Ok(res) = serde_json::from_slice::<StopPlayerRequest>(&str[..]) else {
        return Ok(malformed_request());
    };
    let Ok(player_id) = PlayerId::from(res.name) else {
        return Ok(player_not_found());
    };
    let Some(()) = core.stop_player(&player_id).await? else {
        return Ok(player_not_found());
    };
    Ok(empty_204_request())
}

#[tracing::instrument(skip_all)]
async fn endpoint_set_password(
    request: Request<hyper::body::Incoming>,
    core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> {
    let str = request.collect().await?.to_bytes();
    let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
        return Ok(malformed_request());
    };
    let verdict = core.set_password(&res.player_name, &res.password).await?;
    match verdict {
        UpdatePasswordResult::PasswordUpdated => {}
        UpdatePasswordResult::UserNotFound => {
            return Ok(player_not_found());
        }
    }
    Ok(empty_204_request())
}

#[tracing::instrument(skip_all)]
async fn endpoint_send_room_message(
    request: Request<hyper::body::Incoming>,
    core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> {
    let str = request.collect().await?.to_bytes();
    let Ok(req) = serde_json::from_slice::<rooms::SendMessageReq>(&str[..]) else {
        return Ok(malformed_request());
    };
    let Ok(room_id) = RoomId::from(req.room_id) else {
        return Ok(room_not_found());
    };
    let Ok(player_id) = PlayerId::from(req.author_id) else {
        return Ok(player_not_found());
    };
    let mut player = core.connect_to_player(&player_id).await;
    let res = player.send_message(room_id, req.message.into()).await?;
    match res {
        SendMessageResult::NoSuchRoom => Ok(room_not_found()),
        SendMessageResult::Success(_) => Ok(empty_204_request()),
    }
}

#[tracing::instrument(skip_all)]
async fn endpoint_set_room_topic(
    request: Request<hyper::body::Incoming>,
    core: &LavinaCore,
) -> Result<Response<Full<Bytes>>> {
    let str = request.collect().await?.to_bytes();
    let Ok(req) = serde_json::from_slice::<rooms::SetTopicReq>(&str[..]) else {
        return Ok(malformed_request());
    };
    let Ok(room_id) = RoomId::from(req.room_id) else {
        return Ok(room_not_found());
    };
    let Ok(player_id) = PlayerId::from(req.author_id) else {
        return Ok(player_not_found());
    };
    let mut player = core.connect_to_player(&player_id).await;
    player.change_topic(room_id, req.topic.into()).await?;
    Ok(empty_204_request())
}

fn endpoint_not_found() -> Response<Full<Bytes>> {
    let payload = ErrorResponse {
        code: errors::INVALID_PATH,
        message: "The path does not exist",
    }
    .to_body();

    let mut response = Response::new(payload);
    *response.status_mut() = StatusCode::NOT_FOUND;
    response
}

fn player_not_found() -> Response<Full<Bytes>> {
    let payload = ErrorResponse {
        code: errors::PLAYER_NOT_FOUND,
        message: "No such player exists",
    }
    .to_body();
    let mut response = Response::new(payload);
    *response.status_mut() = StatusCode::UNPROCESSABLE_ENTITY;
    response
}

fn room_not_found() -> Response<Full<Bytes>> {
    let payload = ErrorResponse {
        code: rooms::errors::ROOM_NOT_FOUND,
        message: "No such room exists",
    }
    .to_body();
    let mut response = Response::new(payload);
    *response.status_mut() = StatusCode::UNPROCESSABLE_ENTITY;
    response
}

fn malformed_request() -> Response<Full<Bytes>> {
    let payload = ErrorResponse {
        code: errors::MALFORMED_REQUEST,
        message: "The request payload contains incorrect JSON value",
    }
    .to_body();

    let mut response = Response::new(payload);
    *response.status_mut() = StatusCode::BAD_REQUEST;
    return response;
}

fn empty_204_request() -> Response<Full<Bytes>> {
    let mut response = Response::new(Full::<Bytes>::default());
    *response.status_mut() = StatusCode::NO_CONTENT;
    response
}

trait Or5xx {
    fn or5xx(self) -> Response<Full<Bytes>>;
}

impl Or5xx for Result<Response<Full<Bytes>>> {
    fn or5xx(self) -> Response<Full<Bytes>> {
        self.unwrap_or_else(|e| {
            let mut response = Response::new(Full::new(e.to_string().into()));
            *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
            response
        })
    }
}

trait ToBody {
    fn to_body(&self) -> Full<Bytes>;
}

impl<T> ToBody for T
where
    T: Serialize,
{
    fn to_body(&self) -> Full<Bytes> {
        let mut buffer = vec![];
        serde_json::to_writer(&mut buffer, self).expect("unexpected fail when writing to vec");
        Full::new(Bytes::from(buffer))
    }
}

fn propagade_span_from_headers<T>(req: &Request<T>) {
    use opentelemetry::propagation::Extractor;
    use tracing::Span;
    use tracing_opentelemetry::OpenTelemetrySpanExt;
    struct HttpReqExtractor<'a, T> {
        req: &'a Request<T>,
    }
    impl<'a, T> Extractor for HttpReqExtractor<'a, T> {
        fn get(&self, key: &str) -> Option<&str> {
            self.req.headers().get(key).and_then(|v| v.to_str().ok())
        }

        fn keys(&self) -> Vec<&str> {
            self.req.headers().keys().map(|k| k.as_str()).collect()
        }
    }

    let ctx = opentelemetry::global::get_text_map_propagator(|pp| pp.extract(&HttpReqExtractor { req }));
    Span::current().set_parent(ctx);
}