alkhttp/src/server/adapter.rs

//! `HttpAdapter` — `ProtocolHandler` for `h2`/`http/1.1` (axum over a
//! `BiStream`).
//!
//! Wires the axum `Router` (gateway endpoints + `/healthz` +
//! `/openapi.json` + MCP + custom routes + decoy fallback) and drives
//! hyper's HTTP/1.1 or HTTP/2 connection driver over a single
//! bidirectional stream yielded by `Connection::accept_bi()`. The WS
//! upgrade route lands with the websocket subsystem; until then the
//! router reserves `/alk/channels` for it.

use std::sync::Arc;

use alkcall::core::auth::AuthContext;
use alkcall::core::types::{Connection, HandlerError, StreamError};
use alkcall::registry::registration::OperationRegistry;
use async_trait::async_trait;
use axum::routing::get;
use axum::Router;
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as HyperBuilder;
use hyper_util::service::TowerToHyperService;
use tracing::error;

use super::auth::bearer_auth_middleware;
use super::decoy::decoy_fallback;
use super::healthz::healthz;
use super::state::{DecoyConfig, RouterState};

pub const ALPN_HTTP1: &[u8] = b"http/1.1";
pub const ALPN_H2: &[u8] = b"h2";

/// The WS upgrade path (ADR-067). Reserved in the default surface; the
/// handler is wired by the websocket subsystem task.
pub const WS_UPGRADE_PATH: &str = "/alk/channels";

/// Reserved default-surface paths (ADR-046 collision rule). Custom
/// routes must not collide with these; the default surface wins.
pub const RESERVED_PATHS: &[&str] = &[
    "/search",
    "/schema",
    "/call",
    "/batch",
    "/subscribe",
    "/publish",
    "/healthz",
    "/openapi.json",
    "/mcp",
    WS_UPGRADE_PATH,
];

pub struct HttpAdapter {
    identity_provider: Arc<dyn alkcall::core::auth::IdentityProvider>,
    registry: Arc<OperationRegistry>,
    decoy: DecoyConfig,
    extra_routes: Option<Router>,
    alpn: &'static [u8],
    router: Router,
}

impl HttpAdapter {
    pub fn new(
        identity_provider: Arc<dyn alkcall::core::auth::IdentityProvider>,
        registry: Arc<OperationRegistry>,
    ) -> Self {
        Self::for_alpn(identity_provider, registry, ALPN_HTTP1)
    }

    pub fn h2(
        identity_provider: Arc<dyn alkcall::core::auth::IdentityProvider>,
        registry: Arc<OperationRegistry>,
    ) -> Self {
        Self::for_alpn(identity_provider, registry, ALPN_H2)
    }

    fn for_alpn(
        identity_provider: Arc<dyn alkcall::core::auth::IdentityProvider>,
        registry: Arc<OperationRegistry>,
        alpn: &'static [u8],
    ) -> Self {
        let decoy = DecoyConfig::default();
        let state = RouterState {
            registry: Arc::clone(&registry),
            identity_provider: Arc::clone(&identity_provider),
            decoy: decoy.clone(),
        };
        let router = build_router(state, None);
        Self {
            identity_provider,
            registry,
            decoy,
            extra_routes: None,
            alpn,
            router,
        }
    }

    pub fn with_decoy(mut self, decoy: DecoyConfig) -> Self {
        self.decoy = decoy.clone();
        let state = RouterState {
            registry: Arc::clone(&self.registry),
            identity_provider: Arc::clone(&self.identity_provider),
            decoy,
        };
        self.router = build_router(state, self.extra_routes.take());
        self
    }

    pub fn with_extra_routes(mut self, routes: Router) -> Self {
        let state = RouterState {
            registry: Arc::clone(&self.registry),
            identity_provider: Arc::clone(&self.identity_provider),
            decoy: self.decoy.clone(),
        };
        self.router = build_router(state, Some(routes.clone()));
        self.extra_routes = Some(routes);
        self
    }

    pub fn decoy(&self) -> &DecoyConfig {
        &self.decoy
    }

    pub fn alpn(&self) -> &'static [u8] {
        self.alpn
    }

    pub fn router(&self) -> &Router {
        &self.router
    }
}

fn build_router(state: RouterState, extra_routes: Option<Router>) -> Router {
    let auth_state = Arc::clone(&state.identity_provider);

    let default: Router<RouterState> = Router::new()
        .merge(crate::gateway::routes::gateway_router())
        .route("/healthz", get(healthz))
        .route_layer(from_fn_with_state(
            auth_state.clone(),
            bearer_auth_middleware,
        ))
        .fallback(decoy_fallback);

    let with_extras = match extra_routes {
        Some(extra) => {
            let extra: Router<RouterState> = extra.with_state(());
            default.merge(extra)
        }
        None => default,
    };

    with_extras.with_state(state)
}

use axum::middleware::from_fn_with_state;

#[async_trait]
impl alkcall::core::types::ProtocolHandler for HttpAdapter {
    fn alpn(&self) -> &'static [u8] {
        self.alpn
    }

    async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> {
        if let Some(identity) = auth.identity.clone() {
            let _ = connection.set_identity(identity);
        }

        let stream = connection
            .accept_bi()
            .await
            .map_err(stream_error_to_handler)?;
        self.serve_io(stream).await
    }
}

