Move to modular fanout bus

app/community_mqtt.py

@@ -555,50 +555,3 @@ class CommunityMqttPublisher(BaseMqttPublisher):
                 pass
             return False
         return True
-
-
-# Module-level singleton
-community_publisher = CommunityMqttPublisher()
-
-
-def community_mqtt_broadcast(event_type: str, data: dict[str, Any]) -> None:
-    """Fire-and-forget community MQTT publish for raw packets only."""
-    if event_type != "raw_packet":
-        return
-    if not community_publisher.connected or community_publisher._settings is None:
-        return
-    asyncio.create_task(_community_maybe_publish(data))
-
-
-async def _community_maybe_publish(data: dict[str, Any]) -> None:
-    """Format and publish a raw packet to the community broker."""
-    settings = community_publisher._settings
-    if settings is None or not settings.community_mqtt_enabled:
-        return
-
-    try:
-        from app.keystore import get_public_key
-        from app.radio import radio_manager
-
-        public_key = get_public_key()
-        if public_key is None:
-            return
-
-        pubkey_hex = public_key.hex().upper()
-
-        # Get device name from radio
-        device_name = ""
-        if radio_manager.meshcore and radio_manager.meshcore.self_info:
-            device_name = radio_manager.meshcore.self_info.get("name", "")
-
-        packet = _format_raw_packet(data, device_name, pubkey_hex)
-        iata = settings.community_mqtt_iata.upper().strip()
-        if not _IATA_RE.fullmatch(iata):
-            logger.debug("Community MQTT: skipping publish — no valid IATA code configured")
-            return
-        topic = f"meshcore/{iata}/{pubkey_hex}/packets"
-
-        await community_publisher.publish(topic, packet)
-
-    except Exception as e:
-        logger.warning("Community MQTT broadcast error: %s", e)

app/fanout/AGENTS_fanout.md

@@ -0,0 +1,92 @@
+# Fanout Bus Architecture
+
+The fanout bus is a unified system for dispatching mesh radio events (decoded messages and raw packets) to external integrations. It replaces the previous scattered singleton MQTT publishers with a modular, configurable framework.
+
+## Core Concepts
+
+### FanoutModule (base.py)
+Abstract base class that all integration modules implement:
+- `start()` / `stop()` — lifecycle management
+- `on_message(data)` — receive decoded messages
+- `on_raw(data)` — receive raw packets
+- `status` property — "connected" | "disconnected"
+
+### FanoutManager (manager.py)
+Singleton that owns all active modules and dispatches events:
+- `load_from_db()` — startup: load enabled configs, instantiate modules
+- `reload_config(id)` — CRUD: stop old, start new
+- `remove_config(id)` — delete: stop and remove
+- `broadcast_message(data)` — scope-check + dispatch `on_message`
+- `broadcast_raw(data)` — scope-check + dispatch `on_raw`
+- `stop_all()` — shutdown
+- `get_statuses()` — health endpoint data
+
+### Scope Matching
+Each config has a `scope` JSON blob controlling what events reach it:
+```json
+{"messages": "all", "raw_packets": "all"}
+{"messages": "none", "raw_packets": "all"}
+{"messages": {"channels": ["key1"], "contacts": "all"}, "raw_packets": "none"}
+```
+Community MQTT always enforces `{"messages": "none", "raw_packets": "all"}`.
+
+## Event Flow
+
+```
+Radio Event → packet_processor / event_handler
+  → broadcast_event("message"|"raw_packet", data, realtime=True)
+    → WebSocket broadcast (always)
+    → FanoutManager.broadcast_message/raw (only if realtime=True)
+      → scope check per module
+      → module.on_message / on_raw
+```
+
+Setting `realtime=False` (used during historical decryption) skips fanout dispatch entirely.
+
+## Current Module Types
+
+### mqtt_private (mqtt_private.py)
+Wraps `MqttPublisher` from `app/mqtt.py`. Config blob:
+- `broker_host`, `broker_port`, `username`, `password`
+- `use_tls`, `tls_insecure`, `topic_prefix`
+
+### mqtt_community (mqtt_community.py)
+Wraps `CommunityMqttPublisher` from `app/community_mqtt.py`. Config blob:
+- `broker_host`, `broker_port`, `iata`, `email`
+- Only publishes raw packets (on_message is a no-op)
+
+## Adding a New Integration Type
+
+1. Create `app/fanout/my_type.py` with a class extending `FanoutModule`
+2. Register it in `manager.py` → `_register_module_types()`
+3. Add validation in `app/routers/fanout.py` → `_VALID_TYPES` and validator function
+4. Add frontend editor component in `SettingsFanoutSection.tsx`
+
+## REST API
+
+| Method | Endpoint | Description |
+|--------|----------|-------------|
+| GET | `/api/fanout` | List all fanout configs |
+| POST | `/api/fanout` | Create new config |
+| PATCH | `/api/fanout/{id}` | Update config (triggers module reload) |
+| DELETE | `/api/fanout/{id}` | Delete config (stops module) |
+
+## Database
+
+`fanout_configs` table (created in migration 36):
+- `id` TEXT PRIMARY KEY
+- `type`, `name`, `enabled`, `config` (JSON), `scope` (JSON)
+- `sort_order`, `created_at`
+
+Migration 36 also migrates existing `app_settings` MQTT columns into fanout rows.
+
+## Key Files
+
+- `app/fanout/base.py` — FanoutModule ABC
+- `app/fanout/manager.py` — FanoutManager singleton
+- `app/fanout/mqtt_private.py` — Private MQTT module
+- `app/fanout/mqtt_community.py` — Community MQTT module
+- `app/repository/fanout.py` — Database CRUD
+- `app/routers/fanout.py` — REST API
+- `app/websocket.py` — `broadcast_event()` dispatches to fanout
+- `frontend/src/components/settings/SettingsFanoutSection.tsx` — UI

