move http body guts out to signal-gateway-bin

this is a cleaner separation of concerns

Cargo.lock

@@ -1724,9 +1724,6 @@ dependencies = [
  "conf-extra",
  "displaydoc",
  "futures-util",
- "http",
- "http-body",
- "http-body-util",
  "humantime",
  "jsonrpsee",
  "prometheus-http-client",
@@ -1797,6 +1794,9 @@ dependencies = [
  "conf",
  "conf-extra",
  "dotenvy",
+ "http",
+ "http-body",
+ "http-body-util",
  "hyper",
  "hyper-util",
  "reqwest",

signal-gateway-bin/Cargo.toml

@@ -25,6 +25,9 @@ async-trait = { workspace = true }
 conf = { workspace = true }
 conf-extra = { workspace = true }
 dotenvy = { workspace = true }
+http = { workspace = true }
+http-body = { workspace = true }
+http-body-util = { workspace = true }
 hyper = { workspace = true }
 hyper-util = { workspace = true }
 reqwest = { workspace = true }

signal-gateway-bin/src/main.rs

@@ -3,14 +3,19 @@
 #![deny(missing_docs)]
 
 use conf::Conf;
+use http::{Method, Request, Response, StatusCode};
+use http_body::Body;
+use http_body_util::BodyExt;
 use hyper::service::service_fn;
 use hyper_util::{rt::TokioIo, server::conn::auto};
-use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling};
+use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling, alertmanager::AlertPost};
 use signal_gateway_app_code::AppCodeTools;
 use signal_gateway_assistant_claude::{ClaudeAssistant, ClaudeConfig};
-use std::{env, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
+use std::{
+    convert::Infallible, env, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration,
+};
 use tokio::net::TcpListener;
-use tokio_util::sync::CancellationToken;
+use tokio_util::{bytes::Buf, sync::CancellationToken};
 use tracing::{error, info, warn};
 use tracing_subscriber::EnvFilter;
 
@@ -222,10 +227,7 @@ fn start_http_task(listener: TcpListener, gateway: Arc<Gateway>) -> tokio::task:
                 if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new())
                     .serve_connection(
                         io,
-                        service_fn(|req| {
+                        service_fn(|req| handle_http_request(thread_gateway.clone(), req)),
-                            let thread_gateway = thread_gateway.clone();
-                            async move { thread_gateway.handle_http_request(req).await }
-                        }),
                     )
                     .await
                 {
@@ -236,6 +238,96 @@ fn start_http_task(listener: TcpListener, gateway: Arc<Gateway>) -> tokio::task:
     })
 }
 