impl HttpAdapter {
    async fn serve_io<I>(&self, io: I) -> Result<(), HandlerError>
    where
        I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
    {
        let io = TokioIo::new(io);
        let service = TowerToHyperService::new(self.router.clone());

        #[cfg_attr(not(feature = "h2"), allow(unused_mut))]
        let mut builder = HyperBuilder::new(TokioExecutor::new());
        #[cfg(feature = "h2")]
        {
            builder.http2().enable_connect_protocol();
        }

        let conn = builder.serve_connection_with_upgrades(io, service);
        tokio::pin!(conn);

        let result = (&mut conn).await;
        if let Err(e) = result {
            error!("http adapter: connection closed with error: {e}");
        }
        Ok(())
    }
}

fn stream_error_to_handler(e: StreamError) -> HandlerError {
    HandlerError::from(e)
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkcall::core::auth::IdentityProvider;
    use alkcall::core::types::ProtocolHandler;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    struct NoopProvider;
    impl IdentityProvider for NoopProvider {
        fn resolve_from_fingerprint(&self, _: &str) -> Option<alkcall::core::auth::Identity> {
            None
        }
        fn resolve_from_token(
            &self,
            _: &alkcall::core::auth::AuthToken,
        ) -> Option<alkcall::core::auth::Identity> {
            None
        }
    }

    fn empty_registry() -> Arc<OperationRegistry> {
        Arc::new(OperationRegistry::new())
    }

    fn provider() -> Arc<dyn IdentityProvider> {
        Arc::new(NoopProvider)
    }

    #[test]
    fn alpn_returns_http1_for_default_new() {
        let adapter = HttpAdapter::new(provider(), empty_registry());
        assert_eq!(adapter.alpn(), ALPN_HTTP1);
        assert_eq!(adapter.alpn(), b"http/1.1");
    }

    #[test]
    fn alpn_returns_h2_for_h2_constructor() {
        let adapter = HttpAdapter::h2(provider(), empty_registry());
        assert_eq!(adapter.alpn(), ALPN_H2);
        assert_eq!(adapter.alpn(), b"h2");
    }

    #[test]
    fn decoy_config_default_is_not_found() {
        assert!(matches!(DecoyConfig::default(), DecoyConfig::NotFound));
    }

    #[test]
    fn with_decoy_updates_decoy() {
        let adapter = HttpAdapter::new(provider(), empty_registry());
        let adapter = adapter.with_decoy(DecoyConfig::Redirect {
            to: "https://example.com".to_string(),
        });
        assert!(matches!(adapter.decoy(), DecoyConfig::Redirect { .. }));
    }

    #[tokio::test]
    async fn full_http_request_response_cycle_over_duplex() {
        let extra = Router::new().route("/v1/ping", get(|| async { "pong" }));
        let adapter = HttpAdapter::new(provider(), empty_registry()).with_extra_routes(extra);

        let (client, server) = tokio::io::duplex(64 * 1024);
        let conn = Connection::from_bidi(server, b"http/1.1".to_vec(), None);
        let auth = AuthContext::anonymous(b"http/1.1");

        let server_task =
            tokio::spawn(async move { ProtocolHandler::handle(&adapter, conn, &auth).await });

        let mut client = client;
        client
            .write_all(b"GET /v1/ping HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
            .await
            .unwrap();

        let mut response = Vec::new();
        let _ = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.read_to_end(&mut response),
        )
        .await
        .expect("read timed out")
        .unwrap();

        let text = String::from_utf8_lossy(&response);
        assert!(text.starts_with("HTTP/1.1 200 OK"), "got: {text}");
        assert!(text.contains("pong"), "got: {text}");

        let _ = server_task.await;
    }

    #[tokio::test]
    async fn healthz_served_by_the_adapter_over_duplex() {
        let adapter = HttpAdapter::new(provider(), empty_registry());
        let (client, server) = tokio::io::duplex(64 * 1024);
        let conn = Connection::from_bidi(server, b"http/1.1".to_vec(), None);
        let auth = AuthContext::anonymous(b"http/1.1");
        let server_task = tokio::spawn(async move {
            let _ = ProtocolHandler::handle(&adapter, conn, &auth).await;
        });

        let mut client = client;
        client
            .write_all(b"GET /healthz HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
            .await
            .unwrap();

        let mut response = Vec::new();
        let _ = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.read_to_end(&mut response),
        )
        .await
        .expect("read timed out")
        .unwrap();

        let text = String::from_utf8_lossy(&response);
        assert!(text.starts_with("HTTP/1.1 200 OK"), "got: {text}");
        assert!(text.contains("ok"), "got: {text}");

        let _ = server_task.await;
    }

    #[tokio::test]
    async fn unknown_path_serves_decoy_404_over_duplex() {
        let adapter = HttpAdapter::new(provider(), empty_registry());
        let (client, server) = tokio::io::duplex(64 * 1024);
        let conn = Connection::from_bidi(server, b"http/1.1".to_vec(), None);
        let auth = AuthContext::anonymous(b"http/1.1");
        let server_task = tokio::spawn(async move {
            let _ = ProtocolHandler::handle(&adapter, conn, &auth).await;
        });

        let mut client = client;
        client
            .write_all(b"GET /nowhere HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
            .await
            .unwrap();

        let mut response = Vec::new();
        let _ = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.read_to_end(&mut response),
        )
        .await
        .expect("read timed out")
        .unwrap();

        let text = String::from_utf8_lossy(&response);
        assert!(text.starts_with("HTTP/1.1 404 Not Found"), "got: {text}");
        assert!(
            text.contains("nginx"),
            "decoy should look like nginx: {text}"
        );

        let _ = server_task.await;
    }
}