Initial mqtt implementation

app/AGENTS.md

@@ -27,6 +27,7 @@ app/
 ├── packet_processor.py  # Raw packet pipeline, dedup, path handling
 ├── event_handlers.py    # MeshCore event subscriptions and ACK tracking
 ├── websocket.py         # WS manager + broadcast helpers
+├── mqtt.py              # Optional MQTT publisher (fire-and-forget forwarding)
 ├── bot.py               # Bot execution and outbound bot sends
 ├── dependencies.py      # Shared FastAPI dependency providers
 ├── keystore.py          # Ephemeral private/public key storage for DM decryption
@@ -99,6 +100,19 @@ app/
 - `0` means disabled.
 - Last send time tracked in `app_settings.last_advert_time`.
 
+### MQTT publishing
+
+- Optional forwarding of mesh events to an external MQTT broker.
+- All config in `app_settings` (not env vars): `mqtt_broker_host`, `mqtt_broker_port`, `mqtt_username`, `mqtt_password`, `mqtt_use_tls`, `mqtt_topic_prefix`, `mqtt_publish_messages`, `mqtt_publish_raw_packets`.
+- Disabled when `mqtt_broker_host` is empty.
+- `broadcast_event()` in `websocket.py` calls `mqtt_broadcast()` — single hook covers all message and raw_packet events.
+- `MqttPublisher` (`app/mqtt.py`) runs a background connection loop with auto-reconnect and exponential backoff (5s → 30s).
+- Publishes are fire-and-forget; individual publish failures logged but not surfaced to users.
+- Connection state changes surface via `broadcast_error`/`broadcast_success` toasts.
+- Health endpoint includes `mqtt_status` field (`connected`, `disconnected`, `disabled`).
+- Settings changes trigger `mqtt_publisher.restart()` — no server restart needed.
+- Topics: `{prefix}/dm:{key}`, `{prefix}/gm:{key}`, `{prefix}/raw/dm:{key}`, `{prefix}/raw/gm:{key}`, `{prefix}/raw/unrouted`.
+
 ## API Surface (all under `/api`)
 
 ### Health
@@ -204,6 +218,8 @@ Main tables:
 - `advert_interval`
 - `last_advert_time`
 - `bots`
+- `mqtt_broker_host`, `mqtt_broker_port`, `mqtt_username`, `mqtt_password`
+- `mqtt_use_tls`, `mqtt_tls_insecure`, `mqtt_topic_prefix`, `mqtt_publish_messages`, `mqtt_publish_raw_packets`
 
 ## Security Posture (intentional)
 
@@ -239,6 +255,7 @@ tests/
 ├── test_message_pagination.py  # Cursor-based message pagination
 ├── test_message_prefix_claim.py # Message prefix claim logic
 ├── test_migrations.py          # Schema migration system
+├── test_mqtt.py                # MQTT publisher topic routing and lifecycle
 ├── test_packet_pipeline.py     # End-to-end packet processing
 ├── test_radio.py               # RadioManager, serial detection
 ├── test_radio_operation.py     # radio_operation() context manager

app/main.py

@@ -56,9 +56,20 @@ async def lifespan(app: FastAPI):
     # Always start connection monitor (even if initial connection failed)
     await radio_manager.start_connection_monitor()
 
+    # Start MQTT publisher if configured
+    from app.mqtt import mqtt_publisher
+    from app.repository import AppSettingsRepository
+
+    try:
+        mqtt_settings = await AppSettingsRepository.get()
+        await mqtt_publisher.start(mqtt_settings)
+    except Exception as e:
+        logger.warning("Failed to start MQTT publisher: %s", e)
+
     yield
 
     logger.info("Shutting down")
+    await mqtt_publisher.stop()
     await radio_manager.stop_connection_monitor()
     await stop_message_polling()
     await stop_periodic_advert()

app/migrations.py

@@ -247,6 +247,13 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
         await set_version(conn, 30)
         applied += 1
 
+    # Migration 31: Add MQTT configuration columns to app_settings
+    if version < 31:
+        logger.info("Applying migration 31: add MQTT columns to app_settings")
+        await _migrate_031_add_mqtt_columns(conn)
+        await set_version(conn, 31)
+        applied += 1
+
     if applied > 0:
         logger.info(
             "Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
@@ -1852,3 +1859,35 @@ async def _migrate_030_add_pagination_index(conn: aiosqlite.Connection) -> None:
         )
         await conn.execute("DROP INDEX IF EXISTS idx_messages_conversation")
     await conn.commit()
+
+
+async def _migrate_031_add_mqtt_columns(conn: aiosqlite.Connection) -> None:
+    """Add MQTT configuration columns to app_settings."""
+    # Guard: app_settings may not exist in partial-schema test setups
+    cursor = await conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table' AND name='app_settings'"
+    )
+    if not await cursor.fetchone():
+        await conn.commit()
+        return
+
+    cursor = await conn.execute("PRAGMA table_info(app_settings)")
+    columns = {row[1] for row in await cursor.fetchall()}
+
+    new_columns = [
+        ("mqtt_broker_host", "TEXT DEFAULT ''"),
+        ("mqtt_broker_port", "INTEGER DEFAULT 1883"),
+        ("mqtt_username", "TEXT DEFAULT ''"),
+        ("mqtt_password", "TEXT DEFAULT ''"),
+        ("mqtt_use_tls", "INTEGER DEFAULT 0"),
+        ("mqtt_tls_insecure", "INTEGER DEFAULT 0"),
+        ("mqtt_topic_prefix", "TEXT DEFAULT 'meshcore'"),
+        ("mqtt_publish_messages", "INTEGER DEFAULT 0"),
+        ("mqtt_publish_raw_packets", "INTEGER DEFAULT 0"),
+    ]
+
+    for col_name, col_def in new_columns:
+        if col_name not in columns:
+            await conn.execute(f"ALTER TABLE app_settings ADD COLUMN {col_name} {col_def}")
+
+    await conn.commit()

app/models.py

@@ -173,6 +173,8 @@ class RawPacketDecryptedInfo(BaseModel):
 
     channel_name: str | None = None
     sender: str | None = None
+    channel_key: str | None = None
+    contact_key: str | None = None
 
 
 class RawPacketBroadcast(BaseModel):
@@ -425,6 +427,42 @@ class AppSettings(BaseModel):
         default_factory=list,
         description="List of bot configurations",
     )
+    mqtt_broker_host: str = Field(
+        default="",
+        description="MQTT broker hostname (empty = disabled)",
+    )
+    mqtt_broker_port: int = Field(
+        default=1883,
+        description="MQTT broker port",
+    )
+    mqtt_username: str = Field(
+        default="",
+        description="MQTT username (optional)",
+    )
+    mqtt_password: str = Field(
+        default="",
+        description="MQTT password (optional)",
+    )
+    mqtt_use_tls: bool = Field(
+        default=False,
+        description="Whether to use TLS for MQTT connection",
+    )
+    mqtt_tls_insecure: bool = Field(
+        default=False,
+        description="Skip TLS certificate verification (for self-signed certs)",
+    )
+    mqtt_topic_prefix: str = Field(
+        default="meshcore",
+        description="MQTT topic prefix",
+    )
+    mqtt_publish_messages: bool = Field(
+        default=False,
+        description="Whether to publish decrypted messages to MQTT",
+    )
+    mqtt_publish_raw_packets: bool = Field(
+        default=False,
+        description="Whether to publish raw packets to MQTT",
+    )
 
 
 class BusyChannel(BaseModel):

app/mqtt.py

@@ -0,0 +1,223 @@
+"""MQTT publisher for forwarding mesh network events to an MQTT broker."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import ssl
+from typing import Any
+
+import aiomqtt
+
+from app.models import AppSettings
+
+logger = logging.getLogger(__name__)
+
+# Reconnect backoff: start at 5s, cap at 30s
+_BACKOFF_MIN = 5
+_BACKOFF_MAX = 30
+
+
+class MqttPublisher:
+    """Manages an MQTT connection and publishes mesh network events."""
+
+    def __init__(self) -> None:
+        self._client: aiomqtt.Client | None = None
+        self._task: asyncio.Task[None] | None = None
+        self._settings: AppSettings | None = None
+        self._settings_version: int = 0
+        self._version_event: asyncio.Event = asyncio.Event()
+        self.connected: bool = False
+
+    async def start(self, settings: AppSettings) -> None:
+        """Start the background connection loop."""
+        self._settings = settings
+        self._settings_version += 1
+        self._version_event.set()
+        if self._task is None or self._task.done():
+            self._task = asyncio.create_task(self._connection_loop())
+
+    async def stop(self) -> None:
+        """Cancel the background task and disconnect."""
+        if self._task and not self._task.done():
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+        self._task = None
+        self._client = None
+        self.connected = False
+
+    async def restart(self, settings: AppSettings) -> None:
+        """Called when MQTT settings change — stop + start."""
+        await self.stop()
+        await self.start(settings)
+
+    async def publish(self, topic: str, payload: dict[str, Any]) -> None:
+        """Publish a JSON payload. Drops silently if not connected."""
+        if self._client is None or not self.connected:
+            return
+        try:
+            await self._client.publish(topic, json.dumps(payload))
+        except Exception as e:
+            logger.warning("MQTT publish failed on %s: %s", topic, e)
+            self.connected = False
+            # Wake the connection loop so it exits the wait and reconnects
+            self._settings_version += 1
+            self._version_event.set()
+
+    def _mqtt_configured(self) -> bool:
+        """Check if MQTT is configured (broker host is set)."""
+        return bool(self._settings and self._settings.mqtt_broker_host)
+
+    async def _connection_loop(self) -> None:
+        """Background loop: connect, wait, reconnect on failure."""
+        from app.websocket import broadcast_error, broadcast_success
+
+        backoff = _BACKOFF_MIN
+
+        while True:
+            if not self._mqtt_configured():
+                self.connected = False
+                self._client = None
+                # Wait until settings change (which might configure MQTT)
+                self._version_event.clear()
+                try:
+                    await self._version_event.wait()
+                except asyncio.CancelledError:
+                    return
+                continue
+
+            settings = self._settings
+            assert settings is not None  # guaranteed by _mqtt_configured()
+            version_at_connect = self._settings_version
+
+            try:
+                tls_context = self._build_tls_context(settings)
+
+                async with aiomqtt.Client(
+                    hostname=settings.mqtt_broker_host,
+                    port=settings.mqtt_broker_port,
+                    username=settings.mqtt_username or None,
+                    password=settings.mqtt_password or None,
+                    tls_context=tls_context,
+                ) as client:
+                    self._client = client
+                    self.connected = True
+                    backoff = _BACKOFF_MIN
+
+                    broadcast_success(
+                        "MQTT connected",
+                        f"{settings.mqtt_broker_host}:{settings.mqtt_broker_port}",
+                    )
+                    _broadcast_mqtt_health()
+
+                    # Wait until cancelled or settings version changes.
+                    # The 60s timeout is a housekeeping wake-up; actual connection
+                    # liveness is handled by paho-mqtt's keepalive mechanism.
+                    while self._settings_version == version_at_connect:
+                        self._version_event.clear()
+                        try:
+                            await asyncio.wait_for(self._version_event.wait(), timeout=60)
+                        except asyncio.TimeoutError:
+                            continue
+
+            except asyncio.CancelledError:
+                self.connected = False
+                self._client = None
+                return
+
+            except Exception as e:
+                self.connected = False
+                self._client = None
+
+                broadcast_error(
+                    "MQTT connection failure",
+                    "Please correct the settings or disable.",
+                )
+                _broadcast_mqtt_health()
+                logger.warning("MQTT connection error: %s (reconnecting in %ds)", e, backoff)
+
+                try:
+                    await asyncio.sleep(backoff)
+                except asyncio.CancelledError:
+                    return
+                backoff = min(backoff * 2, _BACKOFF_MAX)
+
+    @staticmethod
+    def _build_tls_context(settings: AppSettings) -> ssl.SSLContext | None:
+        """Build TLS context from settings, or None if TLS is disabled."""
+        if not settings.mqtt_use_tls:
+            return None
+        ctx = ssl.create_default_context()
+        if settings.mqtt_tls_insecure:
+            ctx.check_hostname = False
+            ctx.verify_mode = ssl.CERT_NONE
+        return ctx
+
+
+# Module-level singleton
+mqtt_publisher = MqttPublisher()
+
+
+def _broadcast_mqtt_health() -> None:
+    """Push updated health (including mqtt_status) to all WS clients."""
+    from app.radio import radio_manager
+    from app.websocket import broadcast_health
+
+    broadcast_health(radio_manager.is_connected, radio_manager.connection_info)
+
+
+def mqtt_broadcast(event_type: str, data: dict[str, Any]) -> None:
+    """Fire-and-forget MQTT publish, matching broadcast_event's pattern."""
+    if event_type not in ("message", "raw_packet"):
+        return
+    if not mqtt_publisher.connected or mqtt_publisher._settings is None:
+        return
+    asyncio.create_task(_mqtt_maybe_publish(event_type, data))
+
+
+async def _mqtt_maybe_publish(event_type: str, data: dict[str, Any]) -> None:
+    """Check settings and build topic, then publish."""
+    settings = mqtt_publisher._settings
+    if settings is None:
+        return
+
+    try:
+        if event_type == "message" and settings.mqtt_publish_messages:
+            topic = _build_message_topic(settings.mqtt_topic_prefix, data)
+            await mqtt_publisher.publish(topic, data)
+
+        elif event_type == "raw_packet" and settings.mqtt_publish_raw_packets:
+            topic = _build_raw_packet_topic(settings.mqtt_topic_prefix, data)
+            await mqtt_publisher.publish(topic, data)
+
+    except Exception as e:
+        logger.warning("MQTT broadcast error: %s", e)
+
+
+def _build_message_topic(prefix: str, data: dict[str, Any]) -> str:
+    """Build MQTT topic for a decrypted message."""
+    msg_type = data.get("type", "")
+    conversation_key = data.get("conversation_key", "unknown")
+
+    if msg_type == "PRIV":
+        return f"{prefix}/dm:{conversation_key}"
+    elif msg_type == "CHAN":
+        return f"{prefix}/gm:{conversation_key}"
+    return f"{prefix}/message:{conversation_key}"
+
+
+def _build_raw_packet_topic(prefix: str, data: dict[str, Any]) -> str:
+    """Build MQTT topic for a raw packet."""
+    info = data.get("decrypted_info")
+    if info and isinstance(info, dict):
+        contact_key = info.get("contact_key")
+        channel_key = info.get("channel_key")
+        if contact_key:
+            return f"{prefix}/raw/dm:{contact_key}"
+        if channel_key:
+            return f"{prefix}/raw/gm:{channel_key}"
+    return f"{prefix}/raw/unrouted"

app/packet_processor.py

@@ -576,6 +576,8 @@ async def process_raw_packet(
         decrypted_info=RawPacketDecryptedInfo(
             channel_name=result["channel_name"],
             sender=result["sender"],
+            channel_key=result.get("channel_key"),
+            contact_key=result.get("contact_key"),
         )
         if result["decrypted"]
         else None,
@@ -632,6 +634,7 @@ async def _process_group_text(
             "channel_name": channel.name,
             "sender": decrypted.sender,
             "message_id": msg_id,  # None if duplicate, msg_id if new
+            "channel_key": channel.key,
         }
 
     # Couldn't decrypt with any known key
@@ -888,6 +891,7 @@ async def _process_direct_message(
                 "contact_name": contact.name,
                 "sender": contact.name or contact.public_key[:12],
                 "message_id": msg_id,
+                "contact_key": contact.public_key,
             }
 
     # Couldn't decrypt with any known contact

app/repository/settings.py

@@ -26,7 +26,10 @@ class AppSettingsRepository:
             """
             SELECT max_radio_contacts, favorites, auto_decrypt_dm_on_advert,
                    sidebar_sort_order, last_message_times, preferences_migrated,
-                   advert_interval, last_advert_time, bots
+                   advert_interval, last_advert_time, bots,
+                   mqtt_broker_host, mqtt_broker_port, mqtt_username, mqtt_password,
+                   mqtt_use_tls, mqtt_tls_insecure, mqtt_topic_prefix,
+                   mqtt_publish_messages, mqtt_publish_raw_packets
             FROM app_settings WHERE id = 1
             """
         )
