Move newtwork code under new http module

examples/timing_attack_check.rs

@@ -1,7 +1,7 @@
 use dudect_bencher::{BenchRng, Class, CtRunner, ctbench_main};
 
 use commit_bridge::domain::NonEmptyString;
-use commit_bridge::verify_api_key;
+use commit_bridge::http::router::verify_api_key;
 use dudect_bencher::rand::RngExt;
 use rand::distr::{Alphanumeric, SampleString};
 

src/http.rs

@@ -0,0 +1,20 @@
+//! HTTP networking: router wiring, server runtime, and the outbound HTTP client.
+
+pub mod handler;
+pub mod router;
+pub(crate) mod server;
+pub mod state;
+
+use reqwest::Client;
+
+use crate::{config::Config, error::ClientCreationError};
+
+/// Creates a new HTTP client.
+pub(crate) fn build_http_client(config: &Config) -> Result<Client, ClientCreationError> {
+    let client = Client::builder()
+        .user_agent(config.server.user_agent.to_string())
+        .timeout(config.server.out_request_timeout)
+        .build()?;
+
+    Ok(client)
+}

src/handler.rs → src/http/handler.rs

@@ -56,8 +56,8 @@ mod tests {
     use super::update::update_subscription_inner;
     use crate::domain::{BranchName, EventType, RepoUrl, TargetRepo};
     use crate::error::HandlerError;
+    use crate::http::state::AppState;
     use crate::model::{CreateSubscription, UpdateSubscription};
-    use crate::state::AppState;
     use crate::test_utils::create_test_db;
     use axum::Json;
     use axum::extract::{Path, Query, State};

src/handler/create.rs → src/http/handler/create.rs

@@ -2,9 +2,9 @@
 
 use super::map_to_hal;
 use crate::error::HandlerError;
+use crate::http::state::AppState;
 use crate::model::{CreateSubscription, SubscriptionHal};
 use crate::repository::subscription::SubscriptionRepository;
-use crate::state::AppState;
 use axum::{Json, extract::State};
 use rovo::rovo;
 use tracing::{info, instrument};

src/handler/delete.rs → src/http/handler/delete.rs

@@ -1,8 +1,8 @@
 //! Delete a subscription handler.
 
 use crate::error::HandlerError;
+use crate::http::state::AppState;
 use crate::repository::subscription::SubscriptionRepository;
-use crate::state::AppState;
 use axum::extract::{Path, State};
 use rovo::rovo;
 use tracing::instrument;

src/handler/get.rs → src/http/handler/get.rs

@@ -2,9 +2,9 @@
 
 use super::map_to_hal;
 use crate::error::HandlerError;
+use crate::http::state::AppState;
 use crate::model::SubscriptionHal;
 use crate::repository::subscription::SubscriptionRepository;
-use crate::state::AppState;
 use axum::{
     Json,
     extract::{Path, State},

src/handler/list.rs → src/http/handler/list.rs

@@ -2,9 +2,9 @@
 
 use super::map_to_hal;
 use crate::error::HandlerError;
+use crate::http::state::AppState;
 use crate::model::{HalLink, SubscriptionHal, SubscriptionPage, SubscriptionPageLinks};
 use crate::repository::subscription::SubscriptionRepository;
-use crate::state::AppState;
 use axum::{
     Json,
     extract::{Query, State},

src/handler/update.rs → src/http/handler/update.rs

@@ -2,9 +2,9 @@
 
 use super::map_to_hal;
 use crate::error::HandlerError;
+use crate::http::state::AppState;
 use crate::model::{SubscriptionHal, UpdateSubscription};
 use crate::repository::subscription::SubscriptionRepository;
-use crate::state::AppState;
 use axum::{
     Json,
     extract::{Path, State},

src/http/router.rs

@@ -0,0 +1,223 @@
+//! Axum router wiring and middlewares.
+
+use std::time::Duration;
+
+use axum::{
+    Router,
+    body::Body,
+    extract::{MatchedPath, State},
+    http::{HeaderValue, Request, Response, StatusCode, header},
+    middleware::{self, Next},
+    response::IntoResponse,
+};
+use rovo::Router as RovoRouter;
+use rovo::aide::openapi::OpenApi;
+use rovo::rovo;
+use subtle::ConstantTimeEq;
+use tower_http::timeout::TimeoutLayer;
+use tower_http::trace::{MakeSpan, OnResponse, TraceLayer};
+use tracing::Span;
+
+use crate::{
+    config::Config,
+    domain::NonEmptyString,
+    http::handler::{
+        create_subscription, delete_subscription, get_subscription, list_subscriptions,
+        update_subscription,
+    },
+    http::state::AppState,
+};
+
+/// Middleware to authorize requests with an API key.
+///
+/// Records the `authenticated` attribute onto the `http.request` span
+/// (the span is current while this middleware runs,
+/// since it is layered below the `TraceLayer`).
+async fn auth_middleware(
+    State(state): State<AppState>,
+    req: Request<Body>,
+    next: Next,
+) -> Response<Body> {
+    let needs_authentication =
+        req.uri().path().starts_with("/subscriptions") && !state.config.auth.allow_unauthenticated;
+
+    if needs_authentication {
+        let auth_header = req.headers().get("X-API-KEY").and_then(|v| v.to_str().ok());
+
+        if !verify_api_key(state.config.auth.api_key.as_ref(), auth_header) {
+            tracing::Span::current().record("authenticated", false);
+            return StatusCode::UNAUTHORIZED.into_response();
+        }
+
+        tracing::Span::current().record("authenticated", true);
+    }
+
+    next.run(req).await
+}
+
+/// Uses constant-time verification to check API key correspondence.
+///
+/// If the API keys correspond, returns `true`,
+/// otherwise `false`.
+///
+/// ### Cryptographic security
+///
+/// The function short-circuits if lengths of `expected` and `provided` are unequal.
+/// While this allows an attacker to extract the key length,
+/// it is order of magnitudes safer than using a simple string equality test,
+/// which would allow the attacker to gradually know the exact key
+/// over many requests.
+pub fn verify_api_key(expected: Option<&NonEmptyString>, provided: Option<&str>) -> bool {
+    expected.zip(provided).is_some_and(|(key, header)| {
+        let key_bytes = key.as_bytes();
+        let header_bytes = header.as_bytes();
+        key_bytes.ct_eq(header_bytes).into()
+    })
+}
+
+/// Middleware to set Cache-Control header.
+async fn set_no_cache_header(req: Request<Body>, next: Next) -> Response<Body> {
+    let path = req.uri().path().to_string();
+    let mut response = next.run(req).await;
+    if path.starts_with("/subscriptions") {
+        response.headers_mut().insert(
+            header::CACHE_CONTROL,
+            HeaderValue::from_static("private, no-cache, no-store, must-revalidate, max-age=0"),
+        );
+    }
+    response
+}
+
+/// Records the matched route path (`http.route`)
+/// onto the current HTTP request span.
+///
+/// Runs after routing, so the [`MatchedPath`] extension is available.
+async fn record_http_route(req: Request<Body>, next: Next) -> Response<Body> {
+    if let Some(matched) = req.extensions().get::<MatchedPath>() {
+        tracing::Span::current().record("http.route", matched.as_str());
+    }
+    next.run(req).await
+}
+
+#[allow(missing_docs, clippy::missing_docs_in_private_items)]
+mod health_handler {
+    use super::*;
+    #[rovo]
+    #[tracing::instrument(skip_all, fields(otel.kind = "internal"))]
+    pub async fn health_check(State(_state): State<AppState>) -> &'static str {
+        "CommitBridge is alive"
+    }
+}
+
+/// Span factory for incoming HTTP requests,
+/// following OpenTelemetry semantic conventions.
+///
+/// The span is created within this crate
+/// (instead of using the default `tower_http` span factory)
+/// so that its attributes follow OpenTelemetry conventions.
+#[derive(Clone, Copy)]
+struct HttpRequestSpan;
+
+impl<B> MakeSpan<B> for HttpRequestSpan {
+    fn make_span(&mut self, request: &Request<B>) -> Span {
+        tracing::info_span!(
+            "http.request",
+            otel.kind = "server",
+            http.request.method = %request.method(),
+            url.path = %request.uri().path(),
+            http.route = tracing::field::Empty,
+            http.response.status_code = tracing::field::Empty,
+            otel.status_code = tracing::field::Empty,
+            error.type = tracing::field::Empty,
+            authenticated = tracing::field::Empty,
+        )
+    }
+}
+
+/// Records HTTP response metadata onto the request span,
+/// following OpenTelemetry semantic conventions.
+///
+/// Must be used with [`HttpRequestSpan`],
+/// which declares the fields recorded here.
+#[derive(Clone, Copy)]
+pub(crate) struct HttpRequestOnResponse {
+    /// Whether client error responses (4xx)
+    /// should be marked as errors in exported traces.
+    mark_client_errors: bool,
+}
+
+impl HttpRequestOnResponse {
+    /// Creates a new [`HttpRequestOnResponse`].
+    pub(crate) const fn new(mark_client_errors: bool) -> Self {
+        Self { mark_client_errors }
+    }
+
+    /// Returns `true` if the response status should be marked as an error
+    /// in exported traces.
+    ///
+    /// Server errors (5xx) are always marked;
+    /// client errors (4xx) are only marked when `mark_client_errors` is set.
+    pub(crate) fn should_mark_error(&self, status: StatusCode) -> bool {
+        status.is_server_error() || (self.mark_client_errors && status.is_client_error())
+    }
+}
+
+impl<B> OnResponse<B> for HttpRequestOnResponse {
+    fn on_response(self, response: &Response<B>, _latency: Duration, span: &Span) {
+        span.record("http.response.status_code", response.status().as_u16());
+        if self.should_mark_error(response.status()) {
+            span.record("otel.status_code", "ERROR");
+            span.record("error.type", response.status().as_u16().to_string());
+        }
+    }
+}
+
+/// Builds the application router.
+pub fn build_router(
+    repository: std::sync::Arc<crate::repository::SqliteRepository>,
+    config: &Config,
+) -> Router {
+    let state = AppState {
+        config: std::sync::Arc::new(config.clone()),
+        repository,
+    };
+
+    let mut api = OpenApi::default();
+    api.info.title = "CommitBridge API".to_string();
+    api.info.description =
+        Some("API for managing repository subscriptions and triggering workflows".to_string());
+
+    let subscriptions = RovoRouter::<AppState>::new()
+        .route(
+            "/",
+            rovo::routing::post(create_subscription).get(list_subscriptions),
+        )
+        .route(
+            "/{id}",
+            rovo::routing::get(get_subscription)
+                .patch(update_subscription)
+                .delete(delete_subscription),
+        );
+
+    RovoRouter::<AppState>::new()
+        .route("/health", rovo::routing::get(health_handler::health_check))
+        .nest("/subscriptions", subscriptions)
+        .with_oas(api)
+        .with_scalar("/scalar")
+        .with_state(state.clone())
+        .finish()
+        .layer(middleware::from_fn_with_state(state, auth_middleware))
+        .layer(middleware::from_fn(set_no_cache_header))
+        .layer(middleware::from_fn(record_http_route))
+        .layer(TimeoutLayer::with_status_code(
+            StatusCode::REQUEST_TIMEOUT,
+            config.server.in_request_timeout,
+        ))
+        .layer(
+            TraceLayer::new_for_http()
+                .make_span_with(HttpRequestSpan)
+                .on_response(HttpRequestOnResponse::new(
+                    config.telemetry.mark_client_errors_as_error,
+                )),
+        )
+}

src/http/server.rs

@@ -0,0 +1,55 @@
+//! HTTP server runtime and lifecycle.
+
+use axum::Router;
+use tokio::signal;
+use tokio_util::sync::CancellationToken;
+use tracing::info;
+
+use crate::{config::Config, error::FatalError};
+
+/// Runs the server.
+pub(crate) async fn run_server(
+    app: Router,
+    config: &Config,
+    token: CancellationToken,
+) -> Result<(), FatalError> {
+    let listener = tokio::net::TcpListener::bind(config.server.address)
+        .await
+        .map_err(FatalError::TcpBinding)?;
+    println!("Server listening on http://{}", config.server.address);
+    println!(
+        "Scalar UI available at http://{}/scalar",
+        config.server.address
+    );
+
+    axum::serve(listener, app)
+        .with_graceful_shutdown(shutdown_signal(token))
+        .await
+        .map_err(FatalError::Serve)?;
+
+    Ok(())
+}
+
+/// Creates a future that resolves when a termination signal is received.
+async fn shutdown_signal(token: CancellationToken) {
+    let ctrl_c = signal::ctrl_c();
+
+    #[cfg(unix)]
+    let terminate = async {
+        if let Ok(mut signal) = signal::unix::signal(signal::unix::SignalKind::terminate()) {
+            signal.recv().await;
+        } else {
+            std::future::pending::<()>().await;
+        }
+    };
+
+    #[cfg(not(unix))]
+    let terminate = std::future::pending::<()>();
+
+    tokio::select! {
+        _ = ctrl_c => {},
+        _ = terminate => {},
+        _ = token.cancelled() => {},
+    }
+    info!("Shutdown signal received, initiating graceful shutdown...");
+}

src/state.rs → src/http/state.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 /// Holds data accessible from each [handler].
 ///
 /// <!-- LINKS -->
-/// [handler]: crate::handler
+/// [handler]: crate::http::handler
 #[derive(Debug, Clone)]
 pub struct AppState {
     /// Application configuration.