initial commit

signal-gateway/Cargo.toml

@@ -0,0 +1,37 @@
+[package]
+name = "signal-gateway"
+version = "0.1.0"
+edition.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+prom-client = { workspace = true }
+
+chrono = { workspace = true }
+circular-buffer = { workspace = true }
+conf = { workspace = true }
+conf-extra = { workspace = true }
+displaydoc = { workspace = true }
+dotenvy = { workspace = true }
+futures-util = { workspace = true }
+http-body-util = { workspace = true }
+humantime = { workspace = true }
+hyper = { workspace = true }
+hyper-util = { workspace = true }
+jsonrpsee = { workspace = true }
+rand = { workspace = true }
+reqwest = { workspace = true }
+rust_decimal = { workspace = true }
+rustls = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+syslog_rfc5424 = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tokio-util = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+url = { workspace = true }
+walkdir = { workspace = true }

signal-gateway/src/config.rs

@@ -0,0 +1,18 @@
+use crate::gateway::GatewayConfig;
+use conf::Conf;
+use std::net::SocketAddr;
+
+#[derive(Conf, Debug)]
+pub struct Config {
+    /// If true, just validate config and don't start
+    #[conf(long)]
+    pub dry_run: bool,
+    /// Socket to listen for HTTP requests (GET /health, POST /alert)
+    #[conf(long, env, default_value = "0.0.0.0:8000")]
+    pub http_listen_addr: SocketAddr,
+    /// Socket to listen for UDP messages, in syslog RFC 5424 format
+    #[conf(long, env, default_value = "0.0.0.0:5424")]
+    pub udp_listen_addr: SocketAddr,
+    #[conf(flatten)]
+    pub gateway: GatewayConfig,
+}

signal-gateway/src/gateway/log_handler.rs

@@ -0,0 +1,387 @@
+use super::{AdminMessage, MultiRateLimiter, Origin, RateThreshold, SourceLocationRateLimiter};
+use crate::human_duration::HumanTMinus;
+use chrono::{TimeDelta, Utc};
+use circular_buffer::CircularBuffer;
+use conf::Conf;
+use serde::Deserialize;
+use std::{fmt, time::Duration};
+use syslog_rfc5424::{SyslogMessage, SyslogSeverity};
+use tokio::sync::{Mutex, mpsc::UnboundedSender};
+use tracing::{error, info, warn};
+
+/// Reason why an alert was suppressed
+enum SuppressionReason {
+    /// Suppressed by a configured alert rule (with 0-based rule index)
+    Rule(usize),
+    /// Suppressed by the source-location rate limiter
+    SourceLocation { file: String, line: String },
+}
+
+impl fmt::Display for SuppressionReason {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            SuppressionReason::Rule(idx) => write!(f, "rule[{idx}]"),
+            SuppressionReason::SourceLocation { file, line } => {
+                write!(f, "source-location({file}:{line})")
+            }
+        }
+    }
+}
+
+/// Config options related to the log handler, and what log messages it chooses to alert on.
+#[derive(Clone, Conf, Debug)]
+pub struct LogHandlerConfig {
+    #[conf(long, env, value_parser = serde_json::from_str)]
+    pub alert_rate_limits: Vec<AlertRule>,
+    #[conf(long, env, default_value = "10m", value_parser = conf_extra::parse_duration)]
+    pub overall_alert_limit: Duration,
+    #[conf(long, env)]
+    pub format_module: bool,
+    #[conf(long, env)]
+    pub format_source_location: bool,
+    /// Structured data ID for tracing metadata (module, file, line) in syslog messages
+    #[conf(long, env, default_value = "tracing-meta@64700")]
+    pub sd_id: String,
+}
+
+/// Specifies both a rate limiting threshold, and criteria for the threshold to apply
+#[derive(Clone, Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct AlertRule {
+    #[serde(default)]
+    pub msg_contains: String,
+    #[serde(default)]
+    pub module_equals: String,
+    #[serde(default)]
+    pub file_equals: String,
+    #[serde(default)]
+    pub line_equals: String,
+    pub threshold: RateThreshold,
+}
+
+impl AlertRule {
+    /// Check if a syslog message passes the filter defined by this rule
+    fn eval_filter(&self, syslog_msg: &SyslogMessage, sd_id: &str) -> bool {
+        if !self.msg_contains.is_empty() && !syslog_msg.msg.contains(&self.msg_contains) {
+            return false;
+        }
+
+        if !self.module_equals.is_empty() {
+            match syslog_msg.sd.find_tuple(sd_id, "module") {
+                Some(module) if module == self.module_equals.as_str() => {}
+                _ => return false,
+            }
+        }
+
+        if !self.file_equals.is_empty() {
+            match syslog_msg.sd.find_tuple(sd_id, "file") {
+                Some(file) if file == self.file_equals.as_str() => {}
+                _ => return false,
+            }
+        }
+
+        if !self.line_equals.is_empty() {
+            match syslog_msg.sd.find_tuple(sd_id, "line") {
+                Some(line) if line == self.line_equals.as_str() => {}
+                _ => return false,
+            }
+        }
+
+        true
+    }
+}
+
+/// The log handler takes log messages from a single origin and decides what
+/// to do with them.
+///
+/// 1. Store them in a small circular buffer
+/// 2. If it is an error, and meets other criteria, trigger an alert,
+///    i.e. send a message to admins containing this log and other recent logs.
+/// 3. The maximum rate of alerts can also be configured.
+///
+/// Additionally, the log handler can format the buffer of recent logs into a string,
+/// if requested.
+///
+/// Each origin (app + host pair) gets its own LogHandler instance, managed by the Gateway.
+#[derive(Debug)]
+pub struct LogHandler {
+    config: LogHandlerConfig,
+    admin_mq_tx: UnboundedSender<AdminMessage>,
+    syslog_buffer: Mutex<CircularBuffer<64, SyslogMessage>>,
+    rate_limiters: Vec<(AlertRule, Mutex<MultiRateLimiter>)>,
+    /// Rate limiter keyed by source location (file:line), so different error locations
+    /// can alert independently without suppressing each other.
+    overall_limiter: Mutex<SourceLocationRateLimiter>,
+    /// True if any configured rule uses the module structured data field.
+    any_rule_uses_module: bool,
+    /// True if any configured rule uses the file structured data field.
+    any_rule_uses_file: bool,
+    /// True if any configured rule uses the line structured data field.
+    any_rule_uses_line: bool,
+}
+
+/// Maximum entries in the source-location rate limiter before triggering cleanup
+const OVERALL_LIMITER_MAX_ENTRIES: usize = 2000;
+
+impl LogHandler {
+    /// Initialize a new log handler
+    pub fn new(config: LogHandlerConfig, admin_mq_tx: UnboundedSender<AdminMessage>) -> Self {
+        let any_rule_uses_module = config
+            .alert_rate_limits
+            .iter()
+            .any(|r| !r.module_equals.is_empty());
+        let any_rule_uses_file = config
+            .alert_rate_limits
+            .iter()
+            .any(|r| !r.file_equals.is_empty());
+        let any_rule_uses_line = config
+            .alert_rate_limits
+            .iter()
+            .any(|r| !r.line_equals.is_empty());
+        let rate_limiters = config
+            .alert_rate_limits
+            .iter()
+            .map(|rule| {
+                (
+                    rule.clone(),
+                    Mutex::new(MultiRateLimiter::from(rule.threshold)),
+                )
+            })
+            .collect::<Vec<_>>();
+        let overall_limiter = Mutex::new(SourceLocationRateLimiter::new(
+            config.overall_alert_limit,
+            OVERALL_LIMITER_MAX_ENTRIES,
+        ));
+
+        Self {
+            config,
+            admin_mq_tx,
+            syslog_buffer: Default::default(),
+            rate_limiters,
+            overall_limiter,
+            any_rule_uses_module,
+            any_rule_uses_file,
+            any_rule_uses_line,
+        }
+    }
+
+    /// Format recent logs into a string
+    pub async fn format_logs(&self) -> String {
+        let lk = self.syslog_buffer.lock().await;
+
+        let mut text = format!("{} log messages (newest first):\n", lk.len());
+
+        // Calculate now once for consistent relative timestamps
+        let now = Utc::now().timestamp();
+
+        // Collect and reverse to show newest first
+        let messages: Vec<_> = lk.iter().collect();
+        for syslog_msg in messages.into_iter().rev() {
+            self.write_syslog_msg(&mut text, syslog_msg, now);
+        }
+        text
+    }
+
+    /// Consume a new syslog message from the given origin
+    pub async fn handle_syslog_message(&self, mut syslog_msg: SyslogMessage, origin: Origin) {
+        let suppression_reason = self.check_suppression(&mut syslog_msg).await;
+        if let Some(reason) = &suppression_reason
+            && syslog_msg.severity <= SyslogSeverity::SEV_ERR
+        {
+            let sev = Self::severity_to_str(syslog_msg.severity);
+            info!("Suppressed {sev} ({reason}):\n{}", syslog_msg.msg);
+        }
+
+        // Record this new message.
+        // Then, if we should alert now, also format the whole buffer to a string,
+        // and then release the lock.
+        let formatted_text = {
+            let mut lk = self.syslog_buffer.lock().await;
+            lk.push_back(syslog_msg);
+            if suppression_reason.is_some() {
+                return;
+            }
+
+            let mut text = String::default();
+
+            // Calculate now once for consistent relative timestamps
+            let now = Utc::now().timestamp();
+
+            // Iterate in reverse (newest first) without copying
+            for syslog_msg in lk.iter().rev() {
+                self.write_syslog_msg(&mut text, syslog_msg, now);
+            }
+            lk.clear();
+
+            text
+        };
+
+        if let Err(_err) = self.admin_mq_tx.send(AdminMessage {
+            origin: Some(origin),
+            text: formatted_text,
+            attachment_paths: Default::default(),
+            summary: None,
+        }) {
+            error!("Could not send alert message, queue is closed");
+        }
+    }
+
+    // Convert SyslogSeverity to our own all-caps string that fits in 5 chars
+    fn severity_to_str(severity: SyslogSeverity) -> &'static str {
+        match severity {
+            SyslogSeverity::SEV_EMERG => "EMERG",
+            SyslogSeverity::SEV_ALERT => "ALERT",
+            SyslogSeverity::SEV_CRIT => "CRIT",
+            SyslogSeverity::SEV_ERR => "ERROR",
+            SyslogSeverity::SEV_WARNING => "WARN",
+            SyslogSeverity::SEV_NOTICE => "NOTE",
+            SyslogSeverity::SEV_INFO => "INFO",
+            SyslogSeverity::SEV_DEBUG => "DEBUG",
+        }
+    }
+
+    // Format a syslog message into a Writer, followed by \n, and using any config options to do so
+    fn write_syslog_msg(
+        &self,
+        mut writer: impl std::fmt::Write,
+        syslog_msg: &SyslogMessage,
+        now: i64,
+    ) {
+        let sev = Self::severity_to_str(syslog_msg.severity);
+        let msg = &syslog_msg.msg;
+
+        // Format relative timestamp if available
+        let time_str = if let Some(ts) = syslog_msg.timestamp {
+            let diff_secs = now.saturating_sub(ts);
+            HumanTMinus(TimeDelta::seconds(diff_secs)).to_string()
+        } else {
+            "T-?".to_owned()
+        };
+
+        // Extract metadata from structured data if configured
+        let sd_id = &self.config.sd_id;
+        let mut metadata_parts = Vec::new();
+
+        if self.config.format_module
+            && let Some(module) = syslog_msg.sd.find_tuple(sd_id, "module")
+        {
+            metadata_parts.push(module.to_string());
+        }
+
+        if self.config.format_source_location {
+            let file_opt = syslog_msg.sd.find_tuple(sd_id, "file");
+            let line_opt = syslog_msg.sd.find_tuple(sd_id, "line");
+
+            let location = if let Some(file) = file_opt {
+                // Strip /home/{username}/ prefix if present
+                let trimmed_file = strip_prefix_and_one_slash(file, "/home/");
+                // Strip .cargo/registry/src/{hash}/ if present
+                let trimmed_file = strip_prefix_and_one_slash(trimmed_file, ".cargo/registry/src/");
+
+                if let Some(line) = line_opt {
+                    format!("{trimmed_file}:{line}")
+                } else {
+                    format!("{trimmed_file}:?")
+                }
+            } else {
+                // No file present, use "?" even if line is present
+                "?".to_owned()
+            };
+
+            metadata_parts.push(location);
+        }
+
+        // Format: "ERROR     T-10s [foo bar.rs:42]: message"
+        // Pad severity to 5 chars (left-aligned), time to 8 chars (right-aligned)
+        let result = if metadata_parts.is_empty() {
+            writeln!(writer, "{:<5} {:>8}: {}", sev, time_str, msg)
+        } else {
+            let metadata = metadata_parts.join(" ");
+            writeln!(writer, "{:<5} {:>8} [{}]: {}", sev, time_str, metadata, msg)
+        };
+
+        if let Err(err) = result {
+            error!("Couldn't write syslog message ({err}): {sev}: {msg}");
+        }
+    }
+
+    /// Check if an alert should be suppressed for a given error message.
+    ///
+    /// Returns `None` if the alert should fire, or `Some(reason)` if suppressed.
+    async fn check_suppression(&self, syslog_msg: &mut SyslogMessage) -> Option<SuppressionReason> {
+        let ts_sec = *syslog_msg
+            .timestamp
+            .get_or_insert_with(|| Utc::now().timestamp());
+
+        let high_severity = syslog_msg.severity <= SyslogSeverity::SEV_ERR;
+        if !high_severity {
+            // Low severity messages are always "suppressed" (not alerted on)
+            // but we don't need to log a reason for this
+            return Some(SuppressionReason::Rule(usize::MAX));
+        }
+
+        let sd_id = &self.config.sd_id;
+
+        // Warn if rules expect structured data but the message doesn't have it
+        let has_module = syslog_msg.sd.find_tuple(sd_id, "module").is_some();
+        let has_file = syslog_msg.sd.find_tuple(sd_id, "file").is_some();
+        let has_line = syslog_msg.sd.find_tuple(sd_id, "line").is_some();
+        if (self.any_rule_uses_module && !has_module)
+            || (self.any_rule_uses_file && !has_file)
+            || (self.any_rule_uses_line && !has_line)
+        {
+            warn!(
+                "Error message missing structured data (sd_id={sd_id}), filtering rules may not work: {syslog_msg:#?}"
+            );
+        }
+
+        // Check each configured rule - track which rule suppressed the alert
+        // Note: we check all rules even if one already suppressed, to update all rate limiters
+        let mut suppressed_by_rule: Option<usize> = None;
+        for (idx, (filter, limiter)) in self.rate_limiters.iter().enumerate() {
+            if filter.eval_filter(syslog_msg, sd_id) && !limiter.lock().await.evaluate(ts_sec) {
+                suppressed_by_rule.get_or_insert(idx);
+            }
+        }
+        if let Some(idx) = suppressed_by_rule {
+            return Some(SuppressionReason::Rule(idx));
+        }
+
+        // Extract source location for per-location rate limiting
+        let file = syslog_msg
+            .sd
+            .find_tuple(sd_id, "file")
+            .map_or("?", |s| s.as_str());
+        let line = syslog_msg
+            .sd
+            .find_tuple(sd_id, "line")
+            .map_or("?", |s| s.as_str());
+
+        if !self
+            .overall_limiter
+            .lock()
+            .await
+            .evaluate(file, line, ts_sec)
+        {
+            return Some(SuppressionReason::SourceLocation {
+                file: file.to_owned(),
+                line: line.to_owned(),
+            });
+        }
+
+        None // Alert should fire
+    }
+}
+
+// Strip a prefix, then find the first remaining slash and skip up to that as well.
+fn strip_prefix_and_one_slash<'a>(target: &'a str, prefix: &str) -> &'a str {
+    let Some(target) = target.strip_prefix(prefix) else {
+        return target;
+    };
+
+    if let Some((_, after)) = target.split_once('/') {
+        after
+    } else {
+        target
+    }
+}