+async fn handle_http_request(
+    gateway: Arc<Gateway>,
+    req: Request<hyper::body::Incoming>,
+) -> Result<Response<String>, Infallible> {
+    match handle_http_request_impl(gateway, req).await {
+        Ok(resp) => Ok(resp),
+        Err(resp) => Ok(resp),
+    }
+}
+
+async fn handle_http_request_impl<B>(
+    gateway: Arc<Gateway>,
+    req: Request<B>,
+) -> Result<Response<String>, Response<String>>
+where
+    B: Body + Send,
+    B::Data: Buf + Send,
+    B::Error: std::fmt::Display,
+{
+    info!(
+        "Received http request: {} {} (version: {:?})",
+        req.method(),
+        req.uri().path(),
+        req.version()
+    );
+
+    fn ok_resp() -> Response<String> {
+        Response::new("OK".into())
+    }
+    fn err_resp(code: StatusCode, text: impl Into<String>) -> Response<String> {
+        let mut resp = Response::new(text.into());
+        *resp.status_mut() = code;
+        resp
+    }
+
+    match req.uri().path() {
+        "/" | "/health" | "/ready" => {
+            if !matches!(req.method(), &Method::GET | &Method::HEAD) {
+                Ok(err_resp(
+                    StatusCode::NOT_IMPLEMENTED,
+                    "Use GET or HEAD with this route",
+                ))
+            } else {
+                Ok(ok_resp())
+            }
+        }
+        "/alert" => {
+            if !matches!(req.method(), &Method::POST) {
+                return Ok(err_resp(
+                    StatusCode::NOT_IMPLEMENTED,
+                    "Use POST with this route",
+                ));
+            }
+            let body_bytes = req
+                .into_body()
+                .collect()
+                .await
+                .map_err(|err| {
+                    err_resp(
+                        StatusCode::BAD_REQUEST,
+                        format!("When reading body bytes: {err}"),
+                    )
+                })?
+                .to_bytes()
+                .to_vec();
+
+            let body_text = str::from_utf8(&body_bytes).map_err(|err| {
+                warn!("When reading body bytes: {err}");
+                err_resp(StatusCode::BAD_REQUEST, "Request body was not utf-8")
+            })?;
+
+            let alert_msg: AlertPost = serde_json::from_str(body_text).map_err(|err| {
+                error!("Could not parse json: {err}:\n{body_text}");
+                err_resp(StatusCode::BAD_REQUEST, "Invalid Json")
+            })?;
+
+            if let Err(msg) = gateway.handle_alertmanager_post(alert_msg).await {
+                error!("gateway (handle_alertmanager_post): {msg}");
+                Ok(err_resp(StatusCode::INTERNAL_SERVER_ERROR, msg))
+            } else {
+                Ok(ok_resp())
+            }
+        }
+        _ => Ok(err_resp(
+            StatusCode::NOT_FOUND,
+            format!("Not found '{} {}'", req.method(), req.uri().path()),
+        )),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

signal-gateway/Cargo.toml

@@ -21,9 +21,6 @@ conf = { workspace = true }
 conf-extra = { workspace = true }
 displaydoc = { workspace = true }
 futures-util = { workspace = true }
-http = { workspace = true }
-http-body = { workspace = true }
-http-body-util = { workspace = true }
 humantime = { workspace = true }
 jsonrpsee = { workspace = true }
 regex = { workspace = true }

signal-gateway/src/assistant/mod.rs

@@ -56,7 +56,7 @@ impl AssistantAgent {
     /// Spawns a background worker task that processes requests serially.
     pub fn new(assistant: Box<dyn Assistant>, cancellation_token: CancellationToken) -> Self {
         let (input_tx, input_rx) = mpsc::channel(REQUEST_QUEUE_SIZE);
-        let (stop_tx, stop_rx) = mpsc::channel(REQUEST_QUEUE_SIZE);
+        let (stop_tx, stop_rx) = mpsc::channel(3);
 
         let worker = AssistantWorker::new(assistant, input_rx, stop_rx, cancellation_token.clone());
 

signal-gateway/src/gateway/mod.rs

@@ -16,9 +16,6 @@ use async_trait::async_trait;
 use chrono::Utc;
 use conf::{Conf, Subcommands};
 use futures_util::FutureExt;
-use http::{Method, Request, Response, StatusCode};
-use http_body::Body;
-use http_body_util::BodyExt;
 use prometheus_http_client::{AlertStatus, ExtractLabels};
 use std::{
     fmt::Write,
@@ -31,7 +28,7 @@ use tokio::{
     join,
     sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
 };
-use tokio_util::{bytes::Buf, sync::CancellationToken};
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
 
 mod signal_trust_set;
@@ -655,68 +652,6 @@ impl Gateway {
         }
     }
 
-    /// Handle an incoming HTTP request (e.g., webhooks from Alertmanager).
-    pub async fn handle_http_request<B>(&self, req: Request<B>) -> Result<Response<String>, String>
-    where
-        B: Body + Send,
-        B::Data: Buf + Send,
-        B::Error: std::fmt::Display,
-    {
-        info!(
-            "Received http request: {} {} (version: {:?})",
-            req.method(),
-            req.uri().path(),
-            req.version()
-        );
-
-        fn ok_resp() -> Response<String> {
-            Response::new("OK".into())
-        }
-        fn err_resp(code: StatusCode, text: impl Into<String>) -> Response<String> {
-            let mut resp = Response::new(text.into());
-            *resp.status_mut() = code;
-            resp
-        }
-
-        match req.uri().path() {
-            "/" | "/health" | "/ready" => {
-                if !matches!(req.method(), &Method::GET | &Method::HEAD) {
-                    Ok(err_resp(
-                        StatusCode::NOT_IMPLEMENTED,
-                        "Use GET or HEAD with this route",
-                    ))
-                } else {
-                    Ok(ok_resp())
-                }
-            }
-            "/alert" => {
-                if !matches!(req.method(), &Method::POST) {
-                    return Ok(err_resp(
-                        StatusCode::NOT_IMPLEMENTED,
-                        "Use POST with this route",
-                    ));
-                }
-                let v = req
-                    .into_body()
-                    .collect()
-                    .await
-                    .map_err(|err| format!("When reading body bytes: {err}"))?
-                    .to_bytes()
-                    .to_vec();
-
-                if let Err((code, msg)) = self.handle_post_alert(&v).await {
-                    Ok(err_resp(code, msg))
-                } else {
-                    Ok(ok_resp())
-                }
-            }
-            _ => Ok(err_resp(
-                StatusCode::NOT_FOUND,
-                format!("Not found '{} {}'", req.method(), req.uri().path()),
-            )),
-        }
-    }
-
     async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult {
         match cmd {
             GatewayCommand::Log { filter } => {
@@ -887,17 +822,8 @@ impl Gateway {
         }
     }
 
-    async fn handle_post_alert(&self, body_bytes: &[u8]) -> Result<(), (StatusCode, &'static str)> {
+    /// Handle a POST body from alertmanager
-        let body_text = str::from_utf8(body_bytes).map_err(|err| {
+    pub async fn handle_alertmanager_post(&self, alert_msg: AlertPost) -> Result<(), &'static str> {
-            warn!("When reading body bytes: {err}");
-            (StatusCode::BAD_REQUEST, "Request body was not utf-8")
-        })?;
-
-        let alert_msg: AlertPost = serde_json::from_str(body_text).map_err(|err| {
-            error!("Could not parse json: {err}:\n{body_text}");
-            (StatusCode::BAD_REQUEST, "Invalid Json")
-        })?;
-
         let text = self
             .format_alert_text(&alert_msg)
             .unwrap_or_else(|err| format!("error formatting alert text: {err}:\n{alert_msg:#?}"));
@@ -921,20 +847,16 @@ impl Gateway {
         }
 
         // Build summary: status sigils followed by alert names
-        let summary = alert_msg
+        let mut summary = String::with_capacity(32);
-            .alerts
+        alert_msg.alerts.iter().for_each(|alert| {
-            .iter()
+            let symbol = alert.status.symbol();
-            .map(|alert| {
+            let name = alert
-                let symbol = alert.status.symbol();
+                .labels
-                let name = alert
+                .get("alertname")
-                    .labels
+                .map(|s| s.as_str())
-                    .get("alertname")
+                .unwrap_or("?");
-                    .map(|s| s.as_str())
+            write!(&mut summary, "{symbol}{name} ").unwrap();
-                    .unwrap_or("?");
+        });
-                format!("{symbol}{name}")
-            })
-            .collect::<Vec<_>>()
-            .join(" ");
 
         self.signal_alert_mq_tx
             .send(SignalAlertMessage {
@@ -944,13 +866,7 @@ impl Gateway {
                 summary: Summary::Owned(summary.into()),
                 destination_override: None,
             })
-            .map_err(|_err| {
+            .map_err(|_err| "Can't send signal msg right now, queue is closed")
-                error!("Could not send alert message, queue is closed");
-                (
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                    "Can't send signal msg right now, queue is closed",
-                )
-            })
     }
 
     fn format_alert_text(&self, msg: &AlertPost) -> Result<String, String> {