app/fanout/__init__.py

@@ -0,0 +1,8 @@
+from app.fanout.base import FanoutModule
+from app.fanout.manager import FanoutManager, fanout_manager
+
+__all__ = [
+    "FanoutManager",
+    "FanoutModule",
+    "fanout_manager",
+]

app/fanout/base.py

@@ -0,0 +1,34 @@
+"""Base class for fanout integration modules."""
+
+from __future__ import annotations
+
+
+class FanoutModule:
+    """Base class for all fanout integrations.
+
+    Each module wraps a specific integration (MQTT, webhook, etc.) and
+    receives dispatched messages/packets from the FanoutManager.
+
+    Subclasses must override the ``status`` property.
+    """
+
+    def __init__(self, config_id: str, config: dict) -> None:
+        self.config_id = config_id
+        self.config = config
+
+    async def start(self) -> None:
+        """Start the module (e.g. connect to broker). Override for persistent connections."""
+
+    async def stop(self) -> None:
+        """Stop the module (e.g. disconnect from broker)."""
+
+    async def on_message(self, data: dict) -> None:
+        """Called for decoded messages (DM/channel). Override if needed."""
+
+    async def on_raw(self, data: dict) -> None:
+        """Called for raw RF packets. Override if needed."""
+
+    @property
+    def status(self) -> str:
+        """Return 'connected', 'disconnected', or 'error'."""
+        raise NotImplementedError

app/fanout/manager.py

@@ -0,0 +1,162 @@
+"""FanoutManager: owns all active fanout modules and dispatches events."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.fanout.base import FanoutModule
+
+logger = logging.getLogger(__name__)
+
+# Type string -> module class mapping (extended in Phase 2/3)
+_MODULE_TYPES: dict[str, type] = {}
+
+
+def _register_module_types() -> None:
+    """Lazily populate the type registry to avoid circular imports."""
+    if _MODULE_TYPES:
+        return
+    from app.fanout.mqtt_community import MqttCommunityModule
+    from app.fanout.mqtt_private import MqttPrivateModule
+
+    _MODULE_TYPES["mqtt_private"] = MqttPrivateModule
+    _MODULE_TYPES["mqtt_community"] = MqttCommunityModule
+
+
+def _scope_matches_message(scope: dict, data: dict) -> bool:
+    """Check whether a message event matches the given scope."""
+    messages = scope.get("messages", "none")
+    if messages == "all":
+        return True
+    if messages == "none":
+        return False
+    if isinstance(messages, dict):
+        msg_type = data.get("type", "")
+        conversation_key = data.get("conversation_key", "")
+        if msg_type == "CHAN":
+            channels = messages.get("channels", "none")
+            if channels == "all":
+                return True
+            if channels == "none":
+                return False
+            if isinstance(channels, list):
+                return conversation_key in channels
+        elif msg_type == "PRIV":
+            contacts = messages.get("contacts", "none")
+            if contacts == "all":
+                return True
+            if contacts == "none":
+                return False
+            if isinstance(contacts, list):
+                return conversation_key in contacts
+    return False
+
+
+def _scope_matches_raw(scope: dict, _data: dict) -> bool:
+    """Check whether a raw packet event matches the given scope."""
+    return scope.get("raw_packets", "none") == "all"
+
+
+class FanoutManager:
+    """Owns all active fanout modules and dispatches events."""
+
+    def __init__(self) -> None:
+        self._modules: dict[str, tuple[FanoutModule, dict]] = {}  # id -> (module, scope)
+
+    async def load_from_db(self) -> None:
+        """Read enabled fanout_configs and instantiate modules."""
+        _register_module_types()
+        from app.repository.fanout import FanoutConfigRepository
+
+        configs = await FanoutConfigRepository.get_enabled()
+        for cfg in configs:
+            await self._start_module(cfg)
+
+    async def _start_module(self, cfg: dict[str, Any]) -> None:
+        """Instantiate and start a single module from a config dict."""
+        config_id = cfg["id"]
+        config_type = cfg["type"]
+        config_blob = cfg["config"]
+        scope = cfg["scope"]
+
+        cls = _MODULE_TYPES.get(config_type)
+        if cls is None:
+            logger.warning("Unknown fanout type %r for config %s, skipping", config_type, config_id)
+            return
+
+        try:
+            module = cls(config_id, config_blob)
+            await module.start()
+            self._modules[config_id] = (module, scope)
+            logger.info(
+                "Started fanout module %s (type=%s)", cfg.get("name", config_id), config_type
+            )
+        except Exception:
+            logger.exception("Failed to start fanout module %s", config_id)
+
+    async def reload_config(self, config_id: str) -> None:
+        """Stop old module (if any) and start updated config."""
+        await self.remove_config(config_id)
+
+        from app.repository.fanout import FanoutConfigRepository
+
+        cfg = await FanoutConfigRepository.get(config_id)
+        if cfg is None or not cfg["enabled"]:
+            return
+        await self._start_module(cfg)
+
+    async def remove_config(self, config_id: str) -> None:
+        """Stop and remove a module."""
+        entry = self._modules.pop(config_id, None)
+        if entry is not None:
+            module, _ = entry
+            try:
+                await module.stop()
+            except Exception:
+                logger.exception("Error stopping fanout module %s", config_id)
+
+    async def broadcast_message(self, data: dict) -> None:
+        """Dispatch a decoded message to modules whose scope matches."""
+        for config_id, (module, scope) in list(self._modules.items()):
+            if _scope_matches_message(scope, data):
+                try:
+                    await module.on_message(data)
+                except Exception:
+                    logger.exception("Fanout %s on_message error", config_id)
+
+    async def broadcast_raw(self, data: dict) -> None:
+        """Dispatch a raw packet to modules whose scope matches."""
+        for config_id, (module, scope) in list(self._modules.items()):
+            if _scope_matches_raw(scope, data):
+                try:
+                    await module.on_raw(data)
+                except Exception:
+                    logger.exception("Fanout %s on_raw error", config_id)
+
+    async def stop_all(self) -> None:
+        """Shutdown all modules."""
+        for config_id, (module, _) in list(self._modules.items()):
+            try:
+                await module.stop()
+            except Exception:
+                logger.exception("Error stopping fanout module %s", config_id)
+        self._modules.clear()
+
+    def get_statuses(self) -> dict[str, dict[str, str]]:
+        """Return status info for each active module."""
+        from app.repository.fanout import _configs_cache
+
+        result: dict[str, dict[str, str]] = {}
+        for config_id, (module, _) in self._modules.items():
+            info = _configs_cache.get(config_id, {})
+            result[config_id] = {
+                "name": info.get("name", config_id),
+                "type": info.get("type", "unknown"),
+                "status": module.status,
+            }
+        return result
+
+
+# Module-level singleton
+fanout_manager = FanoutManager()

app/fanout/mqtt_community.py

@@ -0,0 +1,89 @@
+"""Fanout module wrapping the community MQTT publisher."""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+
+from app.community_mqtt import CommunityMqttPublisher, _format_raw_packet
+from app.fanout.base import FanoutModule
+from app.models import AppSettings
+
+logger = logging.getLogger(__name__)
+
+_IATA_RE = re.compile(r"^[A-Z]{3}$")
+
+
+def _config_to_settings(config: dict) -> AppSettings:
+    """Map a fanout config blob to AppSettings for the CommunityMqttPublisher."""
+    return AppSettings(
+        community_mqtt_enabled=True,
+        community_mqtt_broker_host=config.get("broker_host", "mqtt-us-v1.letsmesh.net"),
+        community_mqtt_broker_port=config.get("broker_port", 443),
+        community_mqtt_iata=config.get("iata", ""),
+        community_mqtt_email=config.get("email", ""),
+    )
+
+
+class MqttCommunityModule(FanoutModule):
+    """Wraps a CommunityMqttPublisher for community packet sharing."""
+
+    def __init__(self, config_id: str, config: dict) -> None:
+        super().__init__(config_id, config)
+        self._publisher = CommunityMqttPublisher()
+
+    async def start(self) -> None:
+        settings = _config_to_settings(self.config)
+        await self._publisher.start(settings)
+
+    async def stop(self) -> None:
+        await self._publisher.stop()
+
+    async def on_message(self, data: dict) -> None:
+        # Community MQTT only publishes raw packets, not decoded messages.
+        pass
+
+    async def on_raw(self, data: dict) -> None:
+        if not self._publisher.connected or self._publisher._settings is None:
+            return
+        await _publish_community_packet(self._publisher, self.config, data)
+
+    @property
+    def status(self) -> str:
+        if self._publisher._is_configured():
+            return "connected" if self._publisher.connected else "disconnected"
+        return "disconnected"
+
+
+async def _publish_community_packet(
+    publisher: CommunityMqttPublisher,
+    config: dict,
+    data: dict[str, Any],
+) -> None:
+    """Format and publish a raw packet to the community broker."""
+    try:
+        from app.keystore import get_public_key
+        from app.radio import radio_manager
+
+        public_key = get_public_key()
+        if public_key is None:
+            return
+
+        pubkey_hex = public_key.hex().upper()
+
+        device_name = ""
+        if radio_manager.meshcore and radio_manager.meshcore.self_info:
+            device_name = radio_manager.meshcore.self_info.get("name", "")
+
+        packet = _format_raw_packet(data, device_name, pubkey_hex)
+        iata = config.get("iata", "").upper().strip()
+        if not _IATA_RE.fullmatch(iata):
+            logger.debug("Community MQTT: skipping publish — no valid IATA code configured")
+            return
+        topic = f"meshcore/{iata}/{pubkey_hex}/packets"
+
+        await publisher.publish(topic, packet)
+
+    except Exception as e:
+        logger.warning("Community MQTT broadcast error: %s", e)

app/fanout/mqtt_private.py

@@ -0,0 +1,62 @@
+"""Fanout module wrapping the private MQTT publisher."""
+
+from __future__ import annotations
+
+import logging
+
+from app.fanout.base import FanoutModule
+from app.models import AppSettings
+from app.mqtt import MqttPublisher, _build_message_topic, _build_raw_packet_topic
+
+logger = logging.getLogger(__name__)
+
+
+def _config_to_settings(config: dict) -> AppSettings:
+    """Map a fanout config blob to AppSettings for the MqttPublisher."""
+    return AppSettings(
+        mqtt_broker_host=config.get("broker_host", ""),
+        mqtt_broker_port=config.get("broker_port", 1883),
+        mqtt_username=config.get("username", ""),
+        mqtt_password=config.get("password", ""),
+        mqtt_use_tls=config.get("use_tls", False),
+        mqtt_tls_insecure=config.get("tls_insecure", False),
+        mqtt_topic_prefix=config.get("topic_prefix", "meshcore"),
+        # Always enable both publish flags; the fanout scope controls delivery.
+        mqtt_publish_messages=True,
+        mqtt_publish_raw_packets=True,
+    )
+
+
+class MqttPrivateModule(FanoutModule):
+    """Wraps an MqttPublisher instance for private MQTT forwarding."""
+
+    def __init__(self, config_id: str, config: dict) -> None:
+        super().__init__(config_id, config)
+        self._publisher = MqttPublisher()
+
+    async def start(self) -> None:
+        settings = _config_to_settings(self.config)
+        await self._publisher.start(settings)
+
+    async def stop(self) -> None:
+        await self._publisher.stop()
+
+    async def on_message(self, data: dict) -> None:
+        if not self._publisher.connected or self._publisher._settings is None:
+            return
+        prefix = self.config.get("topic_prefix", "meshcore")
+        topic = _build_message_topic(prefix, data)
+        await self._publisher.publish(topic, data)
+
+    async def on_raw(self, data: dict) -> None:
+        if not self._publisher.connected or self._publisher._settings is None:
+            return
+        prefix = self.config.get("topic_prefix", "meshcore")
+        topic = _build_raw_packet_topic(prefix, data)
+        await self._publisher.publish(topic, data)
+
+    @property
+    def status(self) -> str:
+        if not self.config.get("broker_host"):
+            return "disconnected"
+        return "connected" if self._publisher.connected else "disconnected"

app/main.py

@@ -18,6 +18,7 @@ from app.radio_sync import (
 from app.routers import (
     channels,
     contacts,
+    fanout,
     health,
     messages,
     packets,
@@ -56,23 +57,18 @@ async def lifespan(app: FastAPI):
     # Always start connection monitor (even if initial connection failed)
     await radio_manager.start_connection_monitor()
 
-    # Start MQTT publishers if configured
-    from app.community_mqtt import community_publisher
-    from app.mqtt import mqtt_publisher
-    from app.repository import AppSettingsRepository
+    # Start fanout modules (MQTT, etc.) from database configs
+    from app.fanout.manager import fanout_manager
 
     try:
-        mqtt_settings = await AppSettingsRepository.get()
-        await mqtt_publisher.start(mqtt_settings)
-        await community_publisher.start(mqtt_settings)
+        await fanout_manager.load_from_db()
     except Exception as e:
-        logger.warning("Failed to start MQTT publisher(s): %s", e)
+        logger.warning("Failed to start fanout modules: %s", e)
 
     yield
 
     logger.info("Shutting down")
-    await community_publisher.stop()
-    await mqtt_publisher.stop()
+    await fanout_manager.stop_all()
     await radio_manager.stop_connection_monitor()
     await stop_message_polling()
     await stop_periodic_advert()
@@ -119,6 +115,7 @@ async def radio_disconnected_handler(request: Request, exc: RadioDisconnectedErr
 
 # API routes - all prefixed with /api for production compatibility
 app.include_router(health.router, prefix="/api")
+app.include_router(fanout.router, prefix="/api")
 app.include_router(radio.router, prefix="/api")
 app.include_router(contacts.router, prefix="/api")
 app.include_router(repeaters.router, prefix="/api")

app/migrations.py

@@ -282,6 +282,13 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
         await set_version(conn, 35)
         applied += 1
 
+    # Migration 36: Create fanout_configs table and migrate existing MQTT settings
+    if version < 36:
+        logger.info("Applying migration 36: create fanout_configs and migrate MQTT settings")
+        await _migrate_036_create_fanout_configs(conn)
+        await set_version(conn, 36)
+        applied += 1
+
     if applied > 0:
         logger.info(
             "Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
@@ -2014,3 +2021,131 @@ async def _migrate_035_add_block_lists(conn: aiosqlite.Connection) -> None:
             raise
 
     await conn.commit()
+
+
+async def _migrate_036_create_fanout_configs(conn: aiosqlite.Connection) -> None:
+    """Create fanout_configs table and migrate existing MQTT settings.
+
+    Reads existing MQTT settings from app_settings and creates corresponding
+    fanout_configs rows. Old columns are NOT dropped (rollback safety).
+    """
+    import json
+    import uuid
+
+    # 1. Create fanout_configs table
+    await conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS fanout_configs (
+            id TEXT PRIMARY KEY,
+            type TEXT NOT NULL,
+            name TEXT NOT NULL,
+            enabled INTEGER DEFAULT 0,
+            config TEXT NOT NULL DEFAULT '{}',
+            scope TEXT NOT NULL DEFAULT '{}',
+            sort_order INTEGER DEFAULT 0,
+            created_at INTEGER NOT NULL
+        )
+        """
+    )
+
+    # 2. Read existing MQTT settings
+    try:
+        cursor = await conn.execute(
+            """
+            SELECT 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,
+                   community_mqtt_enabled, community_mqtt_iata,
+                   community_mqtt_broker_host, community_mqtt_broker_port,
+                   community_mqtt_email
+            FROM app_settings WHERE id = 1
+            """
+        )
+        row = await cursor.fetchone()
+    except Exception:
+        row = None
+
+    if row is None:
+        await conn.commit()
+        return
+
+    import time
+
+    now = int(time.time())
+    sort_order = 0
+
+    # 3. Migrate private MQTT if configured
+    broker_host = row["mqtt_broker_host"] or ""
+    if broker_host:
+        publish_messages = bool(row["mqtt_publish_messages"])
+        publish_raw = bool(row["mqtt_publish_raw_packets"])
+        enabled = publish_messages or publish_raw
+
+        config = {
+            "broker_host": broker_host,
+            "broker_port": row["mqtt_broker_port"] or 1883,
+            "username": row["mqtt_username"] or "",
+            "password": row["mqtt_password"] or "",
+            "use_tls": bool(row["mqtt_use_tls"]),
+            "tls_insecure": bool(row["mqtt_tls_insecure"]),
+            "topic_prefix": row["mqtt_topic_prefix"] or "meshcore",
+        }
+
+        scope = {
+            "messages": "all" if publish_messages else "none",
+            "raw_packets": "all" if publish_raw else "none",
+        }
+
+        await conn.execute(
+            """
+            INSERT INTO fanout_configs (id, type, name, enabled, config, scope, sort_order, created_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                str(uuid.uuid4()),
+                "mqtt_private",
+                "Private MQTT",
+                1 if enabled else 0,
+                json.dumps(config),
+                json.dumps(scope),
+                sort_order,
+                now,
+            ),
+        )
+        sort_order += 1
+        logger.info("Migrated private MQTT settings to fanout_configs (enabled=%s)", enabled)
+
+    # 4. Migrate community MQTT if enabled
+    community_enabled = bool(row["community_mqtt_enabled"])
+    if community_enabled:
+        config = {
+            "broker_host": row["community_mqtt_broker_host"] or "mqtt-us-v1.letsmesh.net",
+            "broker_port": row["community_mqtt_broker_port"] or 443,
+            "iata": row["community_mqtt_iata"] or "",
+            "email": row["community_mqtt_email"] or "",
+        }
+
+        scope = {
+            "messages": "none",
+            "raw_packets": "all",
+        }
+
+        await conn.execute(
+            """
+            INSERT INTO fanout_configs (id, type, name, enabled, config, scope, sort_order, created_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                str(uuid.uuid4()),
+                "mqtt_community",
+                "Community MQTT",
+                1,
+                json.dumps(config),
+                json.dumps(scope),
+                sort_order,
+                now,
+            ),
+        )
+        logger.info("Migrated community MQTT settings to fanout_configs")
+
+    await conn.commit()

app/models.py

@@ -533,6 +533,19 @@ class AppSettings(BaseModel):
     )
 
 
+class FanoutConfig(BaseModel):
+    """Configuration for a single fanout integration."""
+
+    id: str
+    type: str  # 'mqtt_private' | 'mqtt_community'
+    name: str
+    enabled: bool
+    config: dict
+    scope: dict
+    sort_order: int = 0
+    created_at: int = 0
+
+
 class BusyChannel(BaseModel):
     channel_key: str
     channel_name: str

app/mqtt.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import logging
 import ssl
 from typing import Any
@@ -54,38 +53,6 @@ class MqttPublisher(BaseMqttPublisher):
         return ctx
 
 
-# Module-level singleton
-mqtt_publisher = MqttPublisher()
-
-
-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", "")

app/packet_processor.py

@@ -209,6 +209,7 @@ async def create_message_from_decrypted(
             sender_name=sender,
             sender_key=resolved_sender_key,
         ).model_dump(),
+        realtime=trigger_bot,
     )
 
     # Run bot if enabled (for incoming channel messages, not historical decryption)
@@ -332,6 +333,7 @@ async def create_dm_message_from_decrypted(
             sender_name=sender_name,
             sender_key=conversation_key if not outgoing else None,
         ).model_dump(),
+        realtime=trigger_bot,
     )
 
     # Update contact's last_contacted timestamp (for sorting)

app/repository/__init__.py

@@ -5,6 +5,7 @@ from app.repository.contacts import (
     ContactNameHistoryRepository,
     ContactRepository,
 )
+from app.repository.fanout import FanoutConfigRepository
 from app.repository.messages import MessageRepository
 from app.repository.raw_packets import RawPacketRepository
 from app.repository.settings import AppSettingsRepository, StatisticsRepository
@@ -16,6 +17,7 @@ __all__ = [
     "ContactAdvertPathRepository",
     "ContactNameHistoryRepository",
     "ContactRepository",
+    "FanoutConfigRepository",
     "MessageRepository",
     "RawPacketRepository",
     "StatisticsRepository",

app/repository/fanout.py

@@ -0,0 +1,137 @@
+"""Repository for fanout_configs table."""
+
+import json
+import logging
+import time
+import uuid
+from typing import Any
+
+from app.database import db
+
+logger = logging.getLogger(__name__)
+
+# In-memory cache of config metadata (name, type) for status reporting.
+# Populated by get_all/get/create/update and read by FanoutManager.get_statuses().
+_configs_cache: dict[str, dict[str, Any]] = {}
+
+
+def _row_to_dict(row: Any) -> dict[str, Any]:
+    """Convert a database row to a config dict."""
+    result = {
+        "id": row["id"],
+        "type": row["type"],
+        "name": row["name"],
+        "enabled": bool(row["enabled"]),
+        "config": json.loads(row["config"]) if row["config"] else {},
+        "scope": json.loads(row["scope"]) if row["scope"] else {},
+        "sort_order": row["sort_order"] or 0,
+        "created_at": row["created_at"] or 0,
+    }
+    _configs_cache[result["id"]] = result
+    return result
+
+
+class FanoutConfigRepository:
+    """CRUD operations for fanout_configs table."""
+
+    @staticmethod
+    async def get_all() -> list[dict[str, Any]]:
+        """Get all fanout configs ordered by sort_order."""
+        cursor = await db.conn.execute(
+            "SELECT * FROM fanout_configs ORDER BY sort_order, created_at"
+        )
+        rows = await cursor.fetchall()
+        return [_row_to_dict(row) for row in rows]
+
+    @staticmethod
+    async def get(config_id: str) -> dict[str, Any] | None:
+        """Get a single fanout config by ID."""
+        cursor = await db.conn.execute("SELECT * FROM fanout_configs WHERE id = ?", (config_id,))
+        row = await cursor.fetchone()
+        if row is None:
+            return None
+        return _row_to_dict(row)
+
+    @staticmethod
+    async def create(
+        config_type: str,
+        name: str,
+        config: dict,
+        scope: dict,
+        enabled: bool = True,
+        config_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Create a new fanout config."""
+        new_id = config_id or str(uuid.uuid4())
+        now = int(time.time())
+
+        # Get next sort_order
+        cursor = await db.conn.execute(
+            "SELECT COALESCE(MAX(sort_order), -1) + 1 FROM fanout_configs"
+        )
+        row = await cursor.fetchone()
+        sort_order = row[0] if row else 0
+
+        await db.conn.execute(
+            """
+            INSERT INTO fanout_configs (id, type, name, enabled, config, scope, sort_order, created_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                new_id,
+                config_type,
+                name,
+                1 if enabled else 0,
+                json.dumps(config),
+                json.dumps(scope),
+                sort_order,
+                now,
+            ),
+        )
+        await db.conn.commit()
+
+        result = await FanoutConfigRepository.get(new_id)
+        assert result is not None
+        return result
+
+    @staticmethod
+    async def update(config_id: str, **fields: Any) -> dict[str, Any] | None:
+        """Update a fanout config. Only provided fields are updated."""
+        updates = []
+        params: list[Any] = []
+
+        for field in ("name", "enabled", "config", "scope", "sort_order"):
+            if field in fields:
+                value = fields[field]
+                if field == "enabled":
+                    value = 1 if value else 0
+                elif field in ("config", "scope"):
+                    value = json.dumps(value)
+                updates.append(f"{field} = ?")
+                params.append(value)
+
+        if not updates:
+            return await FanoutConfigRepository.get(config_id)
+
+        params.append(config_id)
+        query = f"UPDATE fanout_configs SET {', '.join(updates)} WHERE id = ?"
+        await db.conn.execute(query, params)
+        await db.conn.commit()
+
+        return await FanoutConfigRepository.get(config_id)
+
+    @staticmethod
+    async def delete(config_id: str) -> None:
+        """Delete a fanout config."""
+        await db.conn.execute("DELETE FROM fanout_configs WHERE id = ?", (config_id,))
+        await db.conn.commit()
+        _configs_cache.pop(config_id, None)
+
+    @staticmethod
+    async def get_enabled() -> list[dict[str, Any]]:
+        """Get all enabled fanout configs."""
+        cursor = await db.conn.execute(
+            "SELECT * FROM fanout_configs WHERE enabled = 1 ORDER BY sort_order, created_at"
+        )
+        rows = await cursor.fetchall()
+        return [_row_to_dict(row) for row in rows]

app/routers/fanout.py

@@ -0,0 +1,165 @@
+"""REST API for fanout config CRUD."""
+
+import logging
+import re
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from app.repository.fanout import FanoutConfigRepository
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/fanout", tags=["fanout"])
+
+# Valid types in Phase 1
+_VALID_TYPES = {"mqtt_private", "mqtt_community"}
+
+_IATA_RE = re.compile(r"^[A-Z]{3}$")
+
+
+class FanoutConfigCreate(BaseModel):
+    type: str = Field(description="Integration type: 'mqtt_private' or 'mqtt_community'")
+    name: str = Field(min_length=1, description="User-assigned label")
+    config: dict = Field(default_factory=dict, description="Type-specific config blob")
+    scope: dict = Field(default_factory=dict, description="Scope controls")
+    enabled: bool = Field(default=True, description="Whether enabled on creation")
+
+
+class FanoutConfigUpdate(BaseModel):
+    name: str | None = Field(default=None, description="Updated label")
+    config: dict | None = Field(default=None, description="Updated config blob")
+    scope: dict | None = Field(default=None, description="Updated scope controls")
+    enabled: bool | None = Field(default=None, description="Enable/disable toggle")
+
+
+def _validate_mqtt_private_config(config: dict) -> None:
+    """Validate mqtt_private config blob."""
+    if not config.get("broker_host"):
+        raise HTTPException(status_code=400, detail="broker_host is required for mqtt_private")
+    port = config.get("broker_port", 1883)
+    if not isinstance(port, int) or port < 1 or port > 65535:
+        raise HTTPException(status_code=400, detail="broker_port must be between 1 and 65535")
+
+
+def _validate_mqtt_community_config(config: dict) -> None:
+    """Validate mqtt_community config blob."""
+    iata = config.get("iata", "")
+    if iata and not _IATA_RE.fullmatch(iata.upper().strip()):
+        raise HTTPException(
+            status_code=400,
+            detail="IATA code must be exactly 3 uppercase alphabetic characters",
+        )
+
+
+def _enforce_scope(config_type: str, scope: dict) -> dict:
+    """Enforce type-specific scope constraints. Returns normalized scope."""
+    if config_type == "mqtt_community":
+        # Community MQTT always: no messages, all raw packets
+        return {"messages": "none", "raw_packets": "all"}
+    # For mqtt_private, validate scope values
+    messages = scope.get("messages", "all")
+    if messages not in ("all", "none") and not isinstance(messages, dict):
+        messages = "all"
+    raw_packets = scope.get("raw_packets", "all")
+    if raw_packets not in ("all", "none"):
+        raw_packets = "all"
+    return {"messages": messages, "raw_packets": raw_packets}
+
+
+@router.get("")
+async def list_fanout_configs() -> list[dict]:
+    """List all fanout configs."""
+    return await FanoutConfigRepository.get_all()
+
+
+@router.post("")
+async def create_fanout_config(body: FanoutConfigCreate) -> dict:
+    """Create a new fanout config."""
+    if body.type not in _VALID_TYPES:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Invalid type '{body.type}'. Must be one of: {', '.join(sorted(_VALID_TYPES))}",
+        )
+
+    # Only validate config when creating as enabled — disabled configs
+    # are drafts the user hasn't finished configuring yet.
+    if body.enabled:
+        if body.type == "mqtt_private":
+            _validate_mqtt_private_config(body.config)
+        elif body.type == "mqtt_community":
+            _validate_mqtt_community_config(body.config)
+
+    scope = _enforce_scope(body.type, body.scope)
+
+    cfg = await FanoutConfigRepository.create(
+        config_type=body.type,
+        name=body.name,
+        config=body.config,
+        scope=scope,
+        enabled=body.enabled,
+    )
+
+    # Start the module if enabled
+    if cfg["enabled"]:
+        from app.fanout.manager import fanout_manager
+
+        await fanout_manager.reload_config(cfg["id"])
+
+    logger.info("Created fanout config %s (type=%s, name=%s)", cfg["id"], body.type, body.name)
+    return cfg
+
+
+@router.patch("/{config_id}")
+async def update_fanout_config(config_id: str, body: FanoutConfigUpdate) -> dict:
+    """Update a fanout config. Triggers module reload."""
+    existing = await FanoutConfigRepository.get(config_id)
+    if existing is None:
+        raise HTTPException(status_code=404, detail="Fanout config not found")
+
+    kwargs = {}
+    if body.name is not None:
+        kwargs["name"] = body.name
+    if body.enabled is not None:
+        kwargs["enabled"] = body.enabled
+    if body.config is not None:
+        kwargs["config"] = body.config
+    if body.scope is not None:
+        kwargs["scope"] = _enforce_scope(existing["type"], body.scope)
+
+    # Validate config when the result will be enabled
+    will_be_enabled = body.enabled if body.enabled is not None else existing["enabled"]
+    if will_be_enabled:
+        config_to_validate = body.config if body.config is not None else existing["config"]
+        if existing["type"] == "mqtt_private":
+            _validate_mqtt_private_config(config_to_validate)
+        elif existing["type"] == "mqtt_community":
+            _validate_mqtt_community_config(config_to_validate)
+
+    updated = await FanoutConfigRepository.update(config_id, **kwargs)
+    if updated is None:
+        raise HTTPException(status_code=404, detail="Fanout config not found")
+
+    # Reload the module to pick up changes
+    from app.fanout.manager import fanout_manager
+
+    await fanout_manager.reload_config(config_id)
+
+    logger.info("Updated fanout config %s", config_id)
+    return updated
+
+
+@router.delete("/{config_id}")
+async def delete_fanout_config(config_id: str) -> dict:
+    """Delete a fanout config."""
+    existing = await FanoutConfigRepository.get(config_id)
+    if existing is None:
+        raise HTTPException(status_code=404, detail="Fanout config not found")
+
+    # Stop the module first
+    from app.fanout.manager import fanout_manager
+
+    await fanout_manager.remove_config(config_id)
+    await FanoutConfigRepository.delete(config_id)
+
+    logger.info("Deleted fanout config %s", config_id)
+    return {"deleted": True}

app/routers/health.py

@@ -1,4 +1,5 @@
 import os
+from typing import Any
 
 from fastapi import APIRouter
 from pydantic import BaseModel
@@ -16,8 +17,7 @@ class HealthResponse(BaseModel):
     connection_info: str | None
     database_size_mb: float
     oldest_undecrypted_timestamp: int | None
-    mqtt_status: str | None = None
-    community_mqtt_status: str | None = None
+    fanout_statuses: dict[str, dict[str, str]] = {}
     bots_disabled: bool = False
 
 
@@ -36,27 +36,12 @@ 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
+    # Fanout module statuses
+    fanout_statuses: dict[str, Any] = {}
     try:
-        from app.mqtt import mqtt_publisher
+        from app.fanout.manager import fanout_manager
 
-        if mqtt_publisher._is_configured():
-            mqtt_status = "connected" if mqtt_publisher.connected else "disconnected"
-        else:
-            mqtt_status = "disabled"
-    except Exception:
-        pass
-
-    # Community MQTT status
-    community_mqtt_status: str | None = None
-    try:
-        from app.community_mqtt import community_publisher
-
-        if community_publisher._is_configured():
-            community_mqtt_status = "connected" if community_publisher.connected else "disconnected"
-        else:
-            community_mqtt_status = "disabled"
+        fanout_statuses = fanout_manager.get_statuses()
     except Exception:
         pass
 
@@ -66,8 +51,7 @@ async def build_health_data(radio_connected: bool, connection_info: str | None)
         "connection_info": connection_info,
         "database_size_mb": db_size_mb,
         "oldest_undecrypted_timestamp": oldest_ts,
-        "mqtt_status": mqtt_status,
-        "community_mqtt_status": community_mqtt_status,
+        "fanout_statuses": fanout_statuses,
         "bots_disabled": settings.disable_bots,
     }
 

app/routers/settings.py

@@ -1,6 +1,5 @@
 import asyncio
 import logging
-import re
 from typing import Literal
 
 from fastapi import APIRouter, HTTPException
@@ -61,66 +60,6 @@ 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",
-    )
-    community_mqtt_enabled: bool | None = Field(
-        default=None,
-        description="Whether to publish raw packets to the community MQTT broker",
-    )
-    community_mqtt_iata: str | None = Field(
-        default=None,
-        description="IATA region code for community MQTT topic routing (3 alpha chars)",
-    )
-    community_mqtt_broker_host: str | None = Field(
-        default=None,
-        description="Community MQTT broker hostname",
-    )
-    community_mqtt_broker_port: int | None = Field(
-        default=None,
-        ge=1,
-        le=65535,
-        description="Community MQTT broker port",
-    )
-    community_mqtt_email: str | None = Field(
-        default=None,
-        description="Email address for node claiming on the community aggregator",
-    )
     flood_scope: str | None = Field(
         default=None,
         description="Outbound flood scope / region name (empty = disabled)",
@@ -210,53 +149,6 @@ 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
-
-    # Community MQTT fields
-    community_mqtt_changed = False
-    if update.community_mqtt_enabled is not None:
-        kwargs["community_mqtt_enabled"] = update.community_mqtt_enabled
-        community_mqtt_changed = True
-
-    if update.community_mqtt_iata is not None:
-        iata = update.community_mqtt_iata.upper().strip()
-        if iata and not re.fullmatch(r"[A-Z]{3}", iata):
-            raise HTTPException(
-                status_code=400,
-                detail="IATA code must be exactly 3 uppercase alphabetic characters",
-            )
-        kwargs["community_mqtt_iata"] = iata
-        community_mqtt_changed = True
-
-    if update.community_mqtt_broker_host is not None:
-        kwargs["community_mqtt_broker_host"] = update.community_mqtt_broker_host
-        community_mqtt_changed = True
-
-    if update.community_mqtt_broker_port is not None:
-        kwargs["community_mqtt_broker_port"] = update.community_mqtt_broker_port
-        community_mqtt_changed = True
-
-    if update.community_mqtt_email is not None:
-        kwargs["community_mqtt_email"] = update.community_mqtt_email
-        community_mqtt_changed = True
-
     # Block lists
     if update.blocked_keys is not None:
         kwargs["blocked_keys"] = [k.lower() for k in update.blocked_keys]
@@ -270,34 +162,9 @@ async def update_settings(update: AppSettingsUpdate) -> AppSettings:
         kwargs["flood_scope"] = stripped
         flood_scope_changed = True
 
-    # Require IATA when enabling community MQTT
-    if kwargs.get("community_mqtt_enabled", False):
-        # Check the IATA value being set, or fall back to current settings
-        iata_value = kwargs.get("community_mqtt_iata")
-        if iata_value is None:
-            current = await AppSettingsRepository.get()
-            iata_value = current.community_mqtt_iata
-        if not iata_value or not re.fullmatch(r"[A-Z]{3}", iata_value):
-            raise HTTPException(
-                status_code=400,
-                detail="A valid IATA region code is required to enable community sharing",
-            )
-
     if 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)
-
-        # Restart community MQTT publisher if any community settings changed
-        if community_mqtt_changed:
-            from app.community_mqtt import community_publisher
-
-            await community_publisher.restart(result)
-
         # Apply flood scope to radio immediately if changed
         if flood_scope_changed:
             from app.radio import radio_manager

app/websocket.py

@@ -92,21 +92,26 @@ class WebSocketManager:
 ws_manager = WebSocketManager()
 
 
-def broadcast_event(event_type: str, data: dict) -> None:
+def broadcast_event(event_type: str, data: dict, *, realtime: bool = True) -> None:
     """Schedule a broadcast without blocking.
 
     Convenience function that creates an asyncio task to broadcast
-    an event to all connected WebSocket clients and forward to MQTT.
+    an event to all connected WebSocket clients and forward to fanout modules.
+
+    Args:
+        event_type: Event type string (e.g. "message", "raw_packet")
+        data: Event payload dict
+        realtime: If False, skip fanout dispatch (used for historical decryption)
     """
     asyncio.create_task(ws_manager.broadcast(event_type, data))
 
-    from app.mqtt import mqtt_broadcast
+    if realtime:
+        from app.fanout.manager import fanout_manager
 
-    mqtt_broadcast(event_type, data)
-
-    from app.community_mqtt import community_mqtt_broadcast
-
-    community_mqtt_broadcast(event_type, data)
+        if event_type == "message":
+            asyncio.create_task(fanout_manager.broadcast_message(data))
+        elif event_type == "raw_packet":
+            asyncio.create_task(fanout_manager.broadcast_raw(data))
 
 
 def broadcast_error(message: str, details: str | None = None) -> None: