adam/Remote-Terminal-for-MeshCore
1
0
Fork 0
mirror of https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git synced 2026-08-07 17:23:05 +02:00
Code Issues Packages Projects Releases Wiki Activity

Move bots into Fanout & Forwarding

Browse Source
This commit is contained in:
Jack Kingsman
2026-03-05 18:14:03 -08:00
parent 7cd54d14d8
commit 5ecb63fde9
27 changed files with 671 additions and 908 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/event_handlers.py
-18
View File
@@ -1,4 +1,3 @@
import asyncio
import logging
import time
from typing import TYPE_CHECKING
@@ -155,23 +154,6 @@ async def on_contact_message(event: "Event") -> None:
if contact:
await ContactRepository.update_last_contacted(sender_pubkey, received_at)
# Run bot if enabled
from app.bot import run_bot_for_message
asyncio.create_task(
run_bot_for_message(
sender_name=contact.name if contact else None,
sender_key=sender_pubkey,
message_text=payload.get("text", ""),
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=payload.get("sender_timestamp"),
path=payload.get("path"),
is_outgoing=False,
)
)
async def on_rx_log_data(event: "Event") -> None:
"""Store raw RF packet data and process via centralized packet processor.
app/fanout/bot.py
+120
View File
@@ -0,0 +1,120 @@
"""Fanout module wrapping bot execution logic."""
from __future__ import annotations
import asyncio
import logging
from app.fanout.base import FanoutModule
logger = logging.getLogger(__name__)
class BotModule(FanoutModule):
"""Wraps a single bot's code execution and response routing.
Each BotModule represents one bot configuration. It receives decoded
messages via ``on_message``, executes the bot's Python code in a
background task (after a 2-second settle delay), and sends any response
back through the radio.
"""
def __init__(self, config_id: str, config: dict, *, name: str = "Bot") -> None:
super().__init__(config_id, config)
self._name = name
async def on_message(self, data: dict) -> None:
"""Kick off bot execution in a background task so we don't block dispatch."""
asyncio.create_task(self._run_for_message(data))
async def _run_for_message(self, data: dict) -> None:
from app.bot import BOT_EXECUTION_TIMEOUT, execute_bot_code, process_bot_response
code = self.config.get("code", "")
if not code or not code.strip():
return
msg_type = data.get("type", "")
is_dm = msg_type == "PRIV"
# Extract bot parameters from broadcast data
if is_dm:
conversation_key = data.get("conversation_key", "")
sender_key = conversation_key
is_outgoing = data.get("outgoing", False)
message_text = data.get("text", "")
channel_key = None
channel_name = None
# Look up sender name from contacts
from app.repository import ContactRepository
contact = await ContactRepository.get_by_key(conversation_key)
sender_name = contact.name if contact else None
else:
conversation_key = data.get("conversation_key", "")
sender_key = None
is_outgoing = False
sender_name = data.get("sender_name")
channel_key = conversation_key
# Look up channel name
from app.repository import ChannelRepository
channel = await ChannelRepository.get_by_key(conversation_key)
channel_name = channel.name if channel else None
# Strip "sender: " prefix from channel message text
text = data.get("text", "")
if sender_name and text.startswith(f"{sender_name}: "):
message_text = text[len(f"{sender_name}: ") :]
else:
message_text = text
sender_timestamp = data.get("sender_timestamp")
path_value = data.get("path")
# Message model serializes paths as list of dicts; extract first path string
if path_value is None:
paths = data.get("paths")
if paths and isinstance(paths, list) and len(paths) > 0:
path_value = paths[0].get("path") if isinstance(paths[0], dict) else None
# Wait for message to settle (allows retransmissions to be deduped)
await asyncio.sleep(2)
# Execute bot code in thread pool with timeout
from app.bot import _bot_executor, _bot_semaphore
async with _bot_semaphore:
loop = asyncio.get_event_loop()
try:
response = await asyncio.wait_for(
loop.run_in_executor(
_bot_executor,
execute_bot_code,
code,
sender_name,
sender_key,
message_text,
is_dm,
channel_key,
channel_name,
sender_timestamp,
path_value,
is_outgoing,
),
timeout=BOT_EXECUTION_TIMEOUT,
)
except asyncio.TimeoutError:
logger.warning("Bot '%s' execution timed out", self._name)
return
except Exception as e:
logger.warning("Bot '%s' execution error: %s", self._name, e)
return
if response:
await process_bot_response(response, is_dm, sender_key or "", channel_key)
@property
def status(self) -> str:
return "connected"
app/fanout/manager.py
+14 -1
View File
@@ -17,11 +17,13 @@ def _register_module_types() -> None:
"""Lazily populate the type registry to avoid circular imports."""
if _MODULE_TYPES:
return
from app.fanout.bot import BotModule
from app.fanout.mqtt_community import MqttCommunityModule
from app.fanout.mqtt_private import MqttPrivateModule
_MODULE_TYPES["mqtt_private"] = MqttPrivateModule
_MODULE_TYPES["mqtt_community"] = MqttCommunityModule
_MODULE_TYPES["bot"] = BotModule
def _scope_matches_message(scope: dict, data: dict) -> bool:
@@ -80,13 +82,24 @@ class FanoutManager:
config_blob = cfg["config"]
scope = cfg["scope"]
# Skip bot modules when bots are disabled server-wide
if config_type == "bot":
from app.config import settings as server_settings
if server_settings.disable_bots:
logger.info("Skipping bot module %s (bots disabled by server config)", config_id)
return
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)
if config_type == "bot":
module = cls(config_id, config_blob, name=cfg.get("name", "Bot"))
else:
module = cls(config_id, config_blob)
await module.start()
self._modules[config_id] = (module, scope)
logger.info(
app/migrations.py
+65
View File
@@ -289,6 +289,13 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
await set_version(conn, 36)
applied += 1
# Migration 37: Migrate bots from app_settings to fanout_configs
if version < 37:
logger.info("Applying migration 37: migrate bots to fanout_configs")
await _migrate_037_bots_to_fanout(conn)
await set_version(conn, 37)
applied += 1
if applied > 0:
logger.info(
"Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
@@ -2149,3 +2156,61 @@ async def _migrate_036_create_fanout_configs(conn: aiosqlite.Connection) -> None
logger.info("Migrated community MQTT settings to fanout_configs")
await conn.commit()
async def _migrate_037_bots_to_fanout(conn: aiosqlite.Connection) -> None:
"""Migrate bots from app_settings.bots JSON to fanout_configs rows."""
import json
import uuid
try:
cursor = await conn.execute("SELECT bots FROM app_settings WHERE id = 1")
row = await cursor.fetchone()
except Exception:
row = None
if row is None:
await conn.commit()
return
bots_json = row["bots"] or "[]"
try:
bots = json.loads(bots_json)
except (json.JSONDecodeError, TypeError):
bots = []
if not bots:
await conn.commit()
return
import time
now = int(time.time())
# Use sort_order starting at 200 to place bots after MQTT configs (0-99)
for i, bot in enumerate(bots):
bot_name = bot.get("name") or f"Bot {i + 1}"
bot_enabled = bool(bot.get("enabled", False))
bot_code = bot.get("code", "")
config_blob = json.dumps({"code": bot_code})
scope = json.dumps({"messages": "all", "raw_packets": "none"})
await conn.execute(
"""
INSERT INTO fanout_configs (id, type, name, enabled, config, scope, sort_order, created_at)
VALUES (?, 'bot', ?, ?, ?, ?, ?, ?)
""",
(
str(uuid.uuid4()),
bot_name,
1 if bot_enabled else 0,
config_blob,
scope,
200 + i,
now,
),
)
logger.info("Migrated bot '%s' to fanout_configs (enabled=%s)", bot_name, bot_enabled)
await conn.commit()
app/packet_processor.py
+2 -38
View File
@@ -195,7 +195,7 @@ async def create_message_from_decrypted(
# Use "is not None" to include empty string (direct/0-hop messages)
paths = [MessagePath(path=path or "", received_at=received)] if path is not None else None
# Broadcast new message to connected clients
# Broadcast new message to connected clients (and fanout modules when realtime)
broadcast_event(
"message",
Message(
@@ -212,24 +212,6 @@ async def create_message_from_decrypted(
realtime=trigger_bot,
)
# Run bot if enabled (for incoming channel messages, not historical decryption)
if trigger_bot:
from app.bot import run_bot_for_message
asyncio.create_task(
run_bot_for_message(
sender_name=sender,
sender_key=None, # Channel messages don't have a sender public key
message_text=message_text,
is_dm=False,
channel_key=channel_key_normalized,
channel_name=channel_name,
sender_timestamp=timestamp,
path=path,
is_outgoing=False,
)
)
return msg_id
@@ -318,7 +300,7 @@ async def create_dm_message_from_decrypted(
# Build paths array for broadcast
paths = [MessagePath(path=path or "", received_at=received)] if path is not None else None
# Broadcast new message to connected clients
# Broadcast new message to connected clients (and fanout modules when realtime)
broadcast_event(
"message",
Message(
@@ -339,24 +321,6 @@ async def create_dm_message_from_decrypted(
# Update contact's last_contacted timestamp (for sorting)
await ContactRepository.update_last_contacted(conversation_key, received)
# Run bot if enabled (for all real-time DMs, including our own outgoing messages)
if trigger_bot:
from app.bot import run_bot_for_message
asyncio.create_task(
run_bot_for_message(
sender_name=contact.name if contact else None,
sender_key=their_public_key,
message_text=decrypted.message,
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=decrypted.timestamp,
path=path,
is_outgoing=outgoing,
)
)
return msg_id
app/routers/fanout.py
+25 -3
View File
@@ -6,13 +6,13 @@ import re
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from app.config import settings as server_settings
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"}
_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot"}
_IATA_RE = re.compile(r"^[A-Z]{3}$")
@@ -51,11 +51,26 @@ def _validate_mqtt_community_config(config: dict) -> None:
)
def _validate_bot_config(config: dict) -> None:
"""Validate bot config blob (syntax-check the code)."""
code = config.get("code", "")
if not code or not code.strip():
raise HTTPException(status_code=400, detail="Bot code cannot be empty")
try:
compile(code, "<bot_code>", "exec")
except SyntaxError as e:
raise HTTPException(
status_code=400,
detail=f"Bot code has syntax error at line {e.lineno}: {e.msg}",
) from None
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"}
if config_type == "bot":
return {"messages": "all", "raw_packets": "none"}
# For mqtt_private, validate scope values
messages = scope.get("messages", "all")
if messages not in ("all", "none") and not isinstance(messages, dict):
@@ -81,6 +96,9 @@ async def create_fanout_config(body: FanoutConfigCreate) -> dict:
detail=f"Invalid type '{body.type}'. Must be one of: {', '.join(sorted(_VALID_TYPES))}",
)
if body.type == "bot" and server_settings.disable_bots:
raise HTTPException(status_code=403, detail="Bot system disabled by server configuration")
# Only validate config when creating as enabled — disabled configs
# are drafts the user hasn't finished configuring yet.
if body.enabled:
@@ -88,6 +106,8 @@ async def create_fanout_config(body: FanoutConfigCreate) -> dict:
_validate_mqtt_private_config(body.config)
elif body.type == "mqtt_community":
_validate_mqtt_community_config(body.config)
elif body.type == "bot":
_validate_bot_config(body.config)
scope = _enforce_scope(body.type, body.scope)
@@ -134,6 +154,8 @@ async def update_fanout_config(config_id: str, body: FanoutConfigUpdate) -> dict
_validate_mqtt_private_config(config_to_validate)
elif existing["type"] == "mqtt_community":
_validate_mqtt_community_config(config_to_validate)
elif existing["type"] == "bot":
_validate_bot_config(config_to_validate)
updated = await FanoutConfigRepository.update(config_id, **kwargs)
if updated is None:
app/routers/messages.py
+1 -35
View File
@@ -1,4 +1,3 @@
import asyncio
import logging
import time
@@ -176,25 +175,9 @@ async def send_direct_message(request: SendDirectMessageRequest) -> Message:
)
# Broadcast so all connected clients (not just sender) see the outgoing message immediately.
# Fanout modules (including bots) are triggered via broadcast_event's realtime dispatch.
broadcast_event("message", message.model_dump())
# Trigger bots for outgoing DMs (runs in background, doesn't block response)
from app.bot import run_bot_for_message
asyncio.create_task(
run_bot_for_message(
sender_name=None,
sender_key=db_contact.public_key.lower(),
message_text=request.text,
is_dm=True,
channel_key=None,
channel_name=None,
sender_timestamp=now,
path=None,
is_outgoing=True,
)
)
return message
@@ -335,23 +318,6 @@ async def send_channel_message(request: SendChannelMessageRequest) -> Message:
sender_key=our_public_key,
)
# Trigger bots for outgoing channel messages (runs in background, doesn't block response)
from app.bot import run_bot_for_message
asyncio.create_task(
run_bot_for_message(
sender_name=radio_name or None,
sender_key=None,
message_text=request.text,
is_dm=False,
channel_key=channel_key_upper,
channel_name=db_channel.name,
sender_timestamp=now,
path=None,
is_outgoing=True,
)
)
return message
app/routers/settings.py
+2 -37
View File
@@ -2,38 +2,16 @@ import asyncio
import logging
from typing import Literal
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter
from pydantic import BaseModel, Field
from app.config import settings as server_settings
from app.models import AppSettings, BotConfig
from app.models import AppSettings
from app.repository import AppSettingsRepository
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/settings", tags=["settings"])
def validate_bot_code(code: str, bot_name: str | None = None) -> None:
"""Validate bot code syntax. Raises HTTPException on error."""
if not code or not code.strip():
return # Empty code is valid (disables bot)
try:
compile(code, "<bot_code>", "exec")
except SyntaxError as e:
name_part = f"'{bot_name}' " if bot_name else ""
raise HTTPException(
status_code=400,
detail=f"Bot {name_part}has syntax error at line {e.lineno}: {e.msg}",
) from None
def validate_all_bots(bots: list[BotConfig]) -> None:
"""Validate all bots' code syntax. Raises HTTPException on first error."""
for bot in bots:
validate_bot_code(bot.code, bot.name)
class AppSettingsUpdate(BaseModel):
max_radio_contacts: int | None = Field(
default=None,
@@ -56,10 +34,6 @@ class AppSettingsUpdate(BaseModel):
ge=0,
description="Periodic advertisement interval in seconds (0 = disabled, minimum 3600)",
)
bots: list[BotConfig] | None = Field(
default=None,
description="List of bot configurations",
)
flood_scope: str | None = Field(
default=None,
description="Outbound flood scope / region name (empty = disabled)",
@@ -140,15 +114,6 @@ async def update_settings(update: AppSettingsUpdate) -> AppSettings:
logger.info("Updating advert_interval to %d", interval)
kwargs["advert_interval"] = interval
if update.bots is not None:
if server_settings.disable_bots:
raise HTTPException(
status_code=403, detail="Bot system disabled by server configuration"
)
validate_all_bots(update.bots)
logger.info("Updating bots (count=%d)", len(update.bots))
kwargs["bots"] = update.bots
# Block lists
if update.blocked_keys is not None:
kwargs["blocked_keys"] = [k.lower() for k in update.blocked_keys]