Add native MQTT alerter endpoint

Cargo.toml

@@ -53,6 +53,7 @@ svi = "1.2.0"
 
 # ASYNC
 reqwest = { version = "0.13.3", default-features = false, features = ["json", "stream", "form", "query", "rustls"] }
+rumqttc = "0.25.0"
 tokio = { version = "1.52.2", features = ["full"] }
 tokio-util = { version = "0.7.18", features = ["io", "codec"] }
 tokio-stream = { version = "0.1.18", features = ["sync"] }
@@ -137,4 +138,4 @@ bytes = "1.11.1"
 regex = "1.12.3"
 
 [profile.release]
-strip = "debuginfo"
+strip = "debuginfo"

MQTT_ALERTER_PLAN.md

@@ -0,0 +1,39 @@
+# MQTT Alerter Plan
+
+## Minimum Files To Modify
+
+- `client/core/rs/src/entities/alerter.rs`
+  - Add a new `AlerterEndpoint::Mqtt` variant and `MqttAlerterEndpoint` config struct.
+  - Define defaults (topic `komodo/events`, QoS default) and optional auth/client fields.
+  - Keeps API/resource model aligned for backend + UI.
+
+- `client/core/ts/src/types.ts`
+  - Add generated-type equivalents for the new Rust endpoint variant and params.
+  - Ensures UI can type-check new MQTT config fields.
+
+- `bin/core/src/alert/mod.rs`
+  - Register `mod mqtt;`.
+  - Add endpoint dispatch arm for `AlerterEndpoint::Mqtt`.
+
+- `bin/core/src/alert/mqtt.rs` (new)
+  - Implement MQTT transport: connect, serialize existing `Alert` JSON unchanged, publish, and return success/failure.
+
+- `Cargo.toml` and `bin/core/Cargo.toml`
+  - Add MQTT client crate dependency needed by core alert transport.
+
+- `ui/src/resources/alerter/config/endpoint.tsx`
+  - Add `MQTT` to endpoint selector.
+  - Render MQTT-specific config fields: broker URL, topic, username, password, client ID, QoS, retain.
+
+- `docsite/docs/resources.md`
+  - Small docs update to include MQTT as a native Alerter destination.
+
+## Recommended Rust MQTT Library
+
+Use `rumqttc`.
+
+Why:
+- Pure Rust + Tokio-native async flow (fits Komodo core runtime).
+- Lightweight for one-shot publish operations.
+- Actively used in Rust MQTT integrations.
+- Supports auth, QoS levels, retain flag, and broker URL parsing patterns we need.

bin/core/Cargo.toml

@@ -64,6 +64,7 @@ serde_qs.workspace = true
 colored.workspace = true
 tracing.workspace = true
 reqwest.workspace = true
+rumqttc.workspace = true
 dotenvy.workspace = true
 anyhow.workspace = true
 bcrypt.workspace = true
@@ -83,4 +84,4 @@ envy.workspace = true
 hmac.workspace = true
 sha2.workspace = true
 hex.workspace = true
-url.workspace = true
+url.workspace = true

bin/core/src/alert/mod.rs

@@ -18,6 +18,7 @@ use crate::helpers::{
 use crate::{config::core_config, state::db_client};
 
 mod discord;
+mod mqtt;
 mod ntfy;
 mod pushover;
 mod slack;
@@ -148,6 +149,14 @@ pub async fn send_alert_to_alerter(
         )
       })
     }
+    AlerterEndpoint::Mqtt(mqtt_endpoint) => mqtt::send_alert(
+      mqtt_endpoint,
+      alert,
+    )
+    .await
+    .with_context(|| {
+      format!("Failed to send alert to MQTT Alerter {}", alerter.name)
+    }),
   }
 }
 

bin/core/src/alert/mqtt.rs