signal-gateway/src/gateway/mod.rs

@@ -0,0 +1,857 @@
+use crate::{
+    http::{AlertMessage, Status},
+    jsonrpc::{Envelope, RpcClient, RpcClientError, SignalMessage, connect_tcp},
+    plotter::{Plotter, PlotterConfig},
+};
+use conf::{Conf, Subcommands};
+use futures_util::FutureExt;
+use http_body_util::BodyExt;
+use hyper::{Method, Request, Response, StatusCode, body::Incoming};
+use prom_client::{AlertStatus, ExtractLabels};
+//use jsonrpsee::async_client::{Client as JsonRpcClient, Error as JsonRpcError};
+use chrono::Utc;
+use std::{collections::HashMap, error::Error, fmt::Write, net::SocketAddr, time::Duration};
+use syslog_rfc5424::SyslogMessage;
+use tokio::{
+    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
+    join,
+    net::TcpStream,
+    sync::{
+        Mutex, RwLock,
+        mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
+    },
+    time::timeout,
+};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, warn};
+
+mod log_handler;
+use log_handler::{LogHandler, LogHandlerConfig};
+
+mod rate_limiter;
+use rate_limiter::{MultiRateLimiter, RateThreshold, SourceLocationRateLimiter};
+
+#[derive(Conf, Debug)]
+pub struct GatewayConfig {
+    #[conf(long, env, default_value = "127.0.0.1:7583")]
+    pub signal_cli_tcp_addr: SocketAddr,
+    #[conf(long, env)]
+    pub signal_account: String,
+    #[conf(long, env)]
+    pub cbmm_tcp_addr: String,
+    #[conf(long, env, default_value = "5s", value_parser = conf_extra::parse_duration)]
+    pub cbmm_timeout: Duration,
+    #[conf(repeat, long, env)]
+    pub admin_uuid: Vec<String>,
+    #[conf(flatten)]
+    pub plotter: Option<PlotterConfig>,
+    #[conf(flatten)]
+    pub log_handler: LogHandlerConfig,
+}
+
+/// Wrapper for parsing gateway commands
+#[derive(Clone, Debug, Conf)]
+struct GatewayCommandWrapper {
+    #[conf(subcommands)]
+    command: GatewayCommand,
+}
+
+/// Commands that can be sent to the gateway (prefixed with /)
+#[derive(Clone, Debug, Subcommands)]
+enum GatewayCommand {
+    /// Show recent log messages
+    #[conf(name = "log", alias = "LOG")]
+    Log {
+        /// Optional filter: show only origins where app or host contains this string
+        #[conf(pos)]
+        filter: Option<String>,
+    },
+    /// Query prometheus for current values
+    #[conf(name = "query", alias = "QUERY")]
+    Query {
+        /// PromQL query expression
+        #[conf(pos)]
+        query: String,
+    },
+    /// Plot a prometheus query over time
+    #[conf(name = "plot", alias = "PLOT")]
+    Plot {
+        /// PromQL query expression
+        #[conf(pos)]
+        query: String,
+        /// Duration to plot (e.g., 1h, 24h)
+        #[conf(long, short = 'd', default_value = "1h", value_parser = conf_extra::parse_duration)]
+        duration: Duration,
+    },
+    /// List series matching label patterns
+    #[conf(name = "series", alias = "SERIES")]
+    Series {
+        /// Label matchers (e.g., __name__=~".*requests.*")
+        #[conf(repeat, pos)]
+        matchers: Vec<String>,
+    },
+    /// List label names matching patterns
+    #[conf(name = "labels", alias = "LABELS")]
+    Labels {
+        /// Label matchers
+        #[conf(repeat, pos)]
+        matchers: Vec<String>,
+    },
+    /// Show current alerts from prometheus
+    #[conf(name = "alerts", alias = "ALERTS")]
+    Alerts,
+}
+
+/// Parse a gateway command from a string (with or without leading /)
+fn parse_gateway_command(s: &str) -> Result<GatewayCommand, String> {
+    // Remove leading slash if present
+    let s = s.strip_prefix('/').unwrap_or(s).trim();
+
+    if s.is_empty() {
+        return Err("Empty command".to_string());
+    }
+
+    // Parse using Conf, treating the input as command line arguments
+    // Prepend a dummy program name since Conf expects argv[0]
+    let args = std::iter::once("gateway")
+        .chain(s.split_whitespace())
+        .collect::<Vec<_>>();
+
+    GatewayCommandWrapper::try_parse_from::<&str, &str, &str>(args, vec![])
+        .map(|wrapper| wrapper.command)
+        .map_err(|e| e.to_string())
+}
+
+/// Identifies the source of log messages (app name + host).
+/// Used to separate log buffers and rate limiters per source.
+#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)]
+pub struct Origin {
+    pub app: String,
+    pub host: String,
+}
+
+impl std::fmt::Display for Origin {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}@{}", self.app, self.host)
+    }
+}
+
+impl From<&SyslogMessage> for Origin {
+    fn from(msg: &SyslogMessage) -> Self {
+        Self {
+            app: msg.appname.clone().unwrap_or_default(),
+            host: msg.hostname.clone().unwrap_or_default(),
+        }
+    }
+}
+
+impl Origin {
+    /// Check if this origin matches a filter string.
+    ///
+    /// If the filter contains '@', it is split on the first '@':
+    /// - The part before '@' must be a substring of `app`
+    /// - The part after '@' must be a substring of `host`
+    ///
+    /// If the filter does not contain '@', it matches if either `app` or `host`
+    /// contains the filter string.
+    pub fn matches_filter(&self, filter: &str) -> bool {
+        if let Some((app_filter, host_filter)) = filter.split_once('@') {
+            self.app.contains(app_filter) && self.host.contains(host_filter)
+        } else {
+            self.app.contains(filter) || self.host.contains(filter)
+        }
+    }
+}
+
+/// A message queued to be sent to all admins.
+/// This is generally an alert message, which may have attached images.
+#[derive(Clone, Debug, Default)]
+struct AdminMessage {
+    /// The origin of the message (app + host), if from syslog
+    origin: Option<Origin>,
+    text: String,
+    attachment_paths: Vec<String>,
+    /// Short summary for logging (e.g., alert names for prometheus).
+    /// If None, the consumer will use a truncated slice of `text` for logging.
+    summary: Option<String>,
+}
+
+/// The gateway manages sending messages to signal-cli and receiving messages from signal-cli.
+/// It maintains a queue of messages to be sent to all admins, generated by alerts etc.
+/// It also subscribes to messages received from signal and processes them one-by-one, possibly
+/// making TCP request to cbmm to handle them.
+///
+/// This is the only task that communicates directly with signal-cli, and by design it linearizes
+/// all interaction, which prevents any possible races.
+///
+/// The gateway also maintains a buffer of at most 64 syslog messages that it has received.
+/// If an error occurs, that message and the previous messages in the buffer are sent to signal admins,
+/// and then the buffer is purged. This serves as a minimal log-aggregation and alerting system.
+pub struct Gateway {
+    config: GatewayConfig,
+    admin_mq_tx: UnboundedSender<AdminMessage>,
+    admin_mq_rx: Mutex<UnboundedReceiver<AdminMessage>>,
+    token: CancellationToken,
+    plotter: Option<Plotter>,
+    /// Log handlers keyed by origin (app + host). Lazily created when first message from an origin arrives.
+    log_handlers: RwLock<HashMap<Origin, LogHandler>>,
+}
+
+impl Gateway {
+    pub async fn new(config: GatewayConfig, token: CancellationToken) -> Self {
+        let (admin_mq_tx, admin_mq_rx) = unbounded_channel();
+
+        let plotter = config
+            .plotter
+            .as_ref()
+            .map(|plotter_config| Plotter::new(plotter_config.clone()));
+
+        Self {
+            config,
+            admin_mq_tx,
+            admin_mq_rx: Mutex::new(admin_mq_rx),
+            token,
+            plotter,
+            log_handlers: RwLock::new(HashMap::new()),
+        }
+    }
+
+    pub async fn run(&self) {
+        loop {
+            if self.token.is_cancelled() {
+                return;
+            }
+            match connect_tcp(&self.config.signal_cli_tcp_addr).await {
+                Err(err) => {
+                    error!(
+                        "Could not connect to signal_cli @ ({}): {err}",
+                        self.config.signal_cli_tcp_addr
+                    );
+                    tokio::time::sleep(Duration::from_secs(5)).await;
+                }
+                Ok(client) => {
+                    if let Err(err) = self.do_run(&client).await {
+                        error!("Error with signal cli, reconnecting: {err}");
+                        tokio::time::sleep(Duration::from_secs(5)).await;
+                    } else {
+                        continue;
+                    }
+                }
+            }
+        }
+    }
+
+    async fn do_run(&self, signal_cli: &impl RpcClient) -> Result<(), RpcClientError> {
+        let mut admin_mq_rx = self
+            .admin_mq_rx
+            .try_lock()
+            .expect("Mutex should not be contended");
+        let mut signal_rx = signal_cli
+            .subscribe_receive(Some(self.config.signal_account.clone()))
+            .await?;
+
+        loop {
+            tokio::select! {
+                _ = self.token.cancelled() => {
+                    info!("Stop requested");
+                    return Ok(());
+                },
+                outbound_admin_msg = admin_mq_rx.recv() => {
+                    if let Some(msg) = outbound_admin_msg {
+                        // Log summary, or first 500 bytes of text if no summary provided
+                        let summary = msg.summary.as_deref().unwrap_or_else(|| {
+                            let len = msg.text.len().min(500);
+                            &msg.text[..len]
+                        });
+                        info!("Sending alert: {summary}");
+                        // Prepend origin line if present
+                        let message = if let Some(origin) = &msg.origin {
+                            format!("[{origin}]\n{}", msg.text)
+                        } else {
+                            msg.text
+                        };
+                        SignalMessage {
+                            sender: self.config.signal_account.clone(),
+                            recipient: self.config.admin_uuid.clone(),
+                            message,
+                            attachments: msg.attachment_paths,
+                        }.send(signal_cli).await?;
+                    } else {
+                        warn!("admin_mq_rx is closed, halting service");
+                        self.token.cancel();
+                        return Ok(());
+                    }
+                },
+                signal_msg = signal_rx.next() => {
+                    match signal_msg {
+                        None => {
+                            info!("Signal Rx: stream closed");
+                            return Ok(());
+                        },
+                        Some(Err(err)) => {
+                            error!("Signal Rx: {err}");
+                            return Err(RpcClientError::ParseError(err));
+                        }
+                        Some(Ok(msg)) => {
+                            //info!("Signal Rx: {msg:?}");
+                            if msg.envelope.data_message.is_none() {
+                                debug!("Ignoring message which was not a data message: {msg:?}");
+                                continue;
+                            }
+
+                            if !self.config.admin_uuid.contains(&msg.envelope.source_uuid) {
+                                warn!("Ignoring message from non-admin: {msg:?}");
+                                continue;
+                            }
+
+                            let (resp, _) = join!(self.handle_signal_admin_message(&msg.envelope),
+                                            msg.envelope.send_read_receipt(signal_cli, &self.config.signal_account).map(|result| {
+                                                if let Err(err) = result {
+                                                    warn!("Couldn't send read receipt: {err}");
+                                                }
+                                            }));
+                            let (message, attachments) = resp.unwrap_or_else(
+                                |(code, msg)| {
+                                    let text = format!("{code}: {msg}");
+                                    error!("(cbmm) {text}");
+                                    (text, vec![])
+                                }
+                            );
+
+                            SignalMessage {
+                                sender: self.config.signal_account.clone(),
+                                recipient: vec![msg.envelope.source_uuid.clone()],
+                                message,
+                                attachments,
+                            }.send(signal_cli).await?;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    // Returns Err in case of a timeout
+    // Returns Ok when success or error text is generated
+    async fn handle_signal_admin_message(
+        &self,
+        msg: &Envelope,
+    ) -> Result<(String, Vec<String>), (u16, Box<dyn Error>)> {
+        let data = msg.data_message.as_ref().unwrap();
+
+        // Admin messages starting with / are handled by gateway
+        // Other messages are forwarded to cbmm
+        if data.message.starts_with("/") {
+            // Parse the command using conf
+            let cmd = parse_gateway_command(&data.message).map_err(|err| (400, err.into()))?;
+
+            self.handle_gateway_command(cmd).await
+        } else {
+            // Connect to cbmm. Note that we could use a keep-alive strategy here maybe...
+            let mut cbmm_stream = timeout(
+                self.config.cbmm_timeout,
+                TcpStream::connect(&self.config.cbmm_tcp_addr),
+            )
+            .await
+            .map_err(format_err("connecting", 504))?
+            .map_err(format_err("connecting", 502))?;
+
+            timeout(
+                self.config.cbmm_timeout,
+                cbmm_stream.write_all(data.message.as_bytes()),
+            )
+            .await
+            .map_err(format_err("writing", 504))?
+            .map_err(format_err("writing", 502))?;
+
+            let _ = cbmm_stream.shutdown().await;
+
+            // Wrap as BufReader so that we can use "read_until" which simplifies things
+            let mut reader = BufReader::new(cbmm_stream);
+            let mut buf = vec![];
+            timeout(self.config.cbmm_timeout, reader.read_until(b'\r', &mut buf))
+                .await
+                .map_err(format_err("reading", 504))?
+                .map_err(format_err("reading", 502))?;
+
+            let s = str::from_utf8(&buf).map_err(format_err("utf8", 502))?;
+            let text = s.trim().to_owned();
+            Ok((text, vec![]))
+        }
+    }
+
+    // Handler function that processes incoming http requests (push's from alertmanager expected)
+    pub async fn handle_http_request(
+        &self,
+        req: Request<Incoming>,
+    ) -> Result<Response<String>, String> {
+        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.method(), req.uri().path()) {
+            (&Method::GET, "/") => Ok(ok_resp()),
+            (&Method::GET, "/health") => Ok(ok_resp()),
+            (&Method::POST, "/alert") => {
+                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,
+    ) -> Result<(String, Vec<String>), (u16, Box<dyn Error>)> {
+        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![]));
+                }
+                let mut text = String::new();
+                for (origin, handler) in handlers.iter() {
+                    // Apply filter if present
+                    if let Some(ref f) = filter
+                        && !origin.matches_filter(f)
+                    {
+                        continue;
+                    }
+                    writeln!(&mut text, "=== [{origin}] ===").unwrap();
+                    text.push_str(&handler.format_logs().await);
+                    text.push('\n');
+                }
+                if text.is_empty() {
+                    return Ok(("No matching log sources".to_string(), vec![]));
+                }
+                Ok((text, vec![]))
+            }
+            GatewayCommand::Query { query } => {
+                let plotter = self
+                    .plotter
+                    .as_ref()
+                    .ok_or_else(|| (500, "prometheus was not configured".into()))?;
+
+                match plotter.oneoff_query(query).await {
+                    Ok((
+                        ExtractLabels {
+                            name,
+                            common_labels,
+                            specific_labels,
+                        },
+                        ts,
+                    )) => {
+                        let mut text = format!("{name} {common_labels:?}\n");
+
+                        for (mut sl, maybe_val) in specific_labels.into_iter().zip(ts.into_iter()) {
+                            let name = sl.remove("__name__").unwrap_or_default();
+                            let label = format!("{name} {sl:?}");
+
+                            // TODO: Include timestamp?
+                            let val = maybe_val
+                                .map(|(_time, val)| val.to_string())
+                                .unwrap_or_else(|| "-".to_string());
+                            if let Err(err) = writeln!(&mut text, "\t{label}\t\t{val}") {
+                                return Err((500, err.into()));
+                            }
+                        }
+
+                        Ok((text, vec![]))
+                    }
+                    Err(err) => Err((500, err)),
+                }
+            }
+            GatewayCommand::Plot { query, duration } => {
+                let plotter = self
+                    .plotter
+                    .as_ref()
+                    .ok_or_else(|| (500, "prometheus was not configured".into()))?;
+
+                plotter.purge_old_plots();
+                match plotter.create_oneoff_plot(query.clone(), duration).await {
+                    Ok(filename) => Ok((query, vec![filename])),
+                    Err(err) => Err((500, err)),
+                }
+            }
+            GatewayCommand::Series { matchers } => {
+                let plotter = self
+                    .plotter
+                    .as_ref()
+                    .ok_or_else(|| (500, "prometheus was not configured".into()))?;
+
+                let matcher_refs: Vec<&str> = matchers.iter().map(|s| s.as_str()).collect();
+                match plotter.series(&matcher_refs).await {
+                    Ok(data) => {
+                        let mut text = data.iter().fold(String::default(), |mut buf, kv| {
+                            writeln!(&mut buf, "{kv:?}").unwrap();
+                            buf
+                        });
+                        if text.is_empty() {
+                            text = "no matches".into();
+                        }
+                        Ok((text, vec![]))
+                    }
+                    Err(err) => Err((500, err)),
+                }
+            }
+            GatewayCommand::Labels { matchers } => {
+                let plotter = self
+                    .plotter
+                    .as_ref()
+                    .ok_or_else(|| (500, "prometheus was not configured".into()))?;
+
+                let matcher_refs: Vec<&str> = matchers.iter().map(|s| s.as_str()).collect();
+                match plotter.labels(&matcher_refs).await {
+                    Ok(data) => {
+                        let mut text = data.iter().fold(String::default(), |mut buf, l| {
+                            writeln!(&mut buf, "{l}").unwrap();
+                            buf
+                        });
+                        if text.is_empty() {
+                            text = "no matches".into();
+                        }
+                        Ok((text, vec![]))
+                    }
+                    Err(err) => Err((500, err)),
+                }
+            }
+            GatewayCommand::Alerts => {
+                let plotter = self
+                    .plotter
+                    .as_ref()
+                    .ok_or_else(|| (500, "prometheus was not configured".into()))?;
+
+                match plotter.alerts().await {
+                    Ok(data) => {
+                        let now = Utc::now();
+                        let mut text = data.iter().fold(String::default(), |mut buf, alert| {
+                            let symbol = match alert.state {
+                                AlertStatus::Pending => "🟡",
+                                AlertStatus::Firing => "🔴",
+                                AlertStatus::Resolved => "🟢",
+                            };
+                            let since = {
+                                let dur = (now - alert.active_at).to_std().unwrap_or_default();
+                                // reduce precision to at most seconds
+                                let mut secs = dur.as_secs();
+                                // if the duration is more than an hour, then reduce precision to minutes
+                                if secs > 3600 {
+                                    secs -= secs % 60;
+                                }
+                                humantime::format_duration(Duration::new(secs, 0))
+                            };
+                            let annotations = &alert.annotations;
+                            let labels = &alert.labels;
+
+                            writeln!(&mut buf, "{symbol} {since} {annotations:?} {labels:?}")
+                                .unwrap();
+                            buf
+                        });
+                        if text.is_empty() {
+                            text = "no alerts".into();
+                        }
+                        Ok((text, vec![]))
+                    }
+                    Err(err) => Err((500, err)),
+                }
+            }
+        }
+    }
+
+    async fn handle_post_alert(&self, body_bytes: &[u8]) -> Result<(), (StatusCode, &'static str)> {
+        let body_text = str::from_utf8(body_bytes).map_err(|err| {
+            warn!("When reading body bytes: {err}");
+            (StatusCode::BAD_REQUEST, "Request body was not utf-8")
+        })?;
+
+        let alert_msg: AlertMessage = 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:#?}"));
+
+        let mut attachment_paths = vec![];
+        if let Some(plotter) = self.plotter.as_ref() {
+            plotter.purge_old_plots();
+            for alert in alert_msg.alerts.iter() {
+                match plotter.create_alert_plot(alert).await {
+                    Ok(path) => {
+                        attachment_paths.push(path);
+                    }
+                    Err(err) => {
+                        error!("Could not format plot: {err} for {alert:#?}");
+                    }
+                }
+            }
+        }
+
+        // Build summary: status sigils followed by alert names
+        let summary = alert_msg
+            .alerts
+            .iter()
+            .map(|alert| {
+                let symbol = match alert.status {
+                    Status::Firing => "🔴",
+                    Status::Resolved => "🟢",
+                };
+                let name = alert
+                    .labels
+                    .get("alertname")
+                    .map(|s| s.as_str())
+                    .unwrap_or("?");
+                format!("{symbol}{name}")
+            })
+            .collect::<Vec<_>>()
+            .join(" ");
+
+        self.admin_mq_tx
+            .send(AdminMessage {
+                origin: None, // Prometheus alerts don't have a syslog origin
+                text,
+                attachment_paths,
+                summary: Some(summary),
+            })
+            .map_err(|_err| {
+                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: &AlertMessage) -> Result<String, String> {
+        let mut text = "Alert:\n".to_owned();
+        let now = Utc::now();
+
+        for alert in msg.alerts.iter() {
+            let symbol = match alert.status {
+                Status::Firing => "🔴",
+                Status::Resolved => "🟢",
+            };
+            let since = {
+                let dur = (now - alert.starts_at).to_std().unwrap_or_default();
+                // reduce precision to at most seconds
+                let mut secs = dur.as_secs();
+                // if the duration is more than an hour, then reduce precision to minutes
+                if secs > 3600 {
+                    secs -= secs % 60;
+                }
+                humantime::format_duration(Duration::new(secs, 0))
+            };
+            let name = alert
+                .annotations
+                .get("summary")
+                .or_else(|| alert.labels.get("alertname"))
+                .map(|s| s.as_str())
+                .unwrap_or("?");
+
+            let expr = match alert.parse_expr_from_generator_url() {
+                Ok(expr) => expr,
+                Err(err) => {
+                    error!(
+                        "Couldn't parse generator url {}: {err}",
+                        alert.generator_url
+                    );
+                    String::default()
+                }
+            };
+
+            writeln!(&mut text, "{symbol}: ({since}) '{name}' {expr}")
+                .map_err(|err| format!("formatting error: {err}"))?;
+        }
+
+        Ok(text)
+    }
+
+    pub async fn handle_syslog_message(&self, syslog_msg: SyslogMessage) {
+        let origin = Origin::from(&syslog_msg);
+
+        // Try to get existing handler with read lock first
+        {
+            let handlers = self.log_handlers.read().await;
+            if let Some(handler) = handlers.get(&origin) {
+                handler.handle_syslog_message(syslog_msg, origin).await;
+                return;
+            }
+        }
+
+        // Handler doesn't exist, need to create one with write lock
+        let mut handlers = self.log_handlers.write().await;
+        // Double-check in case another task created it while we were waiting for the write lock
+        let handler = handlers.entry(origin.clone()).or_insert_with(|| {
+            info!("Creating new log handler for origin: {origin}");
+            LogHandler::new(self.config.log_handler.clone(), self.admin_mq_tx.clone())
+        });
+        handler.handle_syslog_message(syslog_msg, origin).await;
+    }
+}
+
+impl Drop for Gateway {
+    fn drop(&mut self) {
+        self.token.cancel();
+    }
+}
+
+// Returns a lambda that expresses an error as a (u16, String) with given context info
+fn format_err<E: std::fmt::Display>(
+    context: &'static str,
+    code: u16,
+) -> impl Fn(E) -> (u16, Box<dyn Error>) {
+    move |err: E| -> (u16, Box<dyn Error>) { (code, format!("{context}: {err}").into()) }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_gateway_command() {
+        // Test log command without filter
+        let cmd = parse_gateway_command("/log").unwrap();
+        assert!(matches!(cmd, GatewayCommand::Log { filter: None }));
+
+        // Test log command with filter
+        let cmd = parse_gateway_command("/log myapp").unwrap();
+        if let GatewayCommand::Log { filter } = cmd {
+            assert_eq!(filter, Some("myapp".to_string()));
+        } else {
+            panic!("Expected Log command");
+        }
+
+        // Test uppercase log command
+        let cmd = parse_gateway_command("/LOG").unwrap();
+        assert!(matches!(cmd, GatewayCommand::Log { filter: None }));
+
+        // Test query command
+        let cmd = parse_gateway_command("/query test_metric").unwrap();
+        if let GatewayCommand::Query { query } = cmd {
+            assert_eq!(query, "test_metric");
+        } else {
+            panic!("Expected Query command");
+        }
+
+        // Test query command (uppercase)
+        let cmd = parse_gateway_command("/QUERY test_metric").unwrap();
+        if let GatewayCommand::Query { query } = cmd {
+            assert_eq!(query, "test_metric");
+        } else {
+            panic!("Expected Query command");
+        }
+
+        // Test plot command with default duration
+        let cmd = parse_gateway_command("/plot my_query").unwrap();
+        if let GatewayCommand::Plot { query, duration } = cmd {
+            assert_eq!(query, "my_query");
+            assert_eq!(duration, Duration::from_secs(60 * 60));
+        } else {
+            panic!("Expected Plot command");
+        }
+
+        // Test plot command with custom duration
+        let cmd = parse_gateway_command("/plot my_query -d 24h").unwrap();
+        if let GatewayCommand::Plot { query, duration } = cmd {
+            assert_eq!(query, "my_query");
+            assert_eq!(duration, Duration::from_secs(24 * 60 * 60));
+        } else {
+            panic!("Expected Plot command");
+        }
+
+        // Test series command with matchers
+        let cmd = parse_gateway_command("/series metric1 metric2").unwrap();
+        if let GatewayCommand::Series { matchers } = cmd {
+            assert_eq!(matchers, vec!["metric1", "metric2"]);
+        } else {
+            panic!("Expected Series command");
+        }
+
+        // Test labels command
+        let cmd = parse_gateway_command("/labels foo bar").unwrap();
+        if let GatewayCommand::Labels { matchers } = cmd {
+            assert_eq!(matchers, vec!["foo", "bar"]);
+        } else {
+            panic!("Expected Labels command");
+        }
+
+        // Test alerts command
+        let cmd = parse_gateway_command("/alerts").unwrap();
+        assert!(matches!(cmd, GatewayCommand::Alerts));
+
+        // Test alerts command (uppercase)
+        let cmd = parse_gateway_command("/ALERTS").unwrap();
+        assert!(matches!(cmd, GatewayCommand::Alerts));
+
+        // Test without leading slash
+        let cmd = parse_gateway_command("log").unwrap();
+        assert!(matches!(cmd, GatewayCommand::Log { filter: None }));
+
+        // Test empty command
+        assert!(parse_gateway_command("/").is_err());
+        assert!(parse_gateway_command("").is_err());
+    }
+
+    #[test]
+    fn test_origin_matches_filter() {
+        let origin = Origin {
+            app: "muad-dib".to_string(),
+            host: "tokyo-server".to_string(),
+        };
+
+        // Without @: matches if app OR host contains the string
+        assert!(origin.matches_filter("muad"));
+        assert!(origin.matches_filter("dib"));
+        assert!(origin.matches_filter("tokyo"));
+        assert!(origin.matches_filter("server"));
+        assert!(!origin.matches_filter("paris"));
+
+        // With @: app must contain first part AND host must contain second part
+        assert!(origin.matches_filter("muad@tokyo"));
+        assert!(origin.matches_filter("dib@server"));
+        assert!(origin.matches_filter("muad-dib@tokyo-server"));
+        assert!(!origin.matches_filter("muad@paris"));
+        assert!(!origin.matches_filter("other@tokyo"));
+
+        // Empty parts with @
+        assert!(origin.matches_filter("@tokyo")); // empty app filter matches any app
+        assert!(origin.matches_filter("muad@")); // empty host filter matches any host
+        assert!(origin.matches_filter("@")); // both empty, matches everything
+
+        // Edge case: filter matches the @ in the format but origin has no @
+        let origin2 = Origin {
+            app: "app".to_string(),
+            host: "host".to_string(),
+        };
+        assert!(origin2.matches_filter("app@host"));
+        assert!(!origin2.matches_filter("app@other"));
+    }
+}

signal-gateway/src/gateway/rate_limiter.rs

@@ -0,0 +1,288 @@
+use serde::Deserialize;
+use std::{
+    collections::HashMap,
+    str::FromStr,
+    sync::atomic::{AtomicI64, Ordering},
+    time::Duration,
+};
+
+/// Represents a rate threshold, expressed as a string in the format:
+///
+/// * `1 / 10s`
+/// * `2 / 5m`
+/// * `3 / 1h`
+/// * `> 1 / 10s`
+/// * `>= 2 / 10s`
+///
+/// When the comparator is omitted, it is treated as `>=`
+#[derive(Clone, Copy, Debug, Deserialize)]
+#[serde(try_from = "String")]
+pub struct RateThreshold {
+    pub times: usize,
+    pub duration: Duration,
+}
+
+impl FromStr for RateThreshold {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let Some((first, second)) = s.trim().split_once('/') else {
+            return Err("missing '/' character in rate threshold".into());
+        };
+
+        let duration = conf_extra::parse_duration(second.trim())?;
+
+        let first = first.trim();
+        let maybe_mid = first.as_bytes().iter().position(|b| b.is_ascii_digit());
+        let (comparator, num) = if let Some(mid) = maybe_mid {
+            first.split_at(mid)
+        } else {
+            ("", first)
+        };
+
+        let is_greater_equal = match comparator.trim() {
+            ">" => false,
+            ">=" | "=>" | "" => true,
+            _ => return Err(format!("Unexpected comparator format: {comparator}")),
+        };
+
+        let num = num.trim();
+        let mut times: usize = num
+            .parse()
+            .map_err(|err| format!("invalid number {num}: {err}"))?;
+
+        if !is_greater_equal {
+            times += 1;
+        }
+
+        if times == 0 {
+            return Err("Invalid threshold, times must be > 0".into());
+        }
+
+        Ok(RateThreshold { times, duration })
+    }
+}
+
+impl TryFrom<String> for RateThreshold {
+    type Error = <RateThreshold as FromStr>::Err;
+    fn try_from(s: String) -> Result<Self, Self::Error> {
+        RateThreshold::from_str(&s)
+    }
+}
+
+/// A rate limiter containing a single counter, and a minimum time window for the next event to pass
+#[allow(dead_code)]
+#[derive(Debug, Default)]
+pub struct SimpleRateLimiter {
+    last_timestamp: AtomicI64,
+    window: i64,
+}
+
+#[allow(dead_code)]
+impl SimpleRateLimiter {
+    pub fn new(window: Duration) -> Self {
+        Self {
+            last_timestamp: Default::default(),
+            window: window.as_secs().try_into().unwrap(),
+        }
+    }
+
+    /// Check if a particular new timestamp passes the limit. This also updates the last-known timestamp.
+    pub fn evaluate(&self, ts_sec: i64) -> bool {
+        let last_ts = self.last_timestamp.load(Ordering::SeqCst);
+        let rate_limited = ts_sec - last_ts < self.window;
+        if !rate_limited && ts_sec > last_ts {
+            // If this is called concurrently, guarantee that we keep going
+            // until the max value is stored at self.last_timestamp,
+            // so self.last_timestamp is "eventually" only monotonically increasing.
+            store_max(ts_sec, &self.last_timestamp);
+        }
+        !rate_limited
+    }
+}
+
+#[allow(dead_code)]
+fn store_max(val: i64, at: &AtomicI64) {
+    let prev = at.swap(val, Ordering::SeqCst);
+    if prev > val {
+        store_max(prev, at)
+    }
+}
+
+/// A rate limiter that tracks alerts per source location (file:line).
+///
+/// This allows different error locations to alert independently, preventing one noisy
+/// error from suppressing alerts from completely different code paths.
+#[derive(Debug)]
+pub struct SourceLocationRateLimiter {
+    /// Maps (file, line) -> last alert timestamp
+    last_timestamps: HashMap<(String, String), i64>,
+    /// The rate limiting window in seconds
+    window: i64,
+    /// Maximum entries before triggering cleanup
+    max_entries: usize,
+}
+
+impl SourceLocationRateLimiter {
+    pub fn new(window: Duration, max_entries: usize) -> Self {
+        Self {
+            last_timestamps: HashMap::new(),
+            window: window.as_secs().try_into().unwrap(),
+            max_entries,
+        }
+    }
+
+    /// Check if an error from this source location should trigger an alert.
+    ///
+    /// Returns true if the alert should fire (not rate-limited), false if suppressed.
+    /// Updates the stored timestamp if the alert fires.
+    pub fn evaluate(&mut self, file: &str, line: &str, ts_sec: i64) -> bool {
+        let key = (file.to_owned(), line.to_owned());
+
+        if let Some(&last_ts) = self.last_timestamps.get(&key)
+            && ts_sec - last_ts < self.window
+        {
+            return false; // Rate limited
+        }
+
+        // Alert should fire - update timestamp
+        self.last_timestamps.insert(key, ts_sec);
+
+        // Clean up if we've exceeded max entries
+        if self.last_timestamps.len() > self.max_entries {
+            self.cleanup(ts_sec);
+        }
+
+        true
+    }
+
+    /// Remove entries older than the window
+    fn cleanup(&mut self, now: i64) {
+        self.last_timestamps
+            .retain(|_, &mut ts| now - ts < self.window);
+    }
+}
+
+/// Implements rate-limiting criteria such as 'at least n in the last w seconds'
+#[derive(Debug)]
+pub struct MultiRateLimiter {
+    /// Records the last n events
+    timestamps: Vec<i64>,
+    /// Invariant: Always points to the oldest of the last n timestamps in the buffer
+    idx: usize,
+    /// The length of the window (in seconds)
+    window: i64,
+}
+
+impl MultiRateLimiter {
+    pub fn new(num: usize, window: Duration) -> Self {
+        Self {
+            idx: 0,
+            timestamps: vec![Default::default(); num],
+            window: window.as_secs().try_into().unwrap(),
+        }
+    }
+
+    /// Check if a particular new timestamp passes the limit. This also updates the last-known timestamp.
+    ///
+    /// Note: Assumes that new_timestamp is monotonically increasing, otherwise it might not work right.
+    pub fn evaluate(&mut self, new_timestamp: i64) -> bool {
+        let oldest = self.timestamps[self.idx];
+        if oldest >= new_timestamp {
+            return false;
+        }
+        self.timestamps[self.idx] = new_timestamp;
+        self.idx += 1;
+        self.idx %= self.timestamps.len();
+        let next_oldest = self.timestamps[self.idx];
+        // If the next oldest is within 'window' of the new timestamp,
+        // then all of the most recent n are. Otherwise, at most n-1 of the most recent are.
+        next_oldest + self.window >= new_timestamp
+    }
+}
+
+impl From<RateThreshold> for MultiRateLimiter {
+    fn from(src: RateThreshold) -> Self {
+        Self::new(src.times, src.duration)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_rate_threshold() {
+        let threshold = RateThreshold::from_str("1/10s").unwrap();
+        assert_eq!(threshold.times, 1);
+        assert_eq!(threshold.duration, Duration::from_secs(10));
+        let threshold = RateThreshold::from_str("1 / 10s").unwrap();
+        assert_eq!(threshold.times, 1);
+        assert_eq!(threshold.duration, Duration::from_secs(10));
+
+        let threshold = RateThreshold::from_str("2 / 5m").unwrap();
+        assert_eq!(threshold.times, 2);
+        assert_eq!(threshold.duration, Duration::from_secs(300));
+
+        let threshold = RateThreshold::from_str("> 3 / 10m").unwrap();
+        assert_eq!(threshold.times, 4);
+        assert_eq!(threshold.duration, Duration::from_secs(600));
+
+        let threshold = RateThreshold::from_str(">=3/10m").unwrap();
+        assert_eq!(threshold.times, 3);
+        assert_eq!(threshold.duration, Duration::from_secs(600));
+    }
+
+    #[test]
+    fn source_location_rate_limiter_basic() {
+        let mut limiter = SourceLocationRateLimiter::new(Duration::from_secs(600), 100);
+
+        // First alert from location A should pass
+        assert!(limiter.evaluate("file_a.rs", "10", 1000));
+
+        // Second alert from same location within window should be rate limited
+        assert!(!limiter.evaluate("file_a.rs", "10", 1100));
+
+        // Alert from different location should pass (independent rate limiting)
+        assert!(limiter.evaluate("file_b.rs", "20", 1100));
+
+        // Same location after window passes should alert again
+        assert!(limiter.evaluate("file_a.rs", "10", 1700)); // 1000 + 600 + 100
+    }
+
+    #[test]
+    fn source_location_rate_limiter_different_lines_same_file() {
+        let mut limiter = SourceLocationRateLimiter::new(Duration::from_secs(600), 100);
+
+        // Different lines in same file should be independent
+        assert!(limiter.evaluate("file.rs", "10", 1000));
+        assert!(limiter.evaluate("file.rs", "20", 1000));
+        assert!(limiter.evaluate("file.rs", "30", 1000));
+
+        // Each should still be rate limited individually
+        assert!(!limiter.evaluate("file.rs", "10", 1100));
+        assert!(!limiter.evaluate("file.rs", "20", 1100));
+    }
+
+    #[test]
+    fn source_location_rate_limiter_cleanup() {
+        // Use small max_entries to trigger cleanup
+        let mut limiter = SourceLocationRateLimiter::new(Duration::from_secs(600), 3);
+
+        // Fill up the limiter
+        assert!(limiter.evaluate("file1.rs", "1", 1000));
+        assert!(limiter.evaluate("file2.rs", "2", 1000));
+        assert!(limiter.evaluate("file3.rs", "3", 1000));
+        assert_eq!(limiter.last_timestamps.len(), 3);
+
+        // Add one more, triggering cleanup - but all are fresh so none removed
+        assert!(limiter.evaluate("file4.rs", "4", 1000));
+        // Still have 4 after cleanup since none are old enough
+        assert_eq!(limiter.last_timestamps.len(), 4);
+
+        // Now add with a timestamp far in the future - old entries should be cleaned
+        assert!(limiter.evaluate("file5.rs", "5", 2000));
+        // Should have cleaned up entries from timestamp 1000 (older than 600 sec window)
+        assert_eq!(limiter.last_timestamps.len(), 1);
+    }
+}

signal-gateway/src/http.rs

@@ -0,0 +1,85 @@
+//! Schema for the alertmanager http POST requests that are sent to us
+
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use url::Url;
+
+pub type Timestamp = DateTime<Utc>;
+
+#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Status {
+    Resolved,
+    Firing,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AlertMessage {
+    // should be 4.0
+    pub version: String,
+    pub group_key: String,
+    pub status: Status,
+    pub receiver: String,
+    #[serde(default)]
+    pub group_labels: BTreeMap<String, String>,
+    #[serde(default)]
+    pub common_labels: BTreeMap<String, String>,
+    #[serde(default)]
+    pub common_annotations: BTreeMap<String, String>,
+    #[serde(alias = "externalURL")]
+    pub external_url: String,
+    pub alerts: Vec<Alert>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Alert {
+    pub status: Status,
+    #[serde(default)]
+    pub labels: BTreeMap<String, String>,
+    #[serde(default)]
+    pub annotations: BTreeMap<String, String>,
+    pub starts_at: Timestamp,
+    pub ends_at: Timestamp,
+    #[serde(alias = "generatorURL")]
+    pub generator_url: String,
+    pub fingerprint: String,
+}
+
+impl Alert {
+    pub fn parse_expr_from_generator_url(&self) -> Result<String, String> {
+        let url = Url::parse(&self.generator_url).map_err(|err| err.to_string())?;
+
+        for (k, v) in url.query_pairs() {
+            if k == "g0.expr" {
+                return Ok(v.into_owned());
+            }
+        }
+
+        Err("Couldn't find g0.expr".to_string())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_alert_message_parsing() {
+        let text = r#"{"receiver":"notify-chris","status":"firing","alerts":[{"status":"firing","labels":{"alertname":"Low tick success rate","instance":"172.31.5.8:9000","job":"ec2"},"annotations":{"summary":"Low tick success rate"},"startsAt":"2025-11-07T04:21:46.17Z","endsAt":"0001-01-01T00:00:00Z","generatorURL":"http://ip-172-31-10-138.eu-west-3.compute.internal:9090/graph?g0.expr=rate%28tick_successes%5B5m%5D%29+%3C+0.9\u0026g0.tab=1","fingerprint":"543b6a7a3042ae2c"},{"status":"firing","labels":{"alertname":"Long tail tick times","instance":"172.31.5.8:9000","job":"ec2","quantile":"0.99"},"annotations":{"summary":"Long tail tick times"},"startsAt":"2025-11-07T04:50:01.17Z","endsAt":"0001-01-01T00:00:00Z","generatorURL":"http://ip-172-31-10-138.eu-west-3.compute.internal:9090/graph?g0.expr=tick_time%7Bquantile%3D%220.99%22%7D+%3E+0.8\u0026g0.tab=1","fingerprint":"97130d38ef0ff0a4"}],"groupLabels":{},"commonLabels":{"instance":"172.31.5.8:9000","job":"ec2"},"commonAnnotations":{},"externalURL":"http://ip-172-31-10-138.eu-west-3.compute.internal:9093","version":"4","groupKey":"{}:{}","truncatedAlerts":0}"#;
+
+        let msg: AlertMessage = serde_json::from_str(text).unwrap();
+
+        assert_eq!(&msg.receiver, "notify-chris");
+        assert_eq!(msg.alerts.len(), 2);
+        assert_eq!(
+            msg.alerts[0].annotations.get("summary").unwrap(),
+            "Low tick success rate"
+        );
+
+        let expr = msg.alerts[1].parse_expr_from_generator_url().unwrap();
+        assert_eq!(expr, r#"tick_time{quantile="0.99"} > 0.8"#);
+    }
+}

signal-gateway/src/human_duration.rs

@@ -0,0 +1,63 @@
+//! Display wrapper for formatting chrono::TimeDelta as "T-duration"
+
+use chrono::TimeDelta;
+use std::{
+    fmt::{self, Display},
+    time::Duration,
+};
+
+/// A display wrapper that formats a chrono::TimeDelta as "T-duration" or "T+duration"
+/// with coarse precision (truncated to minutes for durations > 1 hour).
+#[derive(Clone, Copy, Debug)]
+pub struct HumanTMinus(pub TimeDelta);
+
+impl Display for HumanTMinus {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let (prefix, abs_duration) = if self.0.num_seconds() >= 0 {
+            ("T-", self.0)
+        } else {
+            ("T+", -self.0)
+        };
+
+        let std_dur = abs_duration.to_std().unwrap_or_default();
+        let mut secs = std_dur.as_secs();
+
+        // Reduce precision to minutes for durations > 1 hour
+        if secs > 3600 {
+            secs -= secs % 60;
+        }
+
+        let coarse_dur = Duration::new(secs, 0);
+        // Remove spaces for compact format (e.g., "T-1m30s" not "T-1m 30s")
+        let formatted = humantime::format_duration(coarse_dur)
+            .to_string()
+            .replace(' ', "");
+        write!(f, "{}{}", prefix, formatted)
+    }
+}
+
+impl From<TimeDelta> for HumanTMinus {
+    fn from(td: TimeDelta) -> Self {
+        HumanTMinus(td)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_human_t_minus_positive() {
+        assert_eq!(HumanTMinus(TimeDelta::seconds(30)).to_string(), "T-30s");
+        assert_eq!(HumanTMinus(TimeDelta::seconds(90)).to_string(), "T-1m30s");
+        assert_eq!(HumanTMinus(TimeDelta::hours(1)).to_string(), "T-1h");
+        // > 1 hour: truncate to minutes
+        assert_eq!(HumanTMinus(TimeDelta::seconds(3700)).to_string(), "T-1h1m");
+    }
+
+    #[test]
+    fn test_human_t_minus_negative() {
+        assert_eq!(HumanTMinus(TimeDelta::seconds(-30)).to_string(), "T+30s");
+        assert_eq!(HumanTMinus(TimeDelta::seconds(-3700)).to_string(), "T+1h1m");
+    }
+}

signal-gateway/src/init_logging.rs

@@ -0,0 +1,41 @@
+//! Logging initialization for signal-gateway
+
+use std::env;
+use tracing::info;
+
+pub fn init_logging() {
+    // Install rustls crypto provider before any TLS connections are made.
+    // This is needed because we have both aws-lc-rs and ring in our dependency tree,
+    // and rustls 0.23 can't auto-detect which one to use when both are present.
+    rustls::crypto::aws_lc_rs::default_provider()
+        .install_default()
+        .expect("Failed to install rustls crypto provider");
+
+    if env::var("RUST_LOG").is_err() {
+        unsafe {
+            env::set_var("RUST_LOG", "info");
+        }
+    }
+
+    // Build a default tracing subscriber, writing to STDERR
+    tracing_subscriber::fmt()
+        .with_writer(std::io::stderr)
+        .with_file(true)
+        .with_line_number(true)
+        .init();
+
+    // load dotenv file
+    match dotenvy::dotenv() {
+        Ok(path) => info!("Read dotenv file from: {}", path.display()),
+        Err(dotenvy::Error::Io(io_error)) => {
+            if matches!(io_error.kind(), std::io::ErrorKind::NotFound) {
+                info!("Couldn't find a dotenv file");
+            } else {
+                panic!("Io error when reading dot env file: {io_error}")
+            }
+        }
+        Err(err) => {
+            panic!("Error reading dotenv file: {err}")
+        }
+    }
+}

signal-gateway/src/jsonrpc.rs

@@ -0,0 +1,557 @@
+//! This copied from https://github.com/AsamK/signal-cli/blob/f9a36c6e0404d06bd396b24b5ea699e49ed29b89/client/src/jsonrpc.rs
+#![allow(clippy::too_many_arguments)]
+
+use jsonrpsee::async_client::ClientBuilder;
+use jsonrpsee::core::client::SubscriptionClientT;
+use jsonrpsee::proc_macros::rpc;
+use serde::Deserialize;
+use serde_json::Value;
+use tokio::net::ToSocketAddrs;
+
+pub use jsonrpsee::core::ClientError as RpcClientError;
+
+#[rpc(client)]
+pub trait Rpc {
+    #[method(name = "addDevice", param_kind = map)]
+    async fn add_device(
+        &self,
+        account: Option<String>,
+        uri: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "addStickerPack", param_kind = map)]
+    async fn add_sticker_pack(
+        &self,
+        account: Option<String>,
+        uri: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "block", param_kind = map)]
+    fn block(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "deleteLocalAccountData", param_kind = map)]
+    fn delete_local_account_data(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] ignoreRegistered: Option<bool>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "getAttachment", param_kind = map)]
+    fn get_attachment(
+        &self,
+        account: Option<String>,
+        id: String,
+        recipient: Option<String>,
+        #[allow(non_snake_case)] groupId: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "getAvatar", param_kind = map)]
+    fn get_avatar(
+        &self,
+        account: Option<String>,
+        contact: Option<String>,
+        profile: Option<String>,
+        #[allow(non_snake_case)] groupId: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "getSticker", param_kind = map)]
+    fn get_sticker(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] packId: String,
+        #[allow(non_snake_case)] stickerId: u32,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "getUserStatus", param_kind = map)]
+    fn get_user_status(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        usernames: Vec<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "joinGroup", param_kind = map)]
+    fn join_group(&self, account: Option<String>, uri: String) -> Result<Value, ErrorObjectOwned>;
+
+    #[allow(non_snake_case)]
+    #[method(name = "finishChangeNumber", param_kind = map)]
+    fn finish_change_number(
+        &self,
+        account: Option<String>,
+        number: String,
+        verificationCode: String,
+        pin: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "finishLink", param_kind = map)]
+    fn finish_link(
+        &self,
+        #[allow(non_snake_case)] deviceLinkUri: String,
+        #[allow(non_snake_case)] deviceName: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "listAccounts", param_kind = map)]
+    fn list_accounts(&self) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "listContacts", param_kind = map)]
+    fn list_contacts(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] allRecipients: bool,
+        blocked: Option<bool>,
+        name: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "listDevices", param_kind = map)]
+    fn list_devices(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "listGroups", param_kind = map)]
+    fn list_groups(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "listIdentities", param_kind = map)]
+    fn list_identities(
+        &self,
+        account: Option<String>,
+        number: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "listStickerPacks", param_kind = map)]
+    fn list_sticker_packs(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "quitGroup", param_kind = map)]
+    fn quit_group(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] groupId: String,
+        delete: bool,
+        admins: Vec<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "register", param_kind = map)]
+    fn register(
+        &self,
+        account: Option<String>,
+        voice: bool,
+        captcha: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "removeContact", param_kind = map)]
+    fn remove_contact(
+        &self,
+        account: Option<String>,
+        recipient: String,
+        forget: bool,
+        hide: bool,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "removeDevice", param_kind = map)]
+    fn remove_device(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] deviceId: u32,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "removePin", param_kind = map)]
+    fn remove_pin(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "remoteDelete", param_kind = map)]
+    fn remote_delete(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] targetTimestamp: u64,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+        #[allow(non_snake_case)] noteToSelf: bool,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[allow(non_snake_case)]
+    #[method(name = "send", param_kind = map)]
+    fn send(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        groupIds: Vec<String>,
+        noteToSelf: bool,
+        endSession: bool,
+        message: String,
+        attachments: Vec<String>,
+        viewOnce: bool,
+        mentions: Vec<String>,
+        textStyle: Vec<String>,
+        quoteTimestamp: Option<u64>,
+        quoteAuthor: Option<String>,
+        quoteMessage: Option<String>,
+        quoteMention: Vec<String>,
+        quoteTextStyle: Vec<String>,
+        quoteAttachment: Vec<String>,
+        previewUrl: Option<String>,
+        previewTitle: Option<String>,
+        previewDescription: Option<String>,
+        previewImage: Option<String>,
+        sticker: Option<String>,
+        storyTimestamp: Option<u64>,
+        storyAuthor: Option<String>,
+        editTimestamp: Option<u64>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendContacts", param_kind = map)]
+    fn send_contacts(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendPaymentNotification", param_kind = map)]
+    fn send_payment_notification(
+        &self,
+        account: Option<String>,
+        recipient: String,
+        receipt: String,
+        note: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendReaction", param_kind = map)]
+    fn send_reaction(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+        #[allow(non_snake_case)] noteToSelf: bool,
+        emoji: String,
+        #[allow(non_snake_case)] targetAuthor: String,
+        #[allow(non_snake_case)] targetTimestamp: u64,
+        remove: bool,
+        story: bool,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendReceipt", param_kind = map)]
+    fn send_receipt(
+        &self,
+        account: Option<String>,
+        recipient: String,
+        #[allow(non_snake_case)] targetTimestamps: Vec<u64>,
+        r#type: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendSyncRequest", param_kind = map)]
+    fn send_sync_request(&self, account: Option<String>) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendTyping", param_kind = map)]
+    fn send_typing(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+        stop: bool,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "sendMessageRequestResponse", param_kind = map)]
+    fn send_message_request_response(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+        r#type: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "setPin", param_kind = map)]
+    fn set_pin(&self, account: Option<String>, pin: String) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "submitRateLimitChallenge", param_kind = map)]
+    fn submit_rate_limit_challenge(
+        &self,
+        account: Option<String>,
+        challenge: String,
+        captcha: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "startChangeNumber", param_kind = map)]
+    fn start_change_number(
+        &self,
+        account: Option<String>,
+        number: String,
+        voice: bool,
+        captcha: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "startLink", param_kind = map)]
+    fn start_link(&self, account: Option<String>) -> Result<JsonLink, ErrorObjectOwned>;
+
+    #[method(name = "trust", param_kind = map)]
+    fn trust(
+        &self,
+        account: Option<String>,
+        recipient: String,
+        #[allow(non_snake_case)] trustAllKnownKeys: bool,
+        #[allow(non_snake_case)] verifiedSafetyNumber: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "unblock", param_kind = map)]
+    fn unblock(
+        &self,
+        account: Option<String>,
+        recipients: Vec<String>,
+        #[allow(non_snake_case)] groupIds: Vec<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "unregister", param_kind = map)]
+    fn unregister(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] deleteAccount: bool,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[allow(non_snake_case)]
+    #[method(name = "updateAccount", param_kind = map)]
+    fn update_account(
+        &self,
+        account: Option<String>,
+        deviceName: Option<String>,
+        unrestrictedUnidentifiedSender: Option<bool>,
+        discoverableByNumber: Option<bool>,
+        numberSharing: Option<bool>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "updateConfiguration", param_kind = map)]
+    fn update_configuration(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] readReceipts: Option<bool>,
+        #[allow(non_snake_case)] unidentifiedDeliveryIndicators: Option<bool>,
+        #[allow(non_snake_case)] typingIndicators: Option<bool>,
+        #[allow(non_snake_case)] linkPreviews: Option<bool>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "updateContact", param_kind = map)]
+    fn update_contact(
+        &self,
+        account: Option<String>,
+        recipient: String,
+        name: Option<String>,
+        expiration: Option<u32>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "updateGroup", param_kind = map)]
+    fn update_group(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] groupId: Option<String>,
+        name: Option<String>,
+        description: Option<String>,
+        avatar: Option<String>,
+        member: Vec<String>,
+        #[allow(non_snake_case)] removeMember: Vec<String>,
+        admin: Vec<String>,
+        #[allow(non_snake_case)] removeAdmin: Vec<String>,
+        ban: Vec<String>,
+        unban: Vec<String>,
+        #[allow(non_snake_case)] resetLink: bool,
+        #[allow(non_snake_case)] link: Option<String>,
+        #[allow(non_snake_case)] setPermissionAddMember: Option<String>,
+        #[allow(non_snake_case)] setPermissionEditDetails: Option<String>,
+        #[allow(non_snake_case)] setPermissionSendMessages: Option<String>,
+        expiration: Option<u32>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "updateProfile", param_kind = map)]
+    fn update_profile(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] givenName: Option<String>,
+        #[allow(non_snake_case)] familyName: Option<String>,
+        about: Option<String>,
+        #[allow(non_snake_case)] aboutEmoji: Option<String>,
+        #[allow(non_snake_case)] mobileCoinAddress: Option<String>,
+        avatar: Option<String>,
+        #[allow(non_snake_case)] removeAvatar: bool,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "uploadStickerPack", param_kind = map)]
+    fn upload_sticker_pack(
+        &self,
+        account: Option<String>,
+        path: String,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[method(name = "verify", param_kind = map)]
+    fn verify(
+        &self,
+        account: Option<String>,
+        #[allow(non_snake_case)] verificationCode: String,
+        pin: Option<String>,
+    ) -> Result<Value, ErrorObjectOwned>;
+
+    #[subscription(
+        name = "subscribeReceive" => "receive",
+        unsubscribe = "unsubscribeReceive",
+        item = RecvMessage,
+        param_kind = map
+    )]
+    async fn subscribe_receive(&self, account: Option<String>) -> SubscriptionResult;
+
+    #[method(name = "version")]
+    fn version(&self) -> Result<Value, ErrorObjectOwned>;
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct JsonLink {
+    pub device_link_uri: String,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RecvMessage {
+    pub envelope: Envelope,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Envelope {
+    pub source: String,
+    pub source_number: String,
+    pub source_uuid: String,
+    pub source_name: String,
+    pub source_device: i64,
+    pub timestamp: u64,
+    pub data_message: Option<DataMessage>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DataMessage {
+    pub timestamp: u64,
+    pub message: String,
+}
+
+/// Connect to signal-cli over tcp socket
+pub async fn connect_tcp(
+    tcp: impl ToSocketAddrs,
+) -> Result<impl SubscriptionClientT, std::io::Error> {
+    let (sender, receiver) = super::transports::tcp::connect(tcp).await?;
+
+    Ok(ClientBuilder::default().build_with_tokio(sender, receiver))
+}
+
+impl Envelope {
+    /// Send a read-receipt for an envelope
+    pub async fn send_read_receipt(
+        &self,
+        client: &impl RpcClient,
+        account: impl Into<String>,
+    ) -> Result<(), RpcClientError> {
+        if let Some(dm) = self.data_message.as_ref() {
+            let _ = client
+                .send_receipt(
+                    Some(account.into()),
+                    self.source_uuid.clone(),
+                    vec![dm.timestamp],
+                    "read".into(),
+                )
+                .await?;
+        }
+        Ok(())
+    }
+}
+
+/// Helper for invoking send, which has way too many parameters
+pub struct SignalMessage {
+    pub sender: String,
+    pub recipient: Vec<String>,
+    pub message: String,
+    pub attachments: Vec<String>,
+}
+
+impl SignalMessage {
+    #[allow(non_snake_case)]
+    pub async fn send(self, client: &impl RpcClient) -> Result<(), RpcClientError> {
+        // See note about string indexing here: https://github.com/AsamK/signal-cli/wiki/FAQ#string-indexing-units
+        let message_len_utf16: usize = self.message.chars().map(|c| c.len_utf16()).sum();
+        /*
+            account: Option<String>,
+            recipients: Vec<String>,
+            groupIds: Vec<String>,
+            noteToSelf: bool,
+            endSession: bool,
+            message: String,
+            attachments: Vec<String>,
+            viewOnce: bool,
+            mentions: Vec<String>,
+            textStyle: Vec<String>,
+            quoteTimestamp: Option<u64>,
+            quoteAuthor: Option<String>,
+            quoteMessage: Option<String>,
+            quoteMention: Vec<String>,
+            quoteTextStyle: Vec<String>,
+            quoteAttachment: Vec<String>,
+            previewUrl: Option<String>,
+            previewTitle: Option<String>,
+            previewDescription: Option<String>,
+            previewImage: Option<String>,
+            sticker: Option<String>,
+            storyTimestamp: Option<u64>,
+            storyAuthor: Option<String>,
+            editTimestamp: Option<u64>,
+        */
+        let account = Some(self.sender);
+        let recipients = self.recipient;
+        let groupIds = vec![];
+        let noteToSelf = false;
+        let endSession = false;
+        let message = self.message;
+        let attachments = self.attachments;
+        let viewOnce = false;
+        let mentions = vec![];
+        let textStyle = vec![format!("0:{message_len_utf16}:MONOSPACE")];
+        let quoteTimestamp = None;
+        let quoteAuthor = None;
+        let quoteMention = vec![];
+        let quoteMessage = None;
+        let quoteTextStyle = vec![];
+        let quoteAttachment = vec![];
+        let previewUrl = None;
+        let previewTitle = None;
+        let previewDescription = None;
+        let previewImage = None;
+        let sticker = None;
+        let storyTimestamp = None;
+        let storyAuthor = None;
+        let editTimestamp = None;
+
+        let _resp = client
+            .send(
+                account,
+                recipients,
+                groupIds,
+                noteToSelf,
+                endSession,
+                message,
+                attachments,
+                viewOnce,
+                mentions,
+                textStyle,
+                quoteTimestamp,
+                quoteAuthor,
+                quoteMessage,
+                quoteMention,
+                quoteTextStyle,
+                quoteAttachment,
+                previewUrl,
+                previewTitle,
+                previewDescription,
+                previewImage,
+                sticker,
+                storyTimestamp,
+                storyAuthor,
+                editTimestamp,
+            )
+            .await?;
+        Ok(())
+    }
+}

signal-gateway/src/main.rs

@@ -0,0 +1,147 @@
+use conf::Conf;
+use hyper::service::service_fn;
+use hyper_util::rt::TokioIo;
+use hyper_util::server::conn::auto;
+use std::{str::FromStr, sync::Arc, time::Duration};
+use syslog_rfc5424::SyslogMessage;
+use tokio::net::{TcpListener, UdpSocket};
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info, warn};
+
+pub mod config;
+pub mod gateway;
+pub mod http;
+mod human_duration;
+mod init_logging;
+pub mod jsonrpc;
+pub mod plotter;
+pub mod transports;
+
+use config::Config;
+use gateway::Gateway;
+
+#[tokio::main]
+async fn main() {
+    init_logging::init_logging();
+
+    let config = Config::parse();
+
+    info!("Config = {config:#?}");
+
+    if config.dry_run {
+        return;
+    }
+
+    let token = CancellationToken::new();
+
+    let gateway = Arc::new(Gateway::new(config.gateway, token.clone()).await);
+
+    let listener = TcpListener::bind(config.http_listen_addr).await.unwrap();
+    info!("Listening for http on {}", config.http_listen_addr);
+
+    let udp_socket = UdpSocket::bind(config.udp_listen_addr).await.unwrap();
+    info!("Listening for udp on {}", config.udp_listen_addr);
+
+    // Listen for ctrl-c
+    let thread_token = token.clone();
+    tokio::task::spawn(async move {
+        tokio::signal::ctrl_c().await.unwrap();
+        warn!("ctrl-c: Stop requested");
+        thread_token.cancel();
+    });
+
+    // Start the two server tasks
+    let _http_task = start_http_task(listener, gateway.clone());
+    let _udp_task = start_udp_task(udp_socket, gateway.clone());
+
+    // Run gateway task and block on it returning. Note that it exits if the token is canceled.
+    gateway.run().await;
+}
+
+fn start_http_task(listener: TcpListener, gateway: Arc<Gateway>) -> tokio::task::JoinHandle<()> {
+    // Loop waiting for http incoming connections, and pass them to gateway
+    tokio::task::spawn(async move {
+        loop {
+            let Ok((stream, remote_addr)) = listener
+                .accept()
+                .await
+                .inspect_err(|err| error!("Error accepting connection: {err}"))
+            else {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                continue;
+            };
+            info!("New connection from: {}", remote_addr);
+
+            // Spawn a new task to handle each connection
+            let thread_gateway = gateway.clone();
+            tokio::spawn(async move {
+                let io = TokioIo::new(stream);
+
+                // Serve the connection using auto protocol detection (HTTP/1 or HTTP/2)
+                if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new())
+                    .serve_connection(
+                        io,
+                        service_fn(|req| {
+                            let thread_gateway = thread_gateway.clone();
+                            async move { thread_gateway.handle_http_request(req).await }
+                        }),
+                    )
+                    .await
+                {
+                    error!("Error serving connection: {err}");
+                }
+            });
+        }
+    })
+}
+
+fn start_udp_task(udp_socket: UdpSocket, gateway: Arc<Gateway>) -> tokio::task::JoinHandle<()> {
+    // Loop waiting for http incoming connections, and pass them to gateway
+    tokio::task::spawn(async move {
+        let mut buf = vec![0u8; 8192];
+        loop {
+            let Ok((len, _addr)) = udp_socket
+                .recv_from(&mut buf)
+                .await
+                .inspect_err(|err| error!("Error receiving UDP packet: {err}"))
+            else {
+                continue;
+            };
+
+            let Ok(text) = str::from_utf8(&buf[0..len])
+                .inspect_err(|err| error!("UDP packet was not utf8: {err}"))
+            else {
+                continue;
+            };
+
+            let Ok(msg) = SyslogMessage::from_str(text)
+                .inspect_err(|err| error!("UDP packet was not valid syslog: {err}:\n{text}"))
+            else {
+                continue;
+            };
+
+            gateway.handle_syslog_message(msg).await;
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_syslog_parsing() {
+        SyslogMessage::from_str(
+            "<12>1 2025-11-08T02:24:10.815221698+00:00 ip-172-31-5-8 app 92748 - - Dropped 3/4 reports",
+        )
+        .unwrap();
+        SyslogMessage::from_str(
+            "<12>1 2025-11-08T02:24:10.815221698+00:00 ip-172-31-5-8 app 92748 - - Dropped 3/4 reports",
+        )
+        .unwrap();
+        SyslogMessage::from_str(
+            "<12>1 2025-11-08T02:24:10.815+00:00 ip-172-31-5-8 app 92748 - - Dropped 3/4 reports due to staleness"
+        )
+        .unwrap();
+    }
+}

signal-gateway/src/plotter.rs

@@ -0,0 +1,255 @@
+use crate::http::Alert;
+use chrono::Utc;
+use conf::Conf;
+use prom_client::{
+    AlertInfo, AlertsRequest, ExtractLabels, Labels, LabelsRequest, MetricTimeseries, MetricValue,
+    PromRequest, QueryRangeRequest, QueryRequest, SeriesRequest,
+    plot::{PlotStyle, PlotThreshold},
+};
+use rand::RngCore;
+use reqwest::Client as ReqwestClient;
+use rust_decimal::Decimal;
+use std::{error::Error, str::FromStr, time::Duration};
+use tracing::{info, warn};
+use walkdir::WalkDir;
+
+#[derive(Clone, Conf, Debug)]
+pub struct PlotterConfig {
+    #[conf(long, env)]
+    pub prometheus_host: String,
+    #[conf(long, env, default_value = "30m", value_parser = conf_extra::parse_duration)]
+    pub plot_age_limit: Duration,
+}
+
+pub struct Plotter {
+    config: PlotterConfig,
+    plot_dir: String,
+    reqwest_client: ReqwestClient,
+    skip_labels: Vec<String>,
+}
+
+impl Plotter {
+    pub fn new(config: PlotterConfig) -> Self {
+        let plot_dir = "/tmp".into();
+        let reqwest_client = ReqwestClient::new();
+        let skip_labels = vec!["job".into(), "instance".into()];
+
+        Self {
+            config,
+            plot_dir,
+            reqwest_client,
+            skip_labels,
+        }
+    }
+
+    fn create_plot(
+        &self,
+        matrix: &[MetricTimeseries],
+        threshold: Option<PlotThreshold>,
+        title: Option<&str>,
+    ) -> Result<String, Box<dyn Error>> {
+        let filename = format!(
+            "{dir}/plot-{num}.png",
+            dir = self.plot_dir,
+            num = rand::rng().next_u64()
+        );
+        let mut plot_style = PlotStyle::default().dark_mode();
+        plot_style.skip_labels = self.skip_labels.clone();
+        if let Some(title) = title {
+            plot_style.title = Some(title.to_owned());
+        }
+
+        plot_style.plot_timeseries(&filename, matrix, threshold)?;
+        Ok(filename)
+    }
+
+    pub fn purge_old_plots(&self) {
+        for entry in WalkDir::new(&self.plot_dir)
+            .min_depth(1)
+            .max_depth(1)
+            .into_iter()
+            .filter_map(|e| e.ok())
+        {
+            if !entry.file_type().is_file() {
+                continue;
+            }
+            let Some(ext) = entry.path().extension() else {
+                continue;
+            };
+            if ext != "png" && ext != ".png" {
+                continue;
+            }
+            let path = entry.path().display();
+            let Ok(metadata) = entry
+                .metadata()
+                .inspect_err(|err| warn!("Couldn't get metadata for {path}: {err}"))
+            else {
+                continue;
+            };
+            let Ok(time) = metadata
+                .created()
+                .inspect_err(|err| warn!("Couldn't get creation time for {path}: {err}"))
+            else {
+                continue;
+            };
+            let Ok(elapsed) = time
+                .elapsed()
+                .inspect_err(|err| warn!("Elapsed time calculation failed for {path}: {err}"))
+            else {
+                continue;
+            };
+            if elapsed > self.config.plot_age_limit
+                && let Err(err) = std::fs::remove_file(entry.path())
+            {
+                warn!("Couldn't remove old png file {path}: {err}");
+            }
+        }
+    }
+
+    pub async fn create_alert_plot(&self, alert: &Alert) -> Result<String, Box<dyn Error>> {
+        let expr = alert.parse_expr_from_generator_url()?;
+
+        // Parse expressions like "query < 0.09" or "query < 0.09 and on (instance) up{...}"
+        // We look for comparison operators and extract the threshold
+        let (base_query, threshold) = parse_alert_expr(&expr)?;
+
+        // Build label selector from alert labels (excluding job/instance)
+        let label_selector = build_label_selector(&alert.labels, &self.skip_labels);
+        let query = if label_selector.is_empty() {
+            base_query.to_owned()
+        } else {
+            format!("{base_query}{{{label_selector}}}")
+        };
+
+        let now = Utc::now();
+        let elapsed = now - alert.starts_at;
+        // Extend elapsed by 210%, but use at least plot_age_limit (default 30m)
+        let lengthen = elapsed
+            .checked_mul(31)
+            .and_then(|e| e.checked_div(10))
+            .ok_or("timedelta overflow")?;
+        let min_range = chrono::TimeDelta::from_std(self.config.plot_age_limit)?;
+        let range = lengthen.max(min_range);
+
+        info!("Prom range query: {query}");
+        let matrix = QueryRangeRequest::builder(query.clone())
+            .range(now - range..now)
+            .build()
+            .send_with_client(&self.reqwest_client, &self.config.prometheus_host)
+            .await?
+            .into_matrix()?;
+
+        self.create_plot(&matrix, Some(threshold), Some(&query))
+    }
+
+    pub async fn create_oneoff_plot(
+        &self,
+        query: String,
+        since: Duration,
+    ) -> Result<String, Box<dyn Error>> {
+        info!("Prom range query: {query}");
+        let matrix = QueryRangeRequest::builder(query.to_owned())
+            .since(since)
+            .build()
+            .send_with_client(&self.reqwest_client, &self.config.prometheus_host)
+            .await?
+            .into_matrix()?;
+
+        self.create_plot(&matrix, None, Some(&query))
+    }
+
+    #[allow(clippy::type_complexity)]
+    pub async fn oneoff_query(
+        &self,
+        query: String,
+    ) -> Result<(ExtractLabels, Vec<Option<(f64, Decimal)>>), Box<dyn Error>> {
+        info!("Prom query: {query}");
+        let vector: Vec<MetricValue> = QueryRequest { query, time: None }
+            .send_with_client(&self.reqwest_client, &self.config.prometheus_host)
+            .await?
+            .into_vector()?;
+
+        let labels = ExtractLabels::new(vector.iter().map(|mv| &mv.metric), &self.skip_labels);
+        let values = vector.into_iter().map(|mv| mv.value).collect();
+
+        Ok((labels, values))
+    }
+
+    pub async fn series(
+        &self,
+        matches: impl IntoIterator<Item: AsRef<str>>,
+    ) -> Result<Vec<Labels>, Box<dyn Error>> {
+        Ok(SeriesRequest {
+            matches: matches.into_iter().map(|s| s.as_ref().to_owned()).collect(),
+        }
+        .send_with_client(&self.reqwest_client, &self.config.prometheus_host)
+        .await?)
+    }
+
+    pub async fn labels(
+        &self,
+        matches: impl IntoIterator<Item: AsRef<str>>,
+    ) -> Result<Vec<String>, Box<dyn Error>> {
+        Ok(LabelsRequest {
+            matches: matches.into_iter().map(|s| s.as_ref().to_owned()).collect(),
+        }
+        .send_with_client::<()>(&self.reqwest_client, &self.config.prometheus_host)
+        .await?)
+    }
+
+    pub async fn alerts(&self) -> Result<Vec<AlertInfo>, Box<dyn Error>> {
+        Ok(AlertsRequest {}
+            .send_with_client(&self.reqwest_client, &self.config.prometheus_host)
+            .await?
+            .alerts)
+    }
+}
+
+/// Build a prometheus label selector string from alert labels, excluding specified labels.
+/// E.g., {"asset": "ETH", "job": "ec2"} with skip=["job"] -> `asset="ETH"`
+fn build_label_selector(
+    labels: &std::collections::BTreeMap<String, String>,
+    skip: &[String],
+) -> String {
+    labels
+        .iter()
+        .filter(|(k, _)| !skip.contains(k) && *k != "alertname")
+        .map(|(k, v)| format!("{k}=\"{v}\""))
+        .collect::<Vec<_>>()
+        .join(",")
+}
+
+/// Parse an alert expression to extract the base query and threshold.
+///
+/// Handles expressions like:
+/// - "query < 0.09"
+/// - "query > 100"
+/// - "rate(tick_successes[5m]) < 0.09 and on (instance) up{job=\"ec2\"}"
+///
+/// Returns (base_query, threshold) where base_query is the part before the comparator.
+fn parse_alert_expr(expr: &str) -> Result<(String, PlotThreshold), Box<dyn std::error::Error>> {
+    // Look for comparison operators with surrounding spaces
+    for comparator in [" < ", " > "] {
+        if let Some(pos) = expr.find(comparator) {
+            let base_query = expr[..pos].to_owned();
+            let after_comparator = &expr[pos + comparator.len()..];
+
+            // The threshold is the first space-delimited token after the comparator
+            let numeric = after_comparator
+                .split_whitespace()
+                .next()
+                .ok_or("no numeric after comparator")?;
+
+            let limit = f64::from_str(numeric)?;
+            let threshold = if comparator == " < " {
+                PlotThreshold::LessThan(limit)
+            } else {
+                PlotThreshold::GreaterThan(limit)
+            };
+
+            return Ok((base_query, threshold));
+        }
+    }
+
+    Err("no comparator (< or >) found in expression".into())
+}

signal-gateway/src/transports/mod.rs

@@ -0,0 +1,60 @@
+//! This copied from https://github.com/AsamK/signal-cli/blob/f9a36c6e0404d06bd396b24b5ea699e49ed29b89/client/src/jsonrpc.rs
+
+use futures_util::{Sink, SinkExt, Stream, stream::StreamExt};
+use jsonrpsee::core::client::{ReceivedMessage, TransportReceiverT, TransportSenderT};
+use thiserror::Error;
+
+mod stream_codec;
+pub mod tcp;
+
+#[derive(Debug, Error)]
+enum Errors {
+    #[error("Other: {0}")]
+    Other(String),
+    #[error("Closed")]
+    Closed,
+}
+
+struct Sender<T: Send + Sink<String>> {
+    inner: T,
+}
+
+impl<T: Send + Sink<String, Error = impl std::error::Error> + Unpin + 'static> TransportSenderT
+    for Sender<T>
+{
+    type Error = Errors;
+
+    async fn send(&mut self, body: String) -> Result<(), Self::Error> {
+        self.inner
+            .send(body)
+            .await
+            .map_err(|e| Errors::Other(format!("{e:?}")))?;
+        Ok(())
+    }
+
+    async fn close(&mut self) -> Result<(), Self::Error> {
+        self.inner
+            .close()
+            .await
+            .map_err(|e| Errors::Other(format!("{e:?}")))?;
+        Ok(())
+    }
+}
+
+struct Receiver<T: Send + Stream> {
+    inner: T,
+}
+
+impl<T: Send + Stream<Item = Result<String, std::io::Error>> + Unpin + 'static> TransportReceiverT
+    for Receiver<T>
+{
+    type Error = Errors;
+
+    async fn receive(&mut self) -> Result<ReceivedMessage, Self::Error> {
+        match self.inner.next().await {
+            None => Err(Errors::Closed),
+            Some(Ok(msg)) => Ok(ReceivedMessage::Text(msg)),
+            Some(Err(e)) => Err(Errors::Other(format!("{e:?}"))),
+        }
+    }
+}

signal-gateway/src/transports/stream_codec.rs

@@ -0,0 +1,63 @@
+//! This copied from https://github.com/AsamK/signal-cli/blob/f9a36c6e0404d06bd396b24b5ea699e49ed29b89/client/src/jsonrpc.rs
+
+use std::{io, str};
+use tokio_util::bytes::BytesMut;
+use tokio_util::codec::{Decoder, Encoder};
+
+type Separator = u8;
+
+/// Stream codec for streaming protocols (ipc, tcp)
+#[derive(Debug, Default)]
+pub struct StreamCodec {
+    incoming_separator: Separator,
+    outgoing_separator: Separator,
+}
+
+impl StreamCodec {
+    /// Default codec with streaming input data. Input can be both enveloped and not.
+    pub fn stream_incoming() -> Self {
+        StreamCodec::new(b'\n', b'\n')
+    }
+
+    /// New custom stream codec
+    pub fn new(incoming_separator: Separator, outgoing_separator: Separator) -> Self {
+        StreamCodec {
+            incoming_separator,
+            outgoing_separator,
+        }
+    }
+}
+
+impl Decoder for StreamCodec {
+    type Item = String;
+    type Error = io::Error;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Item>> {
+        if let Some(i) = buf
+            .as_ref()
+            .iter()
+            .position(|&b| b == self.incoming_separator)
+        {
+            let line = buf.split_to(i);
+            let _ = buf.split_to(1);
+
+            match str::from_utf8(line.as_ref()) {
+                Ok(s) => Ok(Some(s.to_string())),
+                Err(_) => Err(io::Error::other("invalid UTF-8")),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+impl Encoder<String> for StreamCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, msg: String, buf: &mut BytesMut) -> io::Result<()> {
+        let mut payload = msg.into_bytes();
+        payload.push(self.outgoing_separator);
+        buf.extend_from_slice(&payload);
+        Ok(())
+    }
+}

signal-gateway/src/transports/tcp.rs

@@ -0,0 +1,22 @@
+use std::io::Error;
+
+use futures_util::stream::StreamExt;
+use jsonrpsee::core::client::{TransportReceiverT, TransportSenderT};
+use tokio::net::{TcpStream, ToSocketAddrs};
+use tokio_util::codec::Decoder;
+
+use super::stream_codec::StreamCodec;
+use super::{Receiver, Sender};
+
+/// Connect to a JSON-RPC TCP server.
+pub async fn connect(
+    socket: impl ToSocketAddrs,
+) -> Result<(impl TransportSenderT + Send, impl TransportReceiverT + Send), Error> {
+    let connection = TcpStream::connect(socket).await?;
+    let (sink, stream) = StreamCodec::stream_incoming().framed(connection).split();
+
+    let sender = Sender { inner: sink };
+    let receiver = Receiver { inner: stream };
+
+    Ok((sender, receiver))
+}