@@ -91,6 +94,15 @@ class AppSettingsRepository:
             advert_interval=row["advert_interval"] or 0,
             last_advert_time=row["last_advert_time"] or 0,
             bots=bots,
+            mqtt_broker_host=row["mqtt_broker_host"] or "",
+            mqtt_broker_port=row["mqtt_broker_port"] or 1883,
+            mqtt_username=row["mqtt_username"] or "",
+            mqtt_password=row["mqtt_password"] or "",
+            mqtt_use_tls=bool(row["mqtt_use_tls"]),
+            mqtt_tls_insecure=bool(row["mqtt_tls_insecure"]),
+            mqtt_topic_prefix=row["mqtt_topic_prefix"] or "meshcore",
+            mqtt_publish_messages=bool(row["mqtt_publish_messages"]),
+            mqtt_publish_raw_packets=bool(row["mqtt_publish_raw_packets"]),
         )
 
     @staticmethod
@@ -104,6 +116,15 @@ class AppSettingsRepository:
         advert_interval: int | None = None,
         last_advert_time: int | None = None,
         bots: list[BotConfig] | None = None,
+        mqtt_broker_host: str | None = None,
+        mqtt_broker_port: int | None = None,
+        mqtt_username: str | None = None,
+        mqtt_password: str | None = None,
+        mqtt_use_tls: bool | None = None,
+        mqtt_tls_insecure: bool | None = None,
+        mqtt_topic_prefix: str | None = None,
+        mqtt_publish_messages: bool | None = None,
+        mqtt_publish_raw_packets: bool | None = None,
     ) -> AppSettings:
         """Update app settings. Only provided fields are updated."""
         updates = []
@@ -147,6 +168,42 @@ class AppSettingsRepository:
             bots_json = json.dumps([b.model_dump() for b in bots])
             params.append(bots_json)
 
+        if mqtt_broker_host is not None:
+            updates.append("mqtt_broker_host = ?")
+            params.append(mqtt_broker_host)
+
+        if mqtt_broker_port is not None:
+            updates.append("mqtt_broker_port = ?")
+            params.append(mqtt_broker_port)
+
+        if mqtt_username is not None:
+            updates.append("mqtt_username = ?")
+            params.append(mqtt_username)
+
+        if mqtt_password is not None:
+            updates.append("mqtt_password = ?")
+            params.append(mqtt_password)
+
+        if mqtt_use_tls is not None:
+            updates.append("mqtt_use_tls = ?")
+            params.append(1 if mqtt_use_tls else 0)
+
+        if mqtt_tls_insecure is not None:
+            updates.append("mqtt_tls_insecure = ?")
+            params.append(1 if mqtt_tls_insecure else 0)
+
+        if mqtt_topic_prefix is not None:
+            updates.append("mqtt_topic_prefix = ?")
+            params.append(mqtt_topic_prefix)
+
+        if mqtt_publish_messages is not None:
+            updates.append("mqtt_publish_messages = ?")
+            params.append(1 if mqtt_publish_messages else 0)
+
+        if mqtt_publish_raw_packets is not None:
+            updates.append("mqtt_publish_raw_packets = ?")
+            params.append(1 if mqtt_publish_raw_packets else 0)
+
         if updates:
             query = f"UPDATE app_settings SET {', '.join(updates)} WHERE id = 1"
             await db.conn.execute(query, params)

app/routers/health.py

@@ -16,6 +16,7 @@ class HealthResponse(BaseModel):
     connection_info: str | None
     database_size_mb: float
     oldest_undecrypted_timestamp: int | None
+    mqtt_status: str | None = None
 
 
 async def build_health_data(radio_connected: bool, connection_info: str | None) -> dict:
@@ -33,12 +34,25 @@ async def build_health_data(radio_connected: bool, connection_info: str | None)
     except RuntimeError:
         pass  # Database not connected
 
+    # MQTT status
+    mqtt_status: str | None = None
+    try:
+        from app.mqtt import mqtt_publisher
+
+        if mqtt_publisher._mqtt_configured():
+            mqtt_status = "connected" if mqtt_publisher.connected else "disconnected"
+        else:
+            mqtt_status = "disabled"
+    except Exception:
+        pass
+
     return {
         "status": "ok" if radio_connected else "degraded",
         "radio_connected": radio_connected,
         "connection_info": connection_info,
         "database_size_mb": db_size_mb,
         "oldest_undecrypted_timestamp": oldest_ts,
+        "mqtt_status": mqtt_status,
     }
 
 

app/routers/settings.py

@@ -59,6 +59,44 @@ class AppSettingsUpdate(BaseModel):
         default=None,
         description="List of bot configurations",
     )
+    mqtt_broker_host: str | None = Field(
+        default=None,
+        description="MQTT broker hostname (empty = disabled)",
+    )
+    mqtt_broker_port: int | None = Field(
+        default=None,
+        ge=1,
+        le=65535,
+        description="MQTT broker port",
+    )
+    mqtt_username: str | None = Field(
+        default=None,
+        description="MQTT username (optional)",
+    )
+    mqtt_password: str | None = Field(
+        default=None,
+        description="MQTT password (optional)",
+    )
+    mqtt_use_tls: bool | None = Field(
+        default=None,
+        description="Whether to use TLS for MQTT connection",
+    )
+    mqtt_tls_insecure: bool | None = Field(
+        default=None,
+        description="Skip TLS certificate verification (for self-signed certs)",
+    )
+    mqtt_topic_prefix: str | None = Field(
+        default=None,
+        description="MQTT topic prefix",
+    )
+    mqtt_publish_messages: bool | None = Field(
+        default=None,
+        description="Whether to publish decrypted messages to MQTT",
+    )
+    mqtt_publish_raw_packets: bool | None = Field(
+        default=None,
+        description="Whether to publish raw packets to MQTT",
+    )
 
 
 class FavoriteRequest(BaseModel):
@@ -124,8 +162,35 @@ async def update_settings(update: AppSettingsUpdate) -> AppSettings:
         logger.info("Updating bots (count=%d)", len(update.bots))
         kwargs["bots"] = update.bots
 
+    # MQTT fields
+    mqtt_fields = [
+        "mqtt_broker_host",
+        "mqtt_broker_port",
+        "mqtt_username",
+        "mqtt_password",
+        "mqtt_use_tls",
+        "mqtt_tls_insecure",
+        "mqtt_topic_prefix",
+        "mqtt_publish_messages",
+        "mqtt_publish_raw_packets",
+    ]
+    mqtt_changed = False
+    for field in mqtt_fields:
+        value = getattr(update, field)
+        if value is not None:
+            kwargs[field] = value
+            mqtt_changed = True
+
     if kwargs:
-        return await AppSettingsRepository.update(**kwargs)
+        result = await AppSettingsRepository.update(**kwargs)
+
+        # Restart MQTT publisher if any MQTT settings changed
+        if mqtt_changed:
+            from app.mqtt import mqtt_publisher
+
+            await mqtt_publisher.restart(result)
+
+        return result
 
     return await AppSettingsRepository.get()
 

app/websocket.py

@@ -96,10 +96,14 @@ def broadcast_event(event_type: str, data: dict) -> None:
     """Schedule a broadcast without blocking.
 
     Convenience function that creates an asyncio task to broadcast
-    an event to all connected WebSocket clients.
+    an event to all connected WebSocket clients and forward to MQTT.
     """
     asyncio.create_task(ws_manager.broadcast(event_type, data))
 
+    from app.mqtt import mqtt_broadcast
+
+    mqtt_broadcast(event_type, data)
+
 
 def broadcast_error(message: str, details: str | None = None) -> None:
     """Broadcast an error notification to all connected clients.