@@ -0,0 +1,162 @@
+use std::time::Duration;
+
+use anyhow::{Context, anyhow, bail};
+use komodo_client::entities::alert::AlertDataVariant;
+use rumqttc::{AsyncClient, MqttOptions, Packet, QoS};
+use url::Url;
+
+use super::*;
+
+pub async fn send_alert(
+  endpoint: &MqttAlerterEndpoint,
+  alert: &Alert,
+) -> anyhow::Result<()> {
+  let VariablesAndSecrets { variables, secrets } =
+    get_variables_and_secrets().await?;
+
+  let mut broker_url = endpoint.broker_url.clone();
+  let mut topic = endpoint.topic.clone();
+  let mut username = endpoint.username.clone();
+  let mut password = endpoint.password.clone();
+  let mut client_id = endpoint.client_id.clone();
+
+  let mut interpolator =
+    Interpolator::new(Some(&variables), &secrets);
+
+  let res = async {
+    interpolator.interpolate_string(&mut broker_url)?;
+    interpolator.interpolate_string(&mut topic)?;
+    if let Some(value) = username.as_mut() {
+      interpolator.interpolate_string(value)?;
+    }
+    if let Some(value) = password.as_mut() {
+      interpolator.interpolate_string(value)?;
+    }
+    if let Some(value) = client_id.as_mut() {
+      interpolator.interpolate_string(value)?;
+    }
+    let topic = topic;
+
+    send_message(
+      &broker_url,
+      &topic,
+      username.as_deref(),
+      password.as_deref(),
+      client_id.as_deref(),
+      endpoint.qos,
+      endpoint.retain,
+      alert,
+    )
+    .await
+  }
+  .await;
+
+  res.map_err(|e| {
+    let replacers = interpolator
+      .secret_replacers
+      .into_iter()
+      .collect::<Vec<_>>();
+    let sanitized_error =
+      svi::replace_in_string(&format!("{e:?}"), &replacers);
+    anyhow!("Error with publish to MQTT: {sanitized_error}")
+  })
+}
+
+async fn send_message(
+  broker_url: &str,
+  topic: &str,
+  username: Option<&str>,
+  password: Option<&str>,
+  client_id: Option<&str>,
+  qos: u8,
+  retain: bool,
+  alert: &Alert,
+) -> anyhow::Result<()> {
+  if topic.trim().is_empty() {
+    bail!("MQTT topic cannot be empty");
+  }
+
+  let parsed = Url::parse(broker_url).with_context(|| {
+    format!("Invalid MQTT broker URL: {broker_url}")
+  })?;
+  let scheme = parsed.scheme();
+  if !matches!(scheme, "mqtt" | "tcp") {
+    bail!(
+      "Unsupported MQTT broker URL scheme `{scheme}`. Use mqtt:// or tcp://"
+    );
+  }
+
+  let host = parsed
+    .host_str()
+    .context("MQTT broker URL must include a host")?;
+  let port = parsed.port().unwrap_or(1883);
+
+  let client_id = client_id
+    .filter(|id| !id.trim().is_empty())
+    .map(ToOwned::to_owned)
+    .unwrap_or_else(|| format!("komodo-alert-{}", uuid::Uuid::new_v4()));
+
+  let mut options = MqttOptions::new(client_id, host, port);
+  options.set_keep_alive(Duration::from_secs(10));
+
+  let username = username.filter(|value| !value.trim().is_empty());
+  let password = password.filter(|value| !value.trim().is_empty());
+  let url_username = parsed.username();
+  let url_password = parsed.password().filter(|value| !value.is_empty());
+  match (username, password) {
+    (Some(user), Some(pass)) => {
+      options.set_credentials(user, pass);
+    }
+    (Some(user), None) => {
+      options.set_credentials(user, "");
+    }
+    (None, Some(pass)) => {
+      options.set_credentials("", pass);
+    }
+    (None, None) if !url_username.is_empty() => {
+      options.set_credentials(url_username, url_password.unwrap_or(""));
+    }
+    _ => {}
+  }
+
+  let payload = serde_json::to_vec(alert)
+    .context("Failed to serialize alert payload to JSON")?;
+  let qos = to_qos(qos)?;
+
+  let (client, mut eventloop) = AsyncClient::new(options, 10);
+  client
+    .publish(topic, qos, retain, payload)
+    .await
+    .context("Failed queuing MQTT publish")?;
+
+  let poll_count = if qos == QoS::AtMostOnce { 2 } else { 4 };
+  for _ in 0..poll_count {
+    let event = tokio::time::timeout(
+      Duration::from_secs(5),
+      eventloop.poll(),
+    )
+    .await
+    .context("Timed out waiting for MQTT broker response")?
+    .context("MQTT event loop failed while publishing")?;
+
+    if matches!(
+      event,
+      rumqttc::Event::Incoming(Packet::PubAck(_))
+        | rumqttc::Event::Incoming(Packet::PubComp(_))
+    ) {
+      break;
+    }
+  }
+
+  client.disconnect().await.ok();
+  Ok(())
+}
+
+fn to_qos(qos: u8) -> anyhow::Result<QoS> {
+  match qos {
+    0 => Ok(QoS::AtMostOnce),
+    1 => Ok(QoS::AtLeastOnce),
+    2 => Ok(QoS::ExactlyOnce),
+    _ => bail!("Invalid MQTT QoS `{qos}`. Must be 0, 1, or 2"),
+  }
+}

client/core/rs/src/entities/alerter.rs

@@ -170,6 +170,9 @@ pub enum AlerterEndpoint {
 
   /// Send alert to Pushover
   Pushover(PushoverAlerterEndpoint),
+
+  /// Send alert JSON to an MQTT broker
+  Mqtt(MqttAlerterEndpoint),
 }
 
 impl Default for AlerterEndpoint {
@@ -319,6 +322,66 @@ fn default_pushover_url() -> String {
   )
 }
 
+/// Configuration for an MQTT alerter.
+#[typeshare]
+#[derive(
+  Debug, Clone, PartialEq, Serialize, Deserialize, Builder,
+)]
+#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
+#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
+pub struct MqttAlerterEndpoint {
+  /// MQTT broker URL. Example: `mqtt://localhost:1883`
+  #[serde(default = "default_mqtt_broker_url")]
+  #[builder(default = "default_mqtt_broker_url()")]
+  pub broker_url: String,
+
+  /// Topic to publish alerts to
+  #[serde(default = "default_mqtt_topic")]
+  #[builder(default = "default_mqtt_topic()")]
+  pub topic: String,
+
+  /// Optional username for broker authentication
+  pub username: Option<String>,
+
+  /// Optional password for broker authentication
+  pub password: Option<String>,
+
+  /// Optional client identifier. If empty, core will generate one.
+  pub client_id: Option<String>,
+
+  /// MQTT QoS level: 0, 1, or 2
+  #[serde(default)]
+  #[builder(default)]
+  pub qos: u8,
+
+  /// Whether the broker should retain the message
+  #[serde(default)]
+  #[builder(default)]
+  pub retain: bool,
+}
+
+impl Default for MqttAlerterEndpoint {
+  fn default() -> Self {
+    Self {
+      broker_url: default_mqtt_broker_url(),
+      topic: default_mqtt_topic(),
+      username: None,
+      password: None,
+      client_id: None,
+      qos: 0,
+      retain: false,
+    }
+  }
+}
+
+fn default_mqtt_broker_url() -> String {
+  String::from("mqtt://localhost:1883")
+}
+
+fn default_mqtt_topic() -> String {
+  String::from("komodo/events")
+}
+
 // QUERY
 #[typeshare]
 pub type AlerterQuery = ResourceQuery<AlerterQuerySpecifics>;

client/core/ts/src/types.ts

@@ -227,7 +227,9 @@ export type AlerterEndpoint =
 	/** Send alert to Ntfy */
 	| { type: "Ntfy", params: NtfyAlerterEndpoint }
 	/** Send alert to Pushover */
-	| { type: "Pushover", params: PushoverAlerterEndpoint };
+	| { type: "Pushover", params: PushoverAlerterEndpoint }
+	/** Send alert JSON to an MQTT broker */
+	| { type: "Mqtt", params: MqttAlerterEndpoint };
 
 /** Used to reference a specific resource across all resource types */
 export type ResourceTarget = 
@@ -9150,6 +9152,24 @@ export interface NtfyAlerterEndpoint {
 	email?: string;
 }
 
+/** Configuration for an MQTT alerter. */
+export interface MqttAlerterEndpoint {
+	/** MQTT broker URL. Example: `mqtt://localhost:1883` */
+	broker_url: string;
+	/** Topic to publish alerts to */
+	topic: string;
+	/** Optional username for broker authentication */
+	username?: string;
+	/** Optional password for broker authentication */
+	password?: string;
+	/** Optional client identifier. If empty, core will generate one. */
+	client_id?: string;
+	/** MQTT QoS level: 0, 1, or 2 */
+	qos: number;
+	/** Whether the broker should retain the message */
+	retain: boolean;
+}
+
 /** Pauses all containers on the target server. Response: [Update] */
 export interface PauseAllContainers {
 	/** Name or id */

docsite/docs/resources.md

@@ -70,4 +70,5 @@ All resources which depend on git repos / docker registries are able to use thes
 ## Alerter
 
 - Route alerts to various endpoints.
+- Native endpoints include `Discord`, `Slack`, `Ntfy`, `Pushover`, `MQTT`, and `Custom`.
 - Can configure rules on each Alerter, such as resource whitelist, blacklist, or alert type filter.