unb-server 2.0.3

use std::net::SocketAddr;
use std::sync::Arc;

use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use unb::{handler, Handler, HandlerError, Reply, Request, SendExt, Streaming};
use unb_server::{ErrorCode, Node};
use unb_server::{HostConfig, TcpTransport};

#[derive(Deserialize, Serialize, JsonSchema)]
#[serde(transparent)]
struct EchoPayload(Value);

#[derive(Deserialize, JsonSchema)]
struct Probe {}

#[handler]
async fn echo(request: Request<EchoPayload>) -> Result<Reply<Value>, HandlerError> {
    let player = request
        .headers()
        .get("x-player")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("nobody")
        .to_string();
    Ok(Reply::new(
        json!({ "echo": request.into_payload(), "player": player }),
    ))
}

#[handler]
async fn ticks(_request: Request<Probe>) -> Result<Streaming<Value, HandlerError>, HandlerError> {
    Ok(Streaming::new(futures_util::stream::iter([Ok(
        json!({"n": 1}),
    )])))
}

fn app_node() -> Arc<Node> {
    Node::builder("app-1")
        .service(echo.at_subject("app.echo"))
        .service(ticks.at_subject("app.watch"))
        .insecure_accept_declared_peer_identities()
        .build()
        .unwrap()
}

async fn host(node: &Arc<Node>) -> SocketAddr {
    let hosting = HostConfig::tcp(([127, 0, 0, 1], 0), TcpTransport::plain())
        .start(node)
        .await
        .unwrap();
    let addr = hosting.websocket_addr().unwrap();
    std::mem::forget(hosting);
    addr
}

#[tokio::test(flavor = "multi_thread")]
async fn builder_send_to_a_joined_node_round_trips() {
    let node = app_node();
    let response = http::Request::builder()
        .uri("/app-1/app.echo")
        .header("x-player", "alice")
        .body(json!({"from": "e2"}))
        .send(&node)
        .await
        .unwrap();
    assert_eq!(response.status(), http::StatusCode::OK);
    let body: Value = serde_json::from_slice(response.body()).unwrap();
    assert_eq!(body["echo"]["from"], "e2");
    assert_eq!(body["player"], "alice");
}

#[tokio::test(flavor = "multi_thread")]
async fn builder_send_to_a_server_address_reaches_ingress() {
    let node = app_node();
    let addr = host(&node).await;
    for destination in [format!("http://{addr}"), format!("{addr}")] {
        let response = http::Request::builder()
            .uri("/app-1/app.echo")
            .body(json!({"via": "one-shot"}))
            .send(&destination)
            .await
            .unwrap();
        assert_eq!(response.status(), http::StatusCode::OK, "{destination}");
        let body: Value = serde_json::from_slice(response.body()).unwrap();
        assert_eq!(body["echo"]["via"], "one-shot");
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn send_accepts_dotted_or_segmented_local_subject_form() {
    let node = app_node();
    for uri in ["/app-1/app.echo", "/app-1/app/echo"] {
        let response = http::Request::builder()
            .uri(uri)
            .body(json!({}))
            .send(&node)
            .await
            .unwrap();
        assert_eq!(response.status(), http::StatusCode::OK, "{uri}");
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn send_result_chaining_propagates_the_builder_error() {
    let node = app_node();
    let error = http::Request::builder()
        .uri("http://exa mple.com/bad uri")
        .body(json!({}))
        .send(&node)
        .await
        .unwrap_err();
    assert_eq!(error.code, ErrorCode::InvalidInput);
}

#[tokio::test(flavor = "multi_thread")]
async fn a_stream_subject_over_send_is_an_error() {
    let node = app_node();
    let error = http::Request::builder()
        .uri("/app-1/app.watch")
        .body(json!({}))
        .send(&node)
        .await
        .unwrap_err();
    assert_eq!(error.code, ErrorCode::Protocol);
}

#[tokio::test(flavor = "multi_thread")]
async fn an_unknown_subject_over_send_teaches() {
    let node = app_node();
    let error = http::Request::builder()
        .uri("/app-1/app.ech")
        .body(json!({}))
        .send(&node)
        .await
        .unwrap_err();
    assert_eq!(error.code, ErrorCode::UnknownSubject);
    assert!(error.message.contains("app.echo"), "{}", error.message);
}