From 79dd9de3d4c24f9c50ed12214e6bebaa65d89518 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 03:54:06 -0700 Subject: [PATCH] simplify message handler types --- signal-gateway-bin/src/admin_netcat/mod.rs | 4 +- signal-gateway/src/gateway/mod.rs | 57 +++++--------- signal-gateway/src/lib.rs | 4 +- signal-gateway/src/message_handler.rs | 88 ++++++++++++++++++++++ 4 files changed, 112 insertions(+), 41 deletions(-) create mode 100644 signal-gateway/src/message_handler.rs diff --git a/signal-gateway-bin/src/admin_netcat/mod.rs b/signal-gateway-bin/src/admin_netcat/mod.rs index 808c026..14597f1 100644 --- a/signal-gateway-bin/src/admin_netcat/mod.rs +++ b/signal-gateway-bin/src/admin_netcat/mod.rs @@ -4,7 +4,7 @@ //! writing the message terminated with CRLF, and reading the response until CRLF. use conf::Conf; -use signal_gateway::MessageHandlerResult; +use signal_gateway::{AdminMessageResponse, MessageHandlerResult}; use std::time::Duration; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, @@ -74,5 +74,5 @@ async fn handle_message(config: &AdminNetcatConfig, message: String) -> MessageH .trim_end_matches(['\r', '\n']) .to_owned(); - Ok((text, vec![])) + Ok(AdminMessageResponse::new(text)) } diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 61e38a0..edc5338 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -2,6 +2,7 @@ use crate::{ alertmanager::AlertPost, jsonrpc::{Envelope, RpcClient, RpcClientError, SignalMessage, connect_tcp}, log_message::{LogMessage, Origin}, + message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult}, prometheus::{Prometheus, PrometheusConfig}, }; use chrono::Utc; @@ -11,10 +12,7 @@ use http::{Method, Request, Response, StatusCode}; use http_body::Body; use http_body_util::BodyExt; use prometheus_http_client::{AlertStatus, ExtractLabels}; -use std::{ - collections::HashMap, error::Error, fmt::Write, future::Future, net::SocketAddr, - path::PathBuf, pin::Pin, -}; +use std::{collections::HashMap, fmt::Write, net::SocketAddr, path::PathBuf}; use tokio::{ join, sync::{ @@ -27,15 +25,6 @@ use tokio_util::bytes::Buf; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; -/// Response from a message handler: either success with text and optional attachments, -/// or an error with status code and message. -pub type MessageHandlerResult = Result<(String, Vec), (u16, Box)>; - -/// Handler function for admin messages that don't start with `/`. -/// Takes the message text and returns a response. -pub type MessageHandler = - Box Pin + Send>> + Send + Sync>; - mod circular_buffer; mod log_handler; use log_handler::{LogHandler, LogHandlerConfig}; @@ -291,20 +280,18 @@ impl Gateway { }) ); - let (message, attachments) = resp.unwrap_or_else( - |(code, msg)| { - let text = format!("{code}: {msg}"); - error!("Message handler error: {text}"); - (text, vec![]) - } - ); + let resp = resp.unwrap_or_else(|(code, msg)| { + let text = format!("{code}: {msg}"); + error!("Message handler error: {text}"); + AdminMessageResponse::new(text) + }); - let attachments = attachments.into_iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect(); + let attachments = resp.attachments.into_iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect(); SignalMessage { sender: self.config.signal_account.clone(), recipient: vec![msg.envelope.source_uuid.clone()], - message, + message: resp.text, attachments, }.send(signal_cli).await?; } @@ -316,10 +303,7 @@ impl Gateway { // Returns Err in case of a timeout or handler error // Returns Ok when success or error text is generated - async fn handle_signal_admin_message( - &self, - msg: &Envelope, - ) -> Result<(String, Vec), (u16, Box)> { + async fn handle_signal_admin_message(&self, msg: &Envelope) -> MessageHandlerResult { let data = msg.data_message.as_ref().unwrap(); // Admin messages starting with / are handled by gateway @@ -398,15 +382,12 @@ impl Gateway { } } - async fn handle_gateway_command( - &self, - cmd: GatewayCommand, - ) -> Result<(String, Vec), (u16, Box)> { + async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult { match cmd { GatewayCommand::Log { filter } => { let handlers = self.log_handlers.read().await; if handlers.is_empty() { - return Ok(("No log sources registered yet".to_string(), vec![])); + return Ok(AdminMessageResponse::new("No log sources registered yet")); } let mut text = String::new(); for (origin, handler) in handlers.iter() { @@ -421,9 +402,9 @@ impl Gateway { text.push('\n'); } if text.is_empty() { - return Ok(("No matching log sources".to_string(), vec![])); + return Ok(AdminMessageResponse::new("No matching log sources")); } - Ok((text, vec![])) + Ok(AdminMessageResponse::new(text)) } GatewayCommand::Query { query } => { let prometheus = self @@ -455,7 +436,7 @@ impl Gateway { } } - Ok((text, vec![])) + Ok(AdminMessageResponse::new(text)) } Err(err) => Err((500, err)), } @@ -469,7 +450,7 @@ impl Gateway { prometheus.purge_old_plots(); match prometheus.create_oneoff_plot(query.clone(), duration).await { - Ok(filename) => Ok((query, vec![filename])), + Ok(filename) => Ok(AdminMessageResponse::new(query).with_attachment(filename)), Err(err) => Err((500, err)), } } @@ -493,7 +474,7 @@ impl Gateway { if text.is_empty() { text = "no matches".into(); } - Ok((text, vec![])) + Ok(AdminMessageResponse::new(text)) } Err(err) => Err((500, err)), } @@ -514,7 +495,7 @@ impl Gateway { if text.is_empty() { text = "no matches".into(); } - Ok((text, vec![])) + Ok(AdminMessageResponse::new(text)) } Err(err) => Err((500, err)), } @@ -554,7 +535,7 @@ impl Gateway { if text.is_empty() { text = "no alerts".into(); } - Ok((text, vec![])) + Ok(AdminMessageResponse::new(text)) } Err(err) => Err((500, err)), } diff --git a/signal-gateway/src/lib.rs b/signal-gateway/src/lib.rs index 3ca6281..30cd186 100644 --- a/signal-gateway/src/lib.rs +++ b/signal-gateway/src/lib.rs @@ -1,5 +1,6 @@ pub mod alertmanager; pub mod gateway; +pub mod message_handler; pub(crate) mod human_duration; pub(crate) mod jsonrpc; @@ -7,5 +8,6 @@ pub(crate) mod log_message; pub(crate) mod prometheus; pub(crate) mod transports; -pub use gateway::{Gateway, GatewayConfig, MessageHandler, MessageHandlerResult}; +pub use gateway::{Gateway, GatewayConfig}; pub use log_message::{Level, LogMessage, LogMessageBuilder}; +pub use message_handler::{AdminMessageResponse, MessageHandler, MessageHandlerResult}; diff --git a/signal-gateway/src/message_handler.rs b/signal-gateway/src/message_handler.rs new file mode 100644 index 0000000..65d281b --- /dev/null +++ b/signal-gateway/src/message_handler.rs @@ -0,0 +1,88 @@ +//! Message handler types for admin messages not handled by the gateway. + +use std::{ + error::Error, + future::Future, + path::PathBuf, + pin::Pin, +}; + +/// Response to an admin message. +#[non_exhaustive] +#[derive(Clone, Debug, Default)] +pub struct AdminMessageResponse { + /// The text response to send back to the admin. + pub text: String, + /// Optional file attachments to include with the response. + pub attachments: Vec, +} + +impl AdminMessageResponse { + /// Create a new response with the given text. + pub fn new(text: impl Into) -> Self { + Self { + text: text.into(), + attachments: Vec::new(), + } + } + + /// Create a builder for constructing a response. + pub fn builder() -> AdminMessageResponseBuilder { + AdminMessageResponseBuilder::default() + } + + /// Add an attachment to the response. + pub fn with_attachment(mut self, path: impl Into) -> Self { + self.attachments.push(path.into()); + self + } + + /// Add multiple attachments to the response. + pub fn with_attachments(mut self, paths: impl IntoIterator>) -> Self { + self.attachments.extend(paths.into_iter().map(Into::into)); + self + } +} + +/// Builder for constructing an [`AdminMessageResponse`]. +#[derive(Clone, Debug, Default)] +pub struct AdminMessageResponseBuilder { + text: String, + attachments: Vec, +} + +impl AdminMessageResponseBuilder { + /// Set the text response. + pub fn text(mut self, text: impl Into) -> Self { + self.text = text.into(); + self + } + + /// Add an attachment. + pub fn attachment(mut self, path: impl Into) -> Self { + self.attachments.push(path.into()); + self + } + + /// Add multiple attachments. + pub fn attachments(mut self, paths: impl IntoIterator>) -> Self { + self.attachments.extend(paths.into_iter().map(Into::into)); + self + } + + /// Build the response. + pub fn build(self) -> AdminMessageResponse { + AdminMessageResponse { + text: self.text, + attachments: self.attachments, + } + } +} + +/// Result type for message handler responses. +pub type MessageHandlerResult = Result)>; + +/// Handler function for admin messages that don't start with `/`. +/// Takes the message text and returns a response. +pub type MessageHandler = + Box Pin + Send>> + Send + Sync>;