Add SQS fanout

app/AGENTS.md

@@ -45,7 +45,7 @@ app/
 ├── events.py            # Typed WS event payload serialization
 ├── websocket.py         # WS manager + broadcast helpers
 ├── security.py          # Optional app-wide HTTP Basic auth middleware for HTTP + WS
-├── fanout/              # Fanout bus: MQTT, bots, webhooks, Apprise (see fanout/AGENTS_fanout.md)
+├── fanout/              # Fanout bus: MQTT, bots, webhooks, Apprise, SQS (see fanout/AGENTS_fanout.md)
 ├── dependencies.py      # Shared FastAPI dependency providers
 ├── path_utils.py        # Path hex rendering and hop-width helpers
 ├── keystore.py          # Ephemeral private/public key storage for DM decryption
@@ -132,7 +132,7 @@ app/
 
 ### Fanout bus
 
-- All external integrations (MQTT, bots, webhooks, Apprise) are managed through the fanout bus (`app/fanout/`).
+- All external integrations (MQTT, bots, webhooks, Apprise, SQS) are managed through the fanout bus (`app/fanout/`).
 - Configs stored in `fanout_configs` table, managed via `GET/POST/PATCH/DELETE /api/fanout`.
 - `broadcast_event()` in `websocket.py` dispatches to the fanout manager for `message` and `raw_packet` events.
 - Each integration is a `FanoutModule` with scope-based filtering.

app/fanout/AGENTS_fanout.md

@@ -79,6 +79,14 @@ Push notifications via Apprise library. Config blob:
 - `preserve_identity` — suppress Discord webhook name/avatar override
 - `include_path` — include routing path in notification body
 
+### sqs (sqs.py)
+Amazon SQS delivery. Config blob:
+- `queue_url` — target queue URL
+- `region_name` (optional), `endpoint_url` (optional)
+- `access_key_id`, `secret_access_key`, `session_token` (all optional; blank uses the normal AWS credential chain)
+- Publishes a JSON envelope of the form `{"event_type":"message"|"raw_packet","data":...}`
+- Supports both decoded messages and raw packets via normal scope selection
+
 ## Adding a New Integration Type
 
 ### Step-by-step checklist
@@ -132,7 +140,7 @@ Three changes needed:
 
 **a)** Add to `_VALID_TYPES` set:
 ```python
-_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise", "my_type"}
+_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise", "sqs", "my_type"}
 ```
 
 **b)** Add a validation function:
@@ -280,6 +288,7 @@ Migrations:
 - `app/fanout/bot_exec.py` — Bot code execution, response processing, rate limiting
 - `app/fanout/webhook.py` — Webhook fanout module
 - `app/fanout/apprise_mod.py` — Apprise fanout module
+- `app/fanout/sqs.py` — Amazon SQS fanout module
 - `app/repository/fanout.py` — Database CRUD
 - `app/routers/fanout.py` — REST API
 - `app/websocket.py` — `broadcast_event()` dispatches to fanout

app/fanout/manager.py

@@ -23,6 +23,7 @@ def _register_module_types() -> None:
     from app.fanout.bot import BotModule
     from app.fanout.mqtt_community import MqttCommunityModule
     from app.fanout.mqtt_private import MqttPrivateModule
+    from app.fanout.sqs import SqsModule
     from app.fanout.webhook import WebhookModule
 
     _MODULE_TYPES["mqtt_private"] = MqttPrivateModule
@@ -30,6 +31,7 @@ def _register_module_types() -> None:
     _MODULE_TYPES["bot"] = BotModule
     _MODULE_TYPES["webhook"] = WebhookModule
     _MODULE_TYPES["apprise"] = AppriseModule
+    _MODULE_TYPES["sqs"] = SqsModule
 
 
 def _matches_filter(filter_value: Any, key: str) -> bool:

app/fanout/sqs.py

@@ -0,0 +1,142 @@
+"""Fanout module for Amazon SQS delivery."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import logging
+from functools import partial
+
+import boto3
+from botocore.exceptions import BotoCoreError, ClientError
+
+from app.fanout.base import FanoutModule
+
+logger = logging.getLogger(__name__)
+
+
+def _build_payload(data: dict, *, event_type: str) -> str:
+    """Serialize a fanout event into a stable JSON envelope."""
+    return json.dumps(
+        {
+            "event_type": event_type,
+            "data": data,
+        },
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+
+
+def _is_fifo_queue(queue_url: str) -> bool:
+    """Return True when the configured queue URL points at an SQS FIFO queue."""
+    return queue_url.rstrip("/").endswith(".fifo")
+
+
+def _build_message_group_id(data: dict, *, event_type: str) -> str:
+    """Choose a stable FIFO group ID from the event identity."""
+    if event_type == "message":
+        conversation_key = str(data.get("conversation_key", "")).strip()
+        if conversation_key:
+            return f"message-{conversation_key}"
+        return "message-default"
+    return "raw-packets"
+
+
+def _build_message_deduplication_id(data: dict, *, event_type: str, body: str) -> str:
+    """Choose a deterministic deduplication ID for FIFO queues."""
+    if event_type == "message":
+        message_id = data.get("id")
+        if isinstance(message_id, int):
+            return f"message-{message_id}"
+    else:
+        observation_id = data.get("observation_id")
+        if isinstance(observation_id, str) and observation_id.strip():
+            return f"raw-{observation_id}"
+        packet_id = data.get("id")
+        if isinstance(packet_id, int):
+            return f"raw-{packet_id}"
+    return hashlib.sha256(body.encode()).hexdigest()
+
+
+class SqsModule(FanoutModule):
+    """Delivers message and raw-packet events to an Amazon SQS queue."""
+
+    def __init__(self, config_id: str, config: dict, *, name: str = "") -> None:
+        super().__init__(config_id, config, name=name)
+        self._client = None
+        self._last_error: str | None = None
+
+    async def start(self) -> None:
+        kwargs: dict[str, str] = {}
+        region_name = str(self.config.get("region_name", "")).strip()
+        endpoint_url = str(self.config.get("endpoint_url", "")).strip()
+        access_key_id = str(self.config.get("access_key_id", "")).strip()
+        secret_access_key = str(self.config.get("secret_access_key", "")).strip()
+        session_token = str(self.config.get("session_token", "")).strip()
+
+        if region_name:
+            kwargs["region_name"] = region_name
+        if endpoint_url:
+            kwargs["endpoint_url"] = endpoint_url
+        if access_key_id and secret_access_key:
+            kwargs["aws_access_key_id"] = access_key_id
+            kwargs["aws_secret_access_key"] = secret_access_key
+        if session_token:
+            kwargs["aws_session_token"] = session_token
+
+        self._client = boto3.client("sqs", **kwargs)
+        self._last_error = None
+
+    async def stop(self) -> None:
+        self._client = None
+
+    async def on_message(self, data: dict) -> None:
+        await self._send(data, event_type="message")
+
+    async def on_raw(self, data: dict) -> None:
+        await self._send(data, event_type="raw_packet")
+
+    async def _send(self, data: dict, *, event_type: str) -> None:
+        if self._client is None:
+            return
+
+        queue_url = str(self.config.get("queue_url", "")).strip()
+        if not queue_url:
+            return
+
+        body = _build_payload(data, event_type=event_type)
+        request_kwargs: dict[str, object] = {
+            "QueueUrl": queue_url,
+            "MessageBody": body,
+            "MessageAttributes": {
+                "event_type": {
+                    "DataType": "String",
+                    "StringValue": event_type,
+                }
+            },
+        }
+
+        if _is_fifo_queue(queue_url):
+            request_kwargs["MessageGroupId"] = _build_message_group_id(data, event_type=event_type)
+            request_kwargs["MessageDeduplicationId"] = _build_message_deduplication_id(
+                data, event_type=event_type, body=body
+            )
+
+        try:
+            await asyncio.to_thread(partial(self._client.send_message, **request_kwargs))
+            self._last_error = None
+        except (ClientError, BotoCoreError) as exc:
+            self._last_error = str(exc)
+            logger.warning("SQS %s send error: %s", self.config_id, exc)
+        except Exception as exc:
+            self._last_error = str(exc)
+            logger.exception("Unexpected SQS send error for %s", self.config_id)
+
+    @property
+    def status(self) -> str:
+        if not str(self.config.get("queue_url", "")).strip():
+            return "disconnected"
+        if self._last_error:
+            return "error"
+        return "connected"

app/models.py

@@ -565,7 +565,7 @@ class FanoutConfig(BaseModel):
     """Configuration for a single fanout integration."""
 
     id: str
-    type: str  # 'mqtt_private' | 'mqtt_community' | 'bot' | 'webhook' | 'apprise'
+    type: str  # 'mqtt_private' | 'mqtt_community' | 'bot' | 'webhook' | 'apprise' | 'sqs'
     name: str
     enabled: bool
     config: dict

app/routers/fanout.py

@@ -13,7 +13,7 @@ from app.repository.fanout import FanoutConfigRepository
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/fanout", tags=["fanout"])
 
-_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise"}
+_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise", "sqs"}
 
 _IATA_RE = re.compile(r"^[A-Z]{3}$")
 _DEFAULT_COMMUNITY_MQTT_TOPIC_TEMPLATE = "meshcore/{IATA}/{PUBLIC_KEY}/packets"
@@ -179,6 +179,39 @@ def _validate_webhook_config(config: dict) -> None:
         raise HTTPException(status_code=400, detail="headers must be a JSON object")
 
 
+def _validate_sqs_config(config: dict) -> None:
+    """Validate sqs config blob."""
+    queue_url = str(config.get("queue_url", "")).strip()
+    if not queue_url:
+        raise HTTPException(status_code=400, detail="queue_url is required for sqs")
+    if not queue_url.startswith(("https://", "http://")):
+        raise HTTPException(status_code=400, detail="queue_url must start with http:// or https://")
+
+    endpoint_url = str(config.get("endpoint_url", "")).strip()
+    if endpoint_url and not endpoint_url.startswith(("https://", "http://")):
+        raise HTTPException(
+            status_code=400,
+            detail="endpoint_url must start with http:// or https://",
+        )
+
+    access_key_id = str(config.get("access_key_id", "")).strip()
+    secret_access_key = str(config.get("secret_access_key", "")).strip()
+    session_token = str(config.get("session_token", "")).strip()
+    has_static_keypair = bool(access_key_id) and bool(secret_access_key)
+    has_partial_keypair = bool(access_key_id) != bool(secret_access_key)
+
+    if has_partial_keypair:
+        raise HTTPException(
+            status_code=400,
+            detail="access_key_id and secret_access_key must be set together for sqs",
+        )
+    if session_token and not has_static_keypair:
+        raise HTTPException(
+            status_code=400,
+            detail="session_token requires access_key_id and secret_access_key for sqs",
+        )
+
+
 def _enforce_scope(config_type: str, scope: dict) -> dict:
     """Enforce type-specific scope constraints. Returns normalized scope."""
     if config_type == "mqtt_community":
@@ -193,7 +226,7 @@ def _enforce_scope(config_type: str, scope: dict) -> dict:
                 detail="scope.messages must be 'all', 'none', or a filter object",
             )
         return {"messages": messages, "raw_packets": "none"}
-    # For mqtt_private, validate scope values
+    # For mqtt_private and sqs, validate scope values
     messages = scope.get("messages", "all")
     if messages not in ("all", "none") and not isinstance(messages, dict):
         raise HTTPException(
@@ -240,6 +273,8 @@ async def create_fanout_config(body: FanoutConfigCreate) -> dict:
             _validate_webhook_config(body.config)
         elif body.type == "apprise":
             _validate_apprise_config(body.config)
+        elif body.type == "sqs":
+            _validate_sqs_config(body.config)
 
     scope = _enforce_scope(body.type, body.scope)
 
@@ -295,6 +330,8 @@ async def update_fanout_config(config_id: str, body: FanoutConfigUpdate) -> dict
             _validate_webhook_config(config_to_validate)
         elif existing["type"] == "apprise":
             _validate_apprise_config(config_to_validate)
+        elif existing["type"] == "sqs":
+            _validate_sqs_config(config_to_validate)
 
     updated = await FanoutConfigRepository.update(config_id, **kwargs)
     if updated is None: