From 9e6a3715ed6054471e84fbc3798489dae6a0dc10 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Fri, 3 Jul 2026 20:48:05 +0000 Subject: [PATCH] =?UTF-8?q?feat(dashboard):=20Phase=20E=20=E2=80=94=20Acti?= =?UTF-8?q?vity=20Log=20(per-mesh=20broadcast=20feed);=20remove=20subscrip?= =?UTF-8?q?tion=20backend?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace Alerts with an Activity Log fed by per-mesh broadcast logging (transport+channel+success on mesh_broadcasts_out, additive migration). Remove the entire subscription backend (commands, DM dispatch, storage, API) and its UI. Co-Authored-By: Claude Opus 4.8 (1M context) --- work/dashboard-frontend/src/App.tsx | 5 +- .../src/components/Layout.tsx | 3 +- work/dashboard-frontend/src/lib/api.ts | 23 +- .../src/pages/ActivityLog.tsx | 180 ++++++ work/dashboard-frontend/src/pages/Alerts.tsx | 563 ------------------ work/dashboard-frontend/src/pages/Config.tsx | 3 - .../src/pages/Reference.tsx | 12 - work/meshai/alert_engine.py | 11 - work/meshai/commands/dispatcher.py | 26 - work/meshai/commands/help.py | 39 +- work/meshai/commands/subscribe.py | 381 ------------ work/meshai/dashboard/api/alert_routes.py | 38 +- work/meshai/dashboard/server.py | 1 - work/meshai/main.py | 110 +--- .../notifications/pipeline/dispatcher.py | 126 ++-- work/meshai/notifications/router.py | 39 -- work/meshai/persistence/db.py | 2 +- work/meshai/persistence/migrations/v20.sql | 9 + work/meshai/subscriptions.py | 278 --------- 19 files changed, 312 insertions(+), 1537 deletions(-) create mode 100644 work/dashboard-frontend/src/pages/ActivityLog.tsx delete mode 100644 work/dashboard-frontend/src/pages/Alerts.tsx delete mode 100644 work/meshai/commands/subscribe.py create mode 100644 work/meshai/persistence/migrations/v20.sql delete mode 100644 work/meshai/subscriptions.py diff --git a/work/dashboard-frontend/src/App.tsx b/work/dashboard-frontend/src/App.tsx index f2f8b71..7ac1299 100644 --- a/work/dashboard-frontend/src/App.tsx +++ b/work/dashboard-frontend/src/App.tsx @@ -4,7 +4,7 @@ import Dashboard from './pages/Dashboard' import Mesh from './pages/Mesh' import Environment from './pages/Environment' import Config from './pages/Config' -import Alerts from './pages/Alerts' +import ActivityLog from './pages/ActivityLog' import Notifications from './pages/Notifications' import Reference from './pages/Reference' import AdapterConfig from './pages/AdapterConfig' @@ -34,7 +34,8 @@ function App() { } /> } /> } /> - } /> + } /> + } /> } /> } /> } /> diff --git a/work/dashboard-frontend/src/components/Layout.tsx b/work/dashboard-frontend/src/components/Layout.tsx index dcd936a..59207fc 100644 --- a/work/dashboard-frontend/src/components/Layout.tsx +++ b/work/dashboard-frontend/src/components/Layout.tsx @@ -5,7 +5,6 @@ import { LayoutDashboard, Radio, Cloud, - Bell, BellRing, BookOpen, Sliders, @@ -50,7 +49,7 @@ const navGroups: NavGroup[] = [ { path: '/', label: 'Dashboard', icon: LayoutDashboard }, { path: '/config', label: 'Settings', icon: Settings }, { path: '/environment', label: 'Data Feeds', icon: Cloud }, - { path: '/alerts', label: 'Alerts', icon: Bell }, + { path: '/activity', label: 'Activity Log', icon: Activity }, { path: '/places', label: 'Places', icon: MapPin }, ], }, diff --git a/work/dashboard-frontend/src/lib/api.ts b/work/dashboard-frontend/src/lib/api.ts index 50190e7..c642158 100644 --- a/work/dashboard-frontend/src/lib/api.ts +++ b/work/dashboard-frontend/src/lib/api.ts @@ -111,15 +111,18 @@ export interface AlertHistoryResponse { total: number } -export interface Subscription { +export interface ActivityEntry { id: number - user_id: string - sub_type: string - schedule_time?: string - schedule_day?: string - scope_type: string - scope_value?: string - enabled: boolean + sent_at: number | string | null // epoch seconds (int) on new rows + recipient: string | null + channel: string | number | null + text: string | null + source_event_table: string | null + source_event_pk: string | number | null + bytes_sent: number | null + ack_received: number | null + transport: string | null // 'meshtastic' | 'meshcore' | null (legacy) + success: number | null // 1 sent, 0 skip/fail, null legacy } export interface EnvStatus { @@ -294,8 +297,8 @@ export async function fetchAlertHistory( return fetchJson(`/api/alerts/history?${params.toString()}`) } -export async function fetchSubscriptions(): Promise { - return fetchJson('/api/subscriptions') +export async function fetchActivity(limit = 100): Promise { + return fetchJson(`/api/activity?limit=${limit}`) } export async function fetchEnvStatus(): Promise { diff --git a/work/dashboard-frontend/src/pages/ActivityLog.tsx b/work/dashboard-frontend/src/pages/ActivityLog.tsx new file mode 100644 index 0000000..1e3099d --- /dev/null +++ b/work/dashboard-frontend/src/pages/ActivityLog.tsx @@ -0,0 +1,180 @@ +import { useEffect, useState } from 'react' +import { Activity, Clock, CheckCircle, MinusCircle, Radio } from 'lucide-react' +import { fetchActivity, type ActivityEntry } from '@/lib/api' + +// --- helpers --------------------------------------------------------------- + +// sent_at is stored as int(time.time()) epoch SECONDS on new rows. Guard null +// and detect seconds-vs-ms so we render correct local time either way. +function formatSentAt(sent_at: number | string | null): string { + if (sent_at === null || sent_at === undefined || sent_at === '') return '—' + let d: Date + if (typeof sent_at === 'number') { + // epoch seconds -> ms (values below ~1e12 are seconds) + d = new Date(sent_at < 1e12 ? sent_at * 1000 : sent_at) + } else { + const asNum = Number(sent_at) + d = Number.isFinite(asNum) && sent_at.trim() !== '' + ? new Date(asNum < 1e12 ? asNum * 1000 : asNum) + : new Date(sent_at) + } + return isNaN(d.getTime()) ? '—' : d.toLocaleString() +} + +// Mesh badge styling per transport family. +function transportBadge(transport: string | null) { + switch (transport) { + case 'meshtastic': + return { label: 'Meshtastic', cls: 'bg-blue-500/15 text-blue-400 border border-blue-500/30' } + case 'meshcore': + return { label: 'MeshCore', cls: 'bg-green-500/15 text-green-400 border border-green-500/30' } + default: + return { label: 'Legacy', cls: 'bg-slate-500/15 text-slate-400 border border-slate-600/40' } + } +} + +// Channel label: Meshtastic index (#n) or MeshCore name; '—' when null. +function channelLabel(channel: string | number | null): string { + if (channel === null || channel === undefined || channel === '') return '—' + if (typeof channel === 'number') return `ch ${channel}` + return channel.startsWith('#') ? channel : `#${channel}` +} + +// Type/family tag derived from source_event_table (e.g. 'fires' -> 'fire'). +function familyLabel(table: string | null): string { + if (!table) return 'broadcast' + const t = table.replace(/_/g, ' ').trim() + return t.endsWith('s') ? t.slice(0, -1) : t +} + +// --- component ------------------------------------------------------------- + +export default function ActivityLog() { + const [entries, setEntries] = useState([]) + const [loading, setLoading] = useState(true) + const [error, setError] = useState(null) + + useEffect(() => { + document.title = 'Activity Log — MeshAI' + }, []) + + useEffect(() => { + let alive = true + const load = () => { + fetchActivity() + .then((data) => { + if (!alive) return + setEntries(data) + setError(null) + setLoading(false) + }) + .catch((err) => { + if (!alive) return + setError(err.message) + setLoading(false) + }) + } + load() + const interval = setInterval(load, 5000) + return () => { + alive = false + clearInterval(interval) + } + }, []) + + if (loading) { + return ( +
+
Loading activity…
+
+ ) + } + + if (error) { + return ( +
+
Error: {error}
+
+ ) + } + + return ( +
+
+
+ +

+ Activity Log +

+ + {entries.length} recent broadcast{entries.length === 1 ? '' : 's'} · newest first + +
+ + {entries.length === 0 ? ( +
+ + No outbound broadcasts recorded yet. +
+ ) : ( +
    + {entries.map((e) => { + const badge = transportBadge(e.transport) + return ( +
  • +
    + {/* status indicator */} +
    + {e.success === 1 ? ( + + ) : e.success === 0 ? ( + + ) : ( + + )} +
    + +
    + {/* meta row: mesh badge, channel, family, status */} +
    + + {badge.label} + + + {channelLabel(e.channel)} + + + {familyLabel(e.source_event_table)} + + {e.success === 1 && ( + Sent + )} + {e.success === 0 && ( + Skip + )} + {(e.success === null || e.success === undefined) && ( + — + )} +
    + + {/* message text */} +
    + {e.text || (no text)} +
    + + {/* timestamp */} +
    + + {formatSentAt(e.sent_at)} +
    +
    +
    +
    • + ) + })} +
+ )} +
+
+ ) +} diff --git a/work/dashboard-frontend/src/pages/Alerts.tsx b/work/dashboard-frontend/src/pages/Alerts.tsx deleted file mode 100644 index f62567e..0000000 --- a/work/dashboard-frontend/src/pages/Alerts.tsx +++ /dev/null @@ -1,563 +0,0 @@ -import { useEffect, useState, useCallback } from 'react' -import { - Bell, - AlertTriangle, - AlertCircle, - - CheckCircle, - Clock, - Filter, - ChevronLeft, - ChevronRight, - Radio, - Zap, - - Cloud, - Wifi, - WifiOff, - Battery, - Users, -} from 'lucide-react' -import { - fetchAlerts, - fetchAlertHistory, - fetchSubscriptions, - type Alert, - type AlertHistoryItem, - type Subscription, -} from '@/lib/api' - -interface Node { - node_num: number - node_id_hex: string - short_name: string - long_name: string -} -import { useWebSocket } from '@/hooks/useWebSocket' - -// Alert type icons mapping -const alertTypeIcons: Record = { - infra_offline: WifiOff, - infra_recovery: Wifi, - battery_warning: Battery, - battery_critical: Battery, - battery_emergency: Battery, - hf_blackout: Zap, - uhf_ducting: Radio, - weather_warning: Cloud, - weather_watch: Cloud, - new_router: Radio, - packet_flood: AlertTriangle, - sustained_high_util: AlertTriangle, - region_blackout: AlertCircle, - default: Bell, -} - -function getAlertIcon(type: string) { - return alertTypeIcons[type] || alertTypeIcons.default -} - -function getSeverityStyles(severity: string) { - switch (severity?.toLowerCase()) { - case 'immediate': - return { - bg: 'bg-red-500/10', - border: 'border-red-500', - badge: 'bg-red-500/20 text-red-400', - iconColor: 'text-red-500', - } - case 'priority': - return { - bg: 'bg-amber-500/10', - border: 'border-amber-500', - badge: 'bg-amber-500/20 text-amber-400', - iconColor: 'text-amber-500', - } - case 'routine': - default: - return { - bg: 'bg-[#f59e0b]/10', - border: 'border-[#f59e0b]', - badge: 'bg-[#f59e0b]/20 text-[#f59e0b]', - iconColor: 'text-[#f59e0b]', - } - } -} - -function formatTimeAgo(timestamp: string | number): string { - const date = typeof timestamp === 'number' ? new Date(timestamp * 1000) : new Date(timestamp) - const now = new Date() - const diffMs = now.getTime() - date.getTime() - const diffSec = Math.floor(diffMs / 1000) - const diffMin = Math.floor(diffSec / 60) - const diffHour = Math.floor(diffMin / 60) - const diffDay = Math.floor(diffHour / 24) - - if (diffSec < 60) return 'Just now' - if (diffMin < 60) return `${diffMin}m ago` - if (diffHour < 24) return `${diffHour}h ago` - return `${diffDay}d ago` -} - -function formatDateTime(timestamp: string | number): string { - const date = typeof timestamp === 'number' ? new Date(timestamp * 1000) : new Date(timestamp) - return date.toLocaleString('en-US', { - month: 'short', - day: 'numeric', - hour: '2-digit', - minute: '2-digit', - hour12: false, - }) -} - -function formatDuration(seconds: number): string { - if (seconds < 60) return `${seconds}s` - if (seconds < 3600) return `${Math.floor(seconds / 60)}m` - if (seconds < 86400) return `${Math.floor(seconds / 3600)}h ${Math.floor((seconds % 3600) / 60)}m` - return `${Math.floor(seconds / 86400)}d` -} - -// Active Alert Card Component -function ActiveAlertCard({ - alert, - onAcknowledge, -}: { - alert: Alert - onAcknowledge: (alert: Alert) => void -}) { - const styles = getSeverityStyles(alert.severity) - const Icon = getAlertIcon(alert.type) - - return ( -
-
- -
-
- - {alert.severity?.toUpperCase()} - - {alert.type} -
-
{alert.message}
-
- - - {alert.timestamp ? formatTimeAgo(alert.timestamp) : 'Just now'} - - {alert.scope_value && ( - {alert.scope_type}: {alert.scope_value} - )} -
-
- -
-
- ) -} - -// Alert History Table Component -function AlertHistoryTable({ - history, - typeFilter, - severityFilter, - onTypeFilterChange, - onSeverityFilterChange, - page, - totalPages, - onPageChange, -}: { - history: AlertHistoryItem[] - typeFilter: string - severityFilter: string - onTypeFilterChange: (v: string) => void - onSeverityFilterChange: (v: string) => void - page: number - totalPages: number - onPageChange: (p: number) => void -}) { - const alertTypes = [ - 'all', - 'infra_offline', - 'infra_recovery', - 'battery_warning', - 'battery_critical', - 'hf_blackout', - 'uhf_ducting', - 'weather_warning', - 'new_router', - 'packet_flood', - ] - - const severities = ["all", "immediate", "priority", "routine"] - - return ( -
- {/* Filters */} -
-
- - Filter: -
- - -
- - {/* Table */} -
- - - - - - - - - - - - {history.length > 0 ? ( - history.map((item, i) => { - const styles = getSeverityStyles(item.severity) - return ( - - - - - - - - ) - }) - ) : ( - - - - )} - -
TimeTypeSeverityMessageDuration
- {formatDateTime(item.timestamp)} - - {item.type.replace(/_/g, ' ')} - - - {item.severity} - - - {item.message} - - {item.duration ? formatDuration(item.duration) : '-'} -
- No alert history available -
-
- - {/* Pagination */} - {totalPages > 1 && ( -
- - Page {page} of {totalPages} - -
- - -
-
- )} -
- ) -} - -// Subscription Card Component -function SubscriptionCard({ subscription, nodes }: { subscription: Subscription; nodes: Node[] }) { - const resolveNodeName = (userId: string): string => { - const node = nodes.find(n => - n.node_id_hex === userId || - String(n.node_num) === userId || - n.short_name === userId - ) - if (node) { - return node.long_name && node.long_name !== node.short_name - ? `${node.short_name} (${node.long_name})` - : node.short_name - } - return userId - } - const formatSchedule = () => { - if (subscription.sub_type === 'alerts') { - return 'Real-time' - } - const time = subscription.schedule_time || '0000' - const hours = parseInt(time.slice(0, 2)) - const minutes = time.slice(2) - const period = hours >= 12 ? 'PM' : 'AM' - const displayHour = hours % 12 || 12 - let schedule = `${displayHour}:${minutes} ${period}` - if (subscription.sub_type === 'weekly' && subscription.schedule_day) { - schedule += ` ${subscription.schedule_day.charAt(0).toUpperCase()}${subscription.schedule_day.slice(1)}` - } - return schedule - } - - const getTypeIcon = () => { - switch (subscription.sub_type) { - case 'alerts': - return Bell - case 'daily': - return Clock - case 'weekly': - return Clock - default: - return Bell - } - } - - const Icon = getTypeIcon() - - return ( -
-
-
- -
-
-
- {subscription.sub_type.charAt(0).toUpperCase() + subscription.sub_type.slice(1)} - {subscription.scope_type !== 'mesh' && subscription.scope_value && ( - - ({subscription.scope_type}: {subscription.scope_value}) - - )} -
-
- {formatSchedule()} • {resolveNodeName(subscription.user_id)} -
-
-
-
-
- ) -} - -export default function Alerts() { - const [activeAlerts, setActiveAlerts] = useState([]) - const [history, setHistory] = useState([]) - const [subscriptions, setSubscriptions] = useState([]) - const [nodes, setNodes] = useState([]) - const [loading, setLoading] = useState(true) - const [error, setError] = useState(null) - - // Filters and pagination - const [typeFilter, setTypeFilter] = useState('all') - const [severityFilter, setSeverityFilter] = useState('all') - const [page, setPage] = useState(1) - const [totalPages, setTotalPages] = useState(1) - const pageSize = 20 - - // Acknowledged alerts (local state only) - const [acknowledged, setAcknowledged] = useState>(new Set()) - - const { lastAlert } = useWebSocket() - - // Set page title - useEffect(() => { - document.title = 'Alerts — MeshAI' - }, []) - - // Load data - useEffect(() => { - Promise.all([ - fetchAlerts().catch(() => []), - fetchAlertHistory(pageSize, 0).catch(() => ({ items: [], total: 0 })), - fetchSubscriptions().catch(() => []), - fetch('/api/nodes').then(r => r.json()).catch(() => []), - ]) - .then(([alerts, historyData, subs, nodeData]) => { - setActiveAlerts(alerts) - if (Array.isArray(historyData)) { - setHistory(historyData) - setTotalPages(1) - } else { - setHistory(historyData.items || []) - setTotalPages(Math.ceil((historyData.total || 0) / pageSize)) - } - setSubscriptions(subs) - setNodes(nodeData) - setLoading(false) - }) - .catch((err) => { - setError(err.message) - setLoading(false) - }) - }, []) - - // Handle new alerts from WebSocket - useEffect(() => { - if (lastAlert) { - setActiveAlerts((prev) => { - // Avoid duplicates - const exists = prev.some( - (a) => a.type === lastAlert.type && a.message === lastAlert.message - ) - if (exists) return prev - return [lastAlert, ...prev] - }) - } - }, [lastAlert]) - - // Reload history when filters or page change - useEffect(() => { - const offset = (page - 1) * pageSize - fetchAlertHistory(pageSize, offset, typeFilter, severityFilter) - .then((data) => { - if (Array.isArray(data)) { - setHistory(data) - setTotalPages(1) - } else { - setHistory(data.items || []) - setTotalPages(Math.ceil((data.total || 0) / pageSize)) - } - }) - .catch(() => { - // Keep current data on error - }) - }, [page, typeFilter, severityFilter]) - - const handleAcknowledge = useCallback((alert: Alert) => { - const key = `${alert.type}-${alert.message}-${alert.timestamp}` - setAcknowledged((prev) => new Set([...prev, key])) - }, []) - - // Filter out acknowledged alerts - const visibleAlerts = activeAlerts.filter((alert) => { - const key = `${alert.type}-${alert.message}-${alert.timestamp}` - return !acknowledged.has(key) - }) - - if (loading) { - return ( -
-
Loading alerts...
-
- ) - } - - if (error) { - return ( -
-
Error: {error}
-
- ) - } - - return ( -
- {/* Active Alerts */} -
-

- - Active Alerts ({visibleAlerts.length}) -

- {visibleAlerts.length > 0 ? ( -
- {visibleAlerts.map((alert, i) => ( - - ))} -
- ) : ( -
- - No active alerts — all systems nominal -
- )} -
- - {/* Alert History */} -
-

- - Alert History -

- { - setTypeFilter(v) - setPage(1) - }} - onSeverityFilterChange={(v) => { - setSeverityFilter(v) - setPage(1) - }} - page={page} - totalPages={totalPages} - onPageChange={setPage} - /> -
- - {/* Subscriptions */} -
-

- - Mesh Subscriptions ({subscriptions.length}) -

- {subscriptions.length > 0 ? ( -
- {subscriptions.map((sub) => ( - - ))} -
- ) : ( -
-

No active subscriptions.

-

- Manage subscriptions via !subscribe on mesh. Broadcasts arrive with one of three prefixes — New: (first sight), Update: (material change), or Active: (clock-driven reminder while the event is still live). See Broadcast Types and Reminder System in Reference. -

-
- )} -
-
- ) -} diff --git a/work/dashboard-frontend/src/pages/Config.tsx b/work/dashboard-frontend/src/pages/Config.tsx index 702cfc9..8f08c7d 100644 --- a/work/dashboard-frontend/src/pages/Config.tsx +++ b/work/dashboard-frontend/src/pages/Config.tsx @@ -267,9 +267,6 @@ const AVAILABLE_COMMANDS = [ { name: 'ping', description: 'Test bot responsiveness' }, { name: 'clear', description: 'Clear your conversation history' }, { name: 'reset', description: 'Reset conversation context' }, - { name: 'sub', description: 'Subscribe to scheduled reports or alerts' }, - { name: 'unsub', description: 'Remove a subscription' }, - { name: 'mysubs', description: 'List your active subscriptions' }, { name: 'alerts', description: 'Active NWS weather alerts for mesh area' }, { name: 'solar', description: 'Space weather and HF propagation conditions' }, { name: 'hf', description: 'HF radio propagation (alias for !solar)' }, diff --git a/work/dashboard-frontend/src/pages/Reference.tsx b/work/dashboard-frontend/src/pages/Reference.tsx index accb9a5..493365f 100644 --- a/work/dashboard-frontend/src/pages/Reference.tsx +++ b/work/dashboard-frontend/src/pages/Reference.tsx @@ -1186,18 +1186,6 @@ export default function Reference() { ]} /> - Subscription Commands - !subscribe, 'Lists all alert categories you can subscribe to'], - [!subscribe fire_proximity, 'Subscribe to a specific category'], - [!subscribe all, 'Subscribe to everything'], - [!unsubscribe fire_proximity, 'Unsubscribe from a category'], - [!subscriptions, "Shows what you're currently subscribed to"], - ]} - /> - Conversational

Bang commands are the short, predictable interface. For anything that diff --git a/work/meshai/alert_engine.py b/work/meshai/alert_engine.py index 2547f0b..f8cbda5 100644 --- a/work/meshai/alert_engine.py +++ b/work/meshai/alert_engine.py @@ -9,7 +9,6 @@ if TYPE_CHECKING: from .config import AlertRulesConfig, MeshIntelligenceConfig from .mesh_health import MeshHealthEngine from .mesh_reporter import MeshReporter - from .subscriptions import SubscriptionManager logger = logging.getLogger(__name__) @@ -65,14 +64,12 @@ class AlertEngine: self, health_engine: "MeshHealthEngine", reporter: "MeshReporter", - subscription_manager: "SubscriptionManager", config: "MeshIntelligenceConfig", db_path: str = "", timezone: str = "America/Boise", ): self._health = health_engine self._reporter = reporter - self._subs = subscription_manager self._rules = config.alert_rules self._critical_nodes = set(n.upper() for n in (config.critical_nodes or [])) self._db_path = db_path @@ -580,14 +577,6 @@ class AlertEngine: def clear_pending(self): self._pending_alerts = [] - def get_subscribers_for_alert(self, alert: dict) -> list[dict]: - if not self._subs: - return [] - return self._subs.get_alert_subscribers( - scope_type=alert.get("scope_type"), - scope_value=alert.get("scope_value"), - ) - def check_environmental(self, env_store) -> list[dict]: """Check environmental feeds for alertable conditions. diff --git a/work/meshai/commands/dispatcher.py b/work/meshai/commands/dispatcher.py index f89c4a6..3762c51 100644 --- a/work/meshai/commands/dispatcher.py +++ b/work/meshai/commands/dispatcher.py @@ -160,9 +160,7 @@ def create_dispatcher( mesh_reporter=None, data_store=None, health_engine=None, - subscription_manager=None, env_store=None, - notification_router=None, ) -> CommandDispatcher: """Create and populate command dispatcher with default commands. @@ -173,7 +171,6 @@ def create_dispatcher( mesh_reporter: MeshReporter instance for health commands data_store: MeshDataStore for neighbor data health_engine: MeshHealthEngine for infrastructure detection - subscription_manager: SubscriptionManager for subscription commands env_store: EnvironmentalStore for weather/propagation commands Returns: @@ -186,7 +183,6 @@ def create_dispatcher( from .status import StatusCommand from .weather import WeatherCommand from .health import HealthCommand, RegionCommand, NeighborCommand - from .subscribe import SubCommand, UnsubCommand, MySubsCommand dispatcher = CommandDispatcher(prefix=prefix, disabled_commands=disabled_commands) @@ -224,28 +220,6 @@ def create_dispatcher( alias_handler.name = alias dispatcher.register(alias_handler) - # Register subscription commands - sub_cmd = SubCommand(subscription_manager, mesh_reporter, data_store, notification_router) - dispatcher.register(sub_cmd) - for alias in getattr(sub_cmd, 'aliases', []): - alias_handler = SubCommand(subscription_manager, mesh_reporter, data_store, notification_router) - alias_handler.name = alias - dispatcher.register(alias_handler) - - unsub_cmd = UnsubCommand(subscription_manager, notification_router) - dispatcher.register(unsub_cmd) - for alias in getattr(unsub_cmd, 'aliases', []): - alias_handler = UnsubCommand(subscription_manager, notification_router) - alias_handler.name = alias - dispatcher.register(alias_handler) - - mysubs_cmd = MySubsCommand(subscription_manager, notification_router) - dispatcher.register(mysubs_cmd) - for alias in getattr(mysubs_cmd, 'aliases', []): - alias_handler = MySubsCommand(subscription_manager, notification_router) - alias_handler.name = alias - dispatcher.register(alias_handler) - # Register environmental commands if env_store: from .alerts_cmd import AlertsCommand diff --git a/work/meshai/commands/help.py b/work/meshai/commands/help.py index 71e30dc..ff71291 100644 --- a/work/meshai/commands/help.py +++ b/work/meshai/commands/help.py @@ -32,11 +32,9 @@ class HelpCommand(CommandHandler): # Group by category health_names = {"health", "region", "neighbors"} - sub_names = {"sub", "unsub", "mysubs"} health_cmds = [c for c in unique if c.name.lower() in health_names] - sub_cmds = [c for c in unique if c.name.lower() in sub_names] - other_cmds = [c for c in unique if c.name.lower() not in health_names and c.name.lower() not in sub_names and c.name.lower() != "help"] + other_cmds = [c for c in unique if c.name.lower() not in health_names and c.name.lower() != "help"] lines = ["Commands:"] @@ -46,12 +44,6 @@ class HelpCommand(CommandHandler): for c in sorted(health_cmds, key=lambda x: x.name): lines.append(f" !{c.name} - {c.description}") - if sub_cmds: - lines.append("") - lines.append("Subscriptions:") - for c in sorted(sub_cmds, key=lambda x: x.name): - lines.append(f" !{c.name} - {c.description}") - if other_cmds: lines.append("") lines.append("Other:") @@ -67,9 +59,6 @@ class HelpCommand(CommandHandler): def _command_help(self, cmd_name: str) -> str: """Detailed help for a specific command.""" aliases = { - "sub": "sub", "subscribe": "sub", "subscription": "sub", "subscriptions": "sub", - "unsub": "unsub", "unsubscribe": "unsub", - "mysubs": "mysubs", "subs": "mysubs", "health": "health", "mesh": "health", "region": "region", "reg": "region", "neighbors": "neighbors", "nbr": "neighbors", "nb": "neighbors", @@ -81,32 +70,6 @@ class HelpCommand(CommandHandler): registered = {c.name.lower() for c in self._dispatcher.get_commands()} texts = { - "sub": ( - "Subscribe to Reports & Alerts\n\n" - "Daily report:\n" - " !sub daily 6pm\n" - " !sub daily 7:30am region SCID\n" - " !sub daily 6pm node MHR\n\n" - "Weekly digest:\n" - " !sub weekly 8am sun\n\n" - "Alerts (instant DM on issues):\n" - " !sub alerts\n" - " !sub alerts region Wood River\n\n" - "Time: 6pm, 6:30pm, 1830, 18:30\n" - "Regions: SCID, SWID, Magic Valley, Twin Falls\n\n" - "Manage:\n" - " !mysubs - list yours\n" - " !unsub daily - remove daily\n" - " !unsub all - remove everything" - ), - "unsub": ( - "Unsubscribe\n\n" - " !unsub daily - remove daily report\n" - " !unsub weekly - remove weekly digest\n" - " !unsub alerts - remove alerts\n" - " !unsub all - remove everything" - ), - "mysubs": "!mysubs - list your active subscriptions", "health": ( "Mesh Health\n\n" " !health - 5-pillar health summary\n" diff --git a/work/meshai/commands/subscribe.py b/work/meshai/commands/subscribe.py deleted file mode 100644 index 6a1d0a6..0000000 --- a/work/meshai/commands/subscribe.py +++ /dev/null @@ -1,381 +0,0 @@ -"""Subscription commands for scheduled reports and alerts.""" - -from typing import TYPE_CHECKING - -from .base import CommandContext, CommandHandler - -if TYPE_CHECKING: - from ..mesh_data_store import MeshDataStore - from ..mesh_reporter import MeshReporter - from ..subscriptions import SubscriptionManager - from ..notifications.router import NotificationRouter - - -class SubCommand(CommandHandler): - """Subscribe to scheduled reports or alerts.""" - - name = "sub" - description = "Subscribe to reports or alerts" - usage = "!sub daily|weekly|alerts| [time] [day] [scope]" - aliases = ["subscribe"] - - def __init__( - self, - subscription_manager: "SubscriptionManager" = None, - mesh_reporter: "MeshReporter" = None, - data_store: "MeshDataStore" = None, - notification_router: "NotificationRouter" = None, - ): - self._sub_manager = subscription_manager - self._reporter = mesh_reporter - self._data_store = data_store - self._notification_router = notification_router - - async def execute(self, args: str, context: CommandContext) -> str: - """Handle subscription command.""" - parts = args.strip().split() - - # No args - show available alert categories - if not parts: - return self._show_categories() - - sub_type = parts[0].lower() - - # Check if it's a category subscription - if self._notification_router: - from ..notifications.categories import ALERT_CATEGORIES - if sub_type in ALERT_CATEGORIES or sub_type == "all": - return self._handle_category_subscription(sub_type, context) - - # Legacy subscription types - if sub_type not in ("daily", "weekly", "alerts"): - return self._show_categories() - - if not self._sub_manager: - return "Subscriptions not available." - - try: - if sub_type == "daily": - return self._handle_daily(parts[1:], context) - elif sub_type == "weekly": - return self._handle_weekly(parts[1:], context) - else: # alerts - return self._handle_alerts(parts[1:], context) - except ValueError as e: - return f"Error: {e}" - - def _show_categories(self) -> str: - """Show available alert categories.""" - try: - from ..notifications.categories import ALERT_CATEGORIES - except ImportError: - return self._usage_help() - - lines = ["Available alert categories:"] - for cat_id, cat_info in ALERT_CATEGORIES.items(): - lines.append(f" {cat_id} - {cat_info['description']}") - lines.append("") - lines.append("Usage:") - lines.append(" !sub - subscribe to a category") - lines.append(" !sub all - subscribe to all alerts") - lines.append(" !sub alerts - legacy mesh-wide alerts") - - return "\n".join(lines) - - def _handle_category_subscription(self, category: str, context: CommandContext) -> str: - """Handle category-based alert subscription.""" - node_id = self._get_user_id(context) - - if category == "all": - categories = [] # Empty = all categories - else: - categories = [category] - - # Add subscription via notification router - rule_name = self._notification_router.add_mesh_subscription( - node_id=node_id, - categories=categories, - ) - - if category == "all": - return "Subscribed to all alert categories. Use !unsub to remove." - else: - from ..notifications.categories import get_category - cat_info = get_category(category) - return f"Subscribed to {cat_info['name']} alerts. Use !unsub {category} to remove." - - def _usage_help(self) -> str: - """Return usage help.""" - return """Usage: -!sub daily 1830 - daily mesh report at 6:30 PM -!sub daily 1830 region SCID - daily region report -!sub weekly 0800 sun - weekly digest Sunday 8 AM -!sub alerts - mesh-wide alerts (legacy) -!sub - subscribe to alert category -!sub all - subscribe to all alerts""" - - def _handle_daily(self, args: list, context: CommandContext) -> str: - """Handle daily subscription.""" - if not args: - raise ValueError("Time required. Example: !sub daily 1830") - - schedule_time = args[0] - scope_type, scope_value = self._parse_scope(args[1:]) - scope_value = self._validate_scope(scope_type, scope_value) - - self._sub_manager.add( - user_id=self._get_user_id(context), - sub_type="daily", - schedule_time=schedule_time, - scope_type=scope_type, - scope_value=scope_value, - ) - - time_fmt = self._format_time(schedule_time) - scope_desc = self._format_scope(scope_type, scope_value) - return f"Subscribed: daily {scope_desc}report at {time_fmt}" - - def _handle_weekly(self, args: list, context: CommandContext) -> str: - """Handle weekly subscription.""" - if len(args) < 2: - raise ValueError("Time and day required. Example: !sub weekly 0800 sun") - - schedule_time = args[0] - schedule_day = args[1].lower() - scope_type, scope_value = self._parse_scope(args[2:]) - scope_value = self._validate_scope(scope_type, scope_value) - - self._sub_manager.add( - user_id=self._get_user_id(context), - sub_type="weekly", - schedule_time=schedule_time, - schedule_day=schedule_day, - scope_type=scope_type, - scope_value=scope_value, - ) - - time_fmt = self._format_time(schedule_time) - day_fmt = schedule_day.capitalize() - scope_desc = self._format_scope(scope_type, scope_value) - return f"Subscribed: weekly {scope_desc}report at {time_fmt} {day_fmt}" - - def _handle_alerts(self, args: list, context: CommandContext) -> str: - """Handle alerts subscription (legacy).""" - scope_type, scope_value = self._parse_scope(args) - scope_value = self._validate_scope(scope_type, scope_value) - - self._sub_manager.add( - user_id=self._get_user_id(context), - sub_type="alerts", - scope_type=scope_type, - scope_value=scope_value, - ) - - scope_desc = self._format_scope(scope_type, scope_value) - return f"Subscribed: alerts for {scope_desc.strip() or 'mesh'}" - - def _parse_scope(self, args: list) -> tuple[str, str]: - """Parse scope from remaining args.""" - if not args: - return "mesh", None - - scope_type = "mesh" - scope_value = None - - for i, arg in enumerate(args): - arg_lower = arg.lower() - if arg_lower == "region": - scope_type = "region" - scope_value = " ".join(args[i + 1:]) if i + 1 < len(args) else None - break - elif arg_lower == "node": - scope_type = "node" - scope_value = args[i + 1] if i + 1 < len(args) else None - break - - return scope_type, scope_value - - def _validate_scope(self, scope_type: str, scope_value: str) -> str: - """Validate and resolve scope value.""" - if scope_type == "mesh": - return None - - if not scope_value: - raise ValueError(f"Missing {scope_type} name") - - if scope_type == "region" and self._reporter: - region = self._reporter._find_region(scope_value) - if region: - return region.name - return scope_value - - if scope_type == "node" and self._reporter: - node = self._reporter._find_node(scope_value) - if not node: - raise ValueError(f"Node '{scope_value}' not found") - return node.short_name or str(node.node_num) - - return scope_value - - def _get_user_id(self, context: CommandContext) -> str: - """Extract user ID from context.""" - sender_id = context.sender_id - if sender_id.startswith("!"): - return str(int(sender_id[1:], 16)) - return sender_id - - def _format_time(self, hhmm: str) -> str: - """Format HHMM as readable time.""" - hours = int(hhmm[:2]) - minutes = int(hhmm[2:]) - period = "AM" if hours < 12 else "PM" - display_hour = hours % 12 or 12 - return f"{display_hour}:{minutes:02d} {period}" - - def _format_scope(self, scope_type: str, scope_value: str) -> str: - """Format scope for display.""" - if scope_type == "mesh" or not scope_value: - return "mesh " - return f"{scope_type} {scope_value} " - - -class UnsubCommand(CommandHandler): - """Unsubscribe from reports or alerts.""" - - name = "unsub" - description = "Remove subscription(s)" - usage = "!unsub daily|weekly|alerts||all" - aliases = ["unsubscribe"] - - def __init__( - self, - subscription_manager: "SubscriptionManager" = None, - notification_router: "NotificationRouter" = None, - ): - self._sub_manager = subscription_manager - self._notification_router = notification_router - - async def execute(self, args: str, context: CommandContext) -> str: - """Handle unsubscribe command.""" - sub_type = args.strip().lower() if args else None - - if not sub_type: - return "Usage: !unsub daily|weekly|alerts||all" - - user_id = self._get_user_id(context) - - # Check if it's a category unsubscription - if self._notification_router: - from ..notifications.categories import ALERT_CATEGORIES - if sub_type in ALERT_CATEGORIES or sub_type == "all": - self._notification_router.remove_mesh_subscription(user_id) - return "Removed alert subscriptions" - - # Legacy subscription types - if not self._sub_manager: - return "Subscriptions not available." - - if sub_type not in ("daily", "weekly", "alerts", "all"): - return f"Invalid type '{sub_type}'. Use: daily, weekly, alerts, , or all" - - removed = self._sub_manager.remove(user_id, sub_type if sub_type != "all" else None) - - if removed == 0: - return "No subscriptions found to remove" - elif sub_type == "all": - return f"Removed all {removed} subscription(s)" - else: - return f"Removed {removed} {sub_type} subscription(s)" - - def _get_user_id(self, context: CommandContext) -> str: - """Extract user ID from context.""" - sender_id = context.sender_id - if sender_id.startswith("!"): - return str(int(sender_id[1:], 16)) - return sender_id - - -class MySubsCommand(CommandHandler): - """List active subscriptions.""" - - name = "mysubs" - description = "List your subscriptions" - usage = "!mysubs" - aliases = ["subs", "subscriptions"] - - def __init__( - self, - subscription_manager: "SubscriptionManager" = None, - notification_router: "NotificationRouter" = None, - ): - self._sub_manager = subscription_manager - self._notification_router = notification_router - - async def execute(self, args: str, context: CommandContext) -> str: - """List user's subscriptions.""" - user_id = self._get_user_id(context) - lines = [] - - # Check notification router subscriptions - if self._notification_router: - categories = self._notification_router.get_node_subscriptions(user_id) - if categories: - if categories == ["all"]: - lines.append("Alert subscriptions: all categories") - else: - lines.append(f"Alert subscriptions: {', '.join(categories)}") - - # Check legacy subscriptions - if self._sub_manager: - subs = self._sub_manager.get_user_subs(user_id) - if subs: - if not lines: - lines.append("Your subscriptions:") - else: - lines.append("\nScheduled reports:") - for i, sub in enumerate(subs, 1): - lines.append(f" {i}. {self._format_sub(sub)}") - - if not lines: - return "No active subscriptions. Use !sub to subscribe." - - return "\n".join(lines) - - def _format_sub(self, sub: dict) -> str: - """Format a subscription for display.""" - sub_type = sub["sub_type"] - scope_type = sub.get("scope_type", "mesh") - scope_value = sub.get("scope_value") - - scope_desc = "" - if scope_type == "region" and scope_value: - scope_desc = f"region {scope_value} " - elif scope_type == "node" and scope_value: - scope_desc = f"node {scope_value} " - - if sub_type == "daily": - time_str = self._format_time(sub.get("schedule_time", "0000")) - return f"Daily {scope_desc}report at {time_str}" - elif sub_type == "weekly": - time_str = self._format_time(sub.get("schedule_time", "0000")) - day_str = (sub.get("schedule_day") or "").capitalize() - return f"Weekly {scope_desc}report at {time_str} {day_str}" - else: - return f"Alerts for {scope_desc.strip() or 'mesh'}" - - def _format_time(self, hhmm: str) -> str: - """Format HHMM as readable time.""" - if not hhmm or len(hhmm) != 4: - return hhmm - hours = int(hhmm[:2]) - minutes = int(hhmm[2:]) - period = "AM" if hours < 12 else "PM" - display_hour = hours % 12 or 12 - return f"{display_hour}:{minutes:02d} {period}" - - def _get_user_id(self, context: CommandContext) -> str: - """Extract user ID from context.""" - sender_id = context.sender_id - if sender_id.startswith("!"): - return str(int(sender_id[1:], 16)) - return sender_id diff --git a/work/meshai/dashboard/api/alert_routes.py b/work/meshai/dashboard/api/alert_routes.py index 77cc6d0..a377f48 100644 --- a/work/meshai/dashboard/api/alert_routes.py +++ b/work/meshai/dashboard/api/alert_routes.py @@ -56,31 +56,29 @@ async def get_alert_history( } -@router.get("/subscriptions") -async def get_subscriptions(request: Request): - """Get all alert subscriptions.""" - subscription_manager = getattr(request.app.state, "subscription_manager", None) +@router.get("/activity") +async def get_activity( + request: Request, + limit: int = Query(100, ge=1, le=500), +): + """Activity Log: most recent outbound mesh broadcasts, newest first. - if not subscription_manager: - return [] + Reads mesh_broadcasts_out from the persistence/migration DB (get_db) and + returns every column as a plain dict. Legacy rows keep NULL + transport/success. If the table doesn't exist yet, returns []. + """ + from meshai.persistence import get_db try: - subs = subscription_manager.get_all_subs() - return [ - { - "id": sub["id"], - "user_id": sub["user_id"], - "sub_type": sub["sub_type"], - "schedule_time": sub.get("schedule_time"), - "schedule_day": sub.get("schedule_day"), - "scope_type": sub.get("scope_type", "mesh"), - "scope_value": sub.get("scope_value"), - "enabled": sub.get("enabled", 1) == 1, - } - for sub in subs - ] + conn = get_db() + rows = conn.execute( + "SELECT * FROM mesh_broadcasts_out " + "ORDER BY sent_at DESC, id DESC LIMIT ?", + (limit,), + ).fetchall() except Exception: return [] + return [dict(r) for r in rows] def _map_severity(alert: dict) -> str: diff --git a/work/meshai/dashboard/server.py b/work/meshai/dashboard/server.py index 66400f5..0f4a16c 100644 --- a/work/meshai/dashboard/server.py +++ b/work/meshai/dashboard/server.py @@ -119,7 +119,6 @@ async def start_dashboard(meshai_instance: "MeshAI") -> DashboardBroadcaster: app.state.health_engine = meshai_instance.health_engine app.state.alert_engine = getattr(meshai_instance, "alert_engine", None) app.state.env_store = getattr(meshai_instance, "env_store", None) - app.state.subscription_manager = meshai_instance.subscription_manager app.state.notification_router = getattr(meshai_instance, "notification_router", None) app.state.connector = meshai_instance.connector app.state.bus = getattr(meshai_instance, "event_bus", None) diff --git a/work/meshai/main.py b/work/meshai/main.py index 3c57d24..1f5d6b3 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -45,7 +45,6 @@ class MeshAI: self.data_store = None # Replaces source_manager self.health_engine = None self.mesh_reporter = None - self.subscription_manager = None self.alert_engine = None self.notification_router = None self.event_bus = None # Notification pipeline EventBus (v0.3) @@ -53,7 +52,6 @@ class MeshAI: self.env_store = None # Environmental feeds store self._central_consumer = None # Central NATS consumer (v0.4) self._fire_pacer = None # FirePacer for rate-limited fire broadcasts - self._last_sub_check: float = 0.0 self.router: Optional[MessageRouter] = None self.responder: Optional[Responder] = None self._running = False @@ -223,12 +221,6 @@ class MeshAI: except Exception as e: logger.debug("Env refresh error: %s", e) - # Check scheduled subscriptions (every 60 seconds) - if self.subscription_manager and self.mesh_reporter: - if time.time() - self._last_sub_check >= 60: - await self._check_scheduled_subs() - self._last_sub_check = time.time() - # Periodic cleanup if time.time() - self._last_cleanup >= 3600: await self.history.cleanup_expired() @@ -326,8 +318,6 @@ class MeshAI: if self.data_store: await self.data_store.stop_mqtt_sources() self.data_store.close() - if self.subscription_manager: - self.subscription_manager.close() self._remove_pid() logger.info("MeshAI stopped") @@ -497,22 +487,13 @@ class MeshAI: else: self.mesh_reporter = None - # Subscription manager (uses same db as data_store) - if self.data_store: - from .subscriptions import SubscriptionManager - self.subscription_manager = SubscriptionManager(db_path="/data/mesh_history.db") - logger.info("Subscription manager enabled") - else: - self.subscription_manager = None - - # Alert engine (needs health engine, reporter, and subscription manager) - if self.health_engine and self.mesh_reporter and self.subscription_manager: + # Alert engine (needs health engine and reporter) + if self.health_engine and self.mesh_reporter: from .alert_engine import AlertEngine mi = self.config.mesh_intelligence self.alert_engine = AlertEngine( health_engine=self.health_engine, reporter=self.mesh_reporter, - subscription_manager=self.subscription_manager, config=mi, db_path="/data/mesh_history.db", timezone=self.config.timezone, @@ -613,9 +594,7 @@ class MeshAI: mesh_reporter=self.mesh_reporter, data_store=self.data_store, health_engine=self.health_engine, - subscription_manager=self.subscription_manager, env_store=self.env_store, - notification_router=self.notification_router, ) # Message router @@ -799,94 +778,9 @@ class MeshAI: except Exception as e: logger.error(f"Failed to send channel alert: {e}") - # Fallback: Send DMs to matching subscribers - if self.alert_engine and self.subscription_manager: - subscribers = self.alert_engine.get_subscribers_for_alert(alert) - for sub in subscribers: - user_id = sub["user_id"] - try: - await self._send_sub_dm(user_id, message) - logger.info(f"Alert DM sent to {user_id}: {alert['type']}") - except Exception as e: - logger.error(f"Failed to send alert DM to {user_id}: {e}") - if self.alert_engine: self.alert_engine.clear_pending() - async def _check_scheduled_subs(self) -> None: - """Check for and deliver due scheduled reports.""" - from datetime import datetime - from zoneinfo import ZoneInfo - - tz = ZoneInfo(self.config.timezone) - now = datetime.now(tz) - current_hhmm = now.strftime("%H%M") - current_day = now.strftime("%a").lower() - - due_subs = self.subscription_manager.get_due_subscriptions(current_hhmm, current_day) - - for sub in due_subs: - try: - # Generate report based on scope - report = self._generate_sub_report(sub) - if not report: - continue - - # Send DM to subscriber - user_id = sub["user_id"] - await self._send_sub_dm(user_id, report) - - # Mark as sent - self.subscription_manager.mark_sent(sub["id"]) - logger.info(f"Delivered {sub['sub_type']} report to {user_id}") - - except Exception as e: - logger.error(f"Error delivering subscription {sub['id']}: {e}") - - def _generate_sub_report(self, sub: dict) -> str: - """Generate report content for a subscription.""" - if not self.mesh_reporter: - return None - - sub_type = sub["sub_type"] - scope_type = sub.get("scope_type", "mesh") - scope_value = sub.get("scope_value") - - if scope_type == "region" and scope_value: - # Region-scoped report - region = self.mesh_reporter._find_region(scope_value) - if region: - return self.mesh_reporter.build_region_compact(region.name) - return None - elif scope_type == "node" and scope_value: - # Node-scoped report - return self.mesh_reporter.build_node_compact(scope_value) - else: - # Mesh-wide report - return self.mesh_reporter.build_lora_compact(scope="mesh") - - async def _send_sub_dm(self, node_num: str, message: str) -> None: - """Send a subscription DM to a node.""" - if not self.connector: - return - - # Convert node_num to destination format - try: - dest = int(node_num) - except ValueError: - dest = node_num - - # Send via responder for proper chunking - if self.responder: - await self.responder.send_response( - message, - destination=dest, - channel=0, # DM channel - ) - else: - # Fallback to direct send - self.connector.send_message(message, destination=dest) - def setup_logging(verbose: bool = False) -> None: """Configure logging.""" diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index 3569087..ba28bba 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -445,6 +445,8 @@ class Dispatcher: delivered_any = False for ch_type in ch_types: + rule = None + payload = None try: rule = self._toggle_to_rule(tog, ch_type, event) channel = self._channel_factory(rule, self._connector) @@ -458,15 +460,20 @@ class Dispatcher: if success: delivered_any = True self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{ch_type}") - # v0.5.8b post-broadcast commit. Persistence-side - # bookkeeping that should only happen when a delivery - # actually went out: mesh_broadcasts_out audit row + - # handler-supplied last_broadcast_* UPDATE callback. - self._post_broadcast_commit(event, payload, rule, ch_type) else: self._logger.warning(f"Toggle channel delivery returned False for {fam}/{ch_type}") + # v0.5.8b post-broadcast commit -> v20 per-mesh audit. + # Written ONCE PER MESH CHANNEL with its own transport+success, + # so a fan-out to both meshes yields two rows and a skip + # (deliver()==False) is still visible as success=0. The + # last_broadcast_* callback fires only when success is truthy. + self._post_broadcast_commit(event, payload, rule, ch_type, + success=bool(success)) except Exception: self._logger.exception(f"Toggle channel delivery failed for {fam}/{ch_type}") + # A crashed delivery is still a failed send -> success=0 row. + self._post_broadcast_commit(event, payload, rule, ch_type, + success=False) # ---------- Section 6 — guard commit (v0.6-4, B13 fix) ---------- # Cooldown arming + dedup recording happen ONLY after at least one @@ -600,39 +607,75 @@ class Dispatcher: success = await channel.deliver(payload, rule) except Exception: self._logger.exception( - "scheduled-broadcast: delivery raised for %s; skipping", ch_type) - continue + "scheduled-broadcast: delivery raised for %s", ch_type) + success = False if success: delivered_any = True - # Audit row -- mirrors _post_broadcast_commit for scheduled. - try: - from meshai.persistence import get_db - conn = get_db() - bytes_sent = len(text.encode("utf-8")) if text else 0 - conn.execute( - "INSERT INTO mesh_broadcasts_out(sent_at, recipient, " - "channel, text, source_event_table, source_event_pk, " - "bytes_sent, ack_received) VALUES (?,?,?,?,?,?,?,?)", - (int(time.time()), "broadcast", - rf.broadcast_channel, text, - source_event_table, str(source_event_pk), - bytes_sent, 0), - ) - except Exception: - self._logger.exception( - "scheduled-broadcast: audit row insert failed for %s", ch_type) + + # v20 per-mesh audit row. Written once per mesh channel with its + # own transport+success, so a fan-out to both meshes yields two + # rows and a skip (deliver()==False) is visible as success=0. + try: + from meshai.persistence import get_db + conn = get_db() + bytes_sent = len(text.encode("utf-8")) if text else 0 + transport, channel_id, recipient = self._audit_route(rule, ch_type) + conn.execute( + "INSERT INTO mesh_broadcasts_out(sent_at, recipient, " + "channel, text, source_event_table, source_event_pk, " + "bytes_sent, ack_received, transport, success) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + (int(time.time()), recipient, + channel_id, text, + source_event_table, str(source_event_pk), + bytes_sent, 0, + transport, 1 if success else 0), + ) + except Exception: + self._logger.exception( + "scheduled-broadcast: audit row insert failed for %s", ch_type) return delivered_any - def _post_broadcast_commit(self, event, payload, rule, ch_type: str) -> None: - """Persistence side-effects of an actually-successful broadcast. + @staticmethod + def _audit_route(rule, ch_type: str): + """Resolve (transport, channel_id, recipient) for a mesh delivery. - Inserts the mesh_broadcasts_out audit row when the handler signalled - it wants one via `event.data["_broadcast_audit"]`, then invokes the - handler-supplied `_on_broadcast_committed` callback so the handler - can refresh its own last_broadcast_* bookkeeping. Both calls are - wrapped: a bookkeeping failure must NOT undo the actual broadcast - nor break dispatch for sibling toggles. + transport is the mesh family the row belongs to ("meshtastic" / + "meshcore"); channel_id is the Meshtastic channel INDEX or the + MeshCore channel NAME; recipient is 'broadcast' or the DM target + list. Mirrors create_channel()'s delivery_type routing. + """ + if ch_type == "mesh_broadcast": + return "meshtastic", getattr(rule, "broadcast_channel", None), "broadcast" + if ch_type == "meshcore_broadcast": + return "meshcore", getattr(rule, "meshcore_channel", None), "broadcast" + if ch_type == "mesh_dm": + node_ids = list(getattr(rule, "node_ids", []) or []) + return "meshtastic", None, (",".join(map(str, node_ids)) or "dm") + if ch_type == "meshcore_dm": + contacts = list(getattr(rule, "meshcore_dm_contacts", []) or []) + return "meshcore", None, (",".join(map(str, contacts)) or "meshcore_dm") + # Unknown / non-mesh: leave transport NULL, fall back to legacy channel. + return None, getattr(rule, "broadcast_channel", None), "broadcast" + + def _post_broadcast_commit(self, event, payload, rule, ch_type: str, + *, success: bool = True) -> None: + """Persistence side-effects of a per-mesh broadcast delivery. + + Called ONCE PER MESH CHANNEL (one per delivery_type family), so a + broadcast that fans to both meshes writes TWO mesh_broadcasts_out + rows -- each carrying its own `transport` + `success` flag. The row + is written whenever the handler signalled it wants an audit trail + via `event.data["_broadcast_audit"]`, REGARDLESS of success, so a + skip/failure (e.g. MeshCore channel-not-found -> deliver()==False) + is still visible as success=0. + + The handler-supplied `_on_broadcast_committed` callback (which + refreshes last_broadcast_* bookkeeping) fires ONLY when the send + actually landed (success is truthy). Both calls are wrapped: a + bookkeeping failure must NOT undo the actual broadcast nor break + dispatch for sibling toggles. """ data = getattr(event, "data", None) or {} if not data: @@ -646,23 +689,17 @@ class Dispatcher: conn = get_db() text = payload.message if payload is not None else (event.title or "") bytes_sent = len(text.encode("utf-8")) if text else 0 - if ch_type == "mesh_dm": - node_ids = list(getattr(rule, "node_ids", []) or []) - recipient = ",".join(map(str, node_ids)) or "dm" - elif ch_type == "meshcore_dm": - contacts = list(getattr(rule, "meshcore_dm_contacts", []) or []) - recipient = ",".join(map(str, contacts)) or "meshcore_dm" - else: - recipient = "broadcast" - channel = getattr(rule, "broadcast_channel", None) + transport, channel, recipient = self._audit_route(rule, ch_type) conn.execute( "INSERT INTO mesh_broadcasts_out(sent_at, recipient, channel, " "text, source_event_table, source_event_pk, bytes_sent, " - "ack_received) VALUES (?,?,?,?,?,?,?,?)", + "ack_received, transport, success) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", ( int(committed_at), recipient, channel, text, audit.get("table"), audit.get("pk"), bytes_sent, 0, + transport, 1 if success else 0, ), ) except Exception: @@ -672,6 +709,11 @@ class Dispatcher: audit.get("table"), audit.get("pk"), ) + if not success: + # A failed/skipped send is audited above but must NOT arm the + # handler's last_broadcast_* bookkeeping. + return + cb = data.get("_on_broadcast_committed") if callable(cb): try: diff --git a/work/meshai/notifications/router.py b/work/meshai/notifications/router.py index 507cdf3..3b163d4 100644 --- a/work/meshai/notifications/router.py +++ b/work/meshai/notifications/router.py @@ -735,45 +735,6 @@ class NotificationRouter: return {"matches": False, "conditions": [], "preview": "Unknown rule type"} - def add_mesh_subscription(self, node_id: str, categories: list[str], rule_name: Optional[str] = None) -> str: - """Add a mesh DM subscription for a node.""" - if not rule_name: - rule_name = "sub_%s" % node_id - - for rule in self._rules: - if rule.get("name") == rule_name: - rule["categories"] = categories if categories else [] - rule["node_ids"] = [node_id] - return rule_name - - self._rules.append({ - "name": rule_name, - "enabled": True, - "trigger_type": "condition", - "categories": categories if categories else [], - "min_severity": "priority", - "delivery_type": "mesh_dm", - "node_ids": [node_id], - "cooldown_minutes": 10, - }) - - return rule_name - - def remove_mesh_subscription(self, node_id: str) -> bool: - """Remove a mesh subscription for a node.""" - rule_name = "sub_%s" % node_id - self._rules = [r for r in self._rules if r.get("name") != rule_name] - return True - - def get_node_subscriptions(self, node_id: str) -> list[str]: - """Get categories a node is subscribed to.""" - rule_name = "sub_%s" % node_id - for rule in self._rules: - if rule.get("name") == rule_name: - categories = rule.get("categories", []) - return categories if categories else ["all"] - return [] - async def generate_report(self, report_type: str, env_store, health_engine) -> str: """Generate an LLM-summarized report from current data.""" context_parts = [] diff --git a/work/meshai/persistence/db.py b/work/meshai/persistence/db.py index fd977ef..82c8f14 100644 --- a/work/meshai/persistence/db.py +++ b/work/meshai/persistence/db.py @@ -30,7 +30,7 @@ logger = logging.getLogger(__name__) DEFAULT_DB_PATH = "/data/meshai.sqlite" MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH" -SCHEMA_VERSION = 19 +SCHEMA_VERSION = 20 SCHEMA_META_TABLE = "schema_meta" MIGRATIONS_DIR = Path(__file__).parent / "migrations" diff --git a/work/meshai/persistence/migrations/v20.sql b/work/meshai/persistence/migrations/v20.sql new file mode 100644 index 0000000..2185818 --- /dev/null +++ b/work/meshai/persistence/migrations/v20.sql @@ -0,0 +1,9 @@ +-- v20: per-mesh broadcast audit. transport + success columns on +-- mesh_broadcasts_out so each SEND records which mesh it went to and +-- whether it landed. A broadcast fanning to BOTH meshes writes one row +-- per mesh, each with its own success flag. +-- Nullable, no index, no backfill. Legacy rows keep NULL transport/success +-- (Activity Log treats NULL as legacy/meshtastic-unknown). + +ALTER TABLE mesh_broadcasts_out ADD COLUMN transport TEXT; +ALTER TABLE mesh_broadcasts_out ADD COLUMN success INTEGER; diff --git a/work/meshai/subscriptions.py b/work/meshai/subscriptions.py deleted file mode 100644 index 1695c70..0000000 --- a/work/meshai/subscriptions.py +++ /dev/null @@ -1,278 +0,0 @@ -"""Subscription management for scheduled reports and alerts.""" - -import logging -import sqlite3 -import time -from typing import Optional - -logger = logging.getLogger(__name__) - -# Valid subscription types -VALID_SUB_TYPES = {"daily", "weekly", "alerts"} -VALID_DAYS = {"mon", "tue", "wed", "thu", "fri", "sat", "sun"} -VALID_SCOPE_TYPES = {"mesh", "region", "node"} - - -class SubscriptionManager: - """Manages user subscriptions with SQLite storage.""" - - def __init__(self, db_path: str): - """Initialize subscription manager. - - Args: - db_path: Path to SQLite database (same as mesh_history.db) - """ - self._db_path = db_path - self._db: Optional[sqlite3.Connection] = None - self._init_db() - - def _init_db(self): - """Initialize database connection and schema.""" - self._db = sqlite3.connect(self._db_path, check_same_thread=False) - self._db.row_factory = sqlite3.Row - - self._db.executescript(""" - CREATE TABLE IF NOT EXISTS subscriptions ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - user_id TEXT NOT NULL, - sub_type TEXT NOT NULL, - schedule_time TEXT, - schedule_day TEXT, - scope_type TEXT DEFAULT 'mesh', - scope_value TEXT, - created_at REAL NOT NULL, - last_sent REAL DEFAULT 0, - enabled INTEGER DEFAULT 1 - ); - CREATE INDEX IF NOT EXISTS idx_sub_user ON subscriptions(user_id); - CREATE INDEX IF NOT EXISTS idx_sub_type ON subscriptions(sub_type); - """) - self._db.commit() - logger.info("Subscription manager initialized") - - def _row_to_dict(self, row: sqlite3.Row) -> dict: - """Convert sqlite Row to dict.""" - return dict(row) - - def add(self, user_id: str, sub_type: str, schedule_time: str = None, - schedule_day: str = None, scope_type: str = "mesh", - scope_value: str = None) -> dict: - """Add a subscription. - - Args: - user_id: Subscriber node_num - sub_type: "daily", "weekly", or "alerts" - schedule_time: HHMM format (required for daily/weekly) - schedule_day: mon-sun (required for weekly) - scope_type: "mesh", "region", or "node" - scope_value: Region name or node identifier - - Returns: - Created subscription dict - - Raises: - ValueError: If validation fails - """ - # Validate sub_type - if sub_type not in VALID_SUB_TYPES: - raise ValueError(f"Invalid type '{sub_type}'. Use: daily, weekly, or alerts") - - # Validate schedule_time for daily/weekly - if sub_type in ("daily", "weekly"): - if not schedule_time: - raise ValueError(f"Time required for {sub_type} subscription. Use HHMM format (e.g., 1830)") - if not self._validate_time(schedule_time): - raise ValueError("Invalid time format. Use HHMM (e.g., 1830 for 6:30 PM)") - - # Validate schedule_day for weekly - if sub_type == "weekly": - if not schedule_day: - raise ValueError("Day required for weekly subscription. Use: mon, tue, wed, thu, fri, sat, sun") - if schedule_day.lower() not in VALID_DAYS: - raise ValueError("Invalid day. Use: mon, tue, wed, thu, fri, sat, sun") - schedule_day = schedule_day.lower() - - # Validate scope_type - if scope_type not in VALID_SCOPE_TYPES: - raise ValueError(f"Invalid scope '{scope_type}'. Use: mesh, region, or node") - - # Check for duplicates - existing = self._db.execute(""" - SELECT id FROM subscriptions - WHERE user_id = ? AND sub_type = ? AND scope_type = ? - AND (scope_value = ? OR (scope_value IS NULL AND ? IS NULL)) - AND enabled = 1 - """, (user_id, sub_type, scope_type, scope_value, scope_value)).fetchone() - - if existing: - scope_desc = f" for {scope_type} {scope_value}" if scope_value else "" - raise ValueError(f"Already subscribed to {sub_type}{scope_desc}") - - # Insert subscription - now = time.time() - cursor = self._db.execute(""" - INSERT INTO subscriptions (user_id, sub_type, schedule_time, schedule_day, - scope_type, scope_value, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?) - """, (user_id, sub_type, schedule_time, schedule_day, scope_type, scope_value, now)) - self._db.commit() - - sub_id = cursor.lastrowid - return self._get_by_id(sub_id) - - def _validate_time(self, time_str: str) -> bool: - """Validate HHMM time format.""" - if not time_str or len(time_str) != 4 or not time_str.isdigit(): - return False - hours = int(time_str[:2]) - minutes = int(time_str[2:]) - return 0 <= hours <= 23 and 0 <= minutes <= 59 - - def _get_by_id(self, sub_id: int) -> dict: - """Get subscription by ID.""" - row = self._db.execute( - "SELECT * FROM subscriptions WHERE id = ?", (sub_id,) - ).fetchone() - return self._row_to_dict(row) if row else None - - def remove(self, user_id: str, sub_type: str = None) -> int: - """Remove subscription(s). - - Args: - user_id: Subscriber node_num - sub_type: "daily", "weekly", "alerts", or None for all - - Returns: - Number of subscriptions removed - """ - if sub_type and sub_type != "all": - cursor = self._db.execute( - "DELETE FROM subscriptions WHERE user_id = ? AND sub_type = ?", - (user_id, sub_type) - ) - else: - cursor = self._db.execute( - "DELETE FROM subscriptions WHERE user_id = ?", - (user_id,) - ) - self._db.commit() - return cursor.rowcount - - def get_user_subs(self, user_id: str) -> list[dict]: - """Get all subscriptions for a user.""" - rows = self._db.execute( - "SELECT * FROM subscriptions WHERE user_id = ? AND enabled = 1 ORDER BY created_at", - (user_id,) - ).fetchall() - return [self._row_to_dict(r) for r in rows] - - def get_due_subscriptions(self, current_time_hhmm: str, current_day: str) -> list[dict]: - """Get subscriptions that should fire right now. - - Args: - current_time_hhmm: Current time as "HHMM" (e.g., "1830") - current_day: Current day as 3-letter lowercase (e.g., "sun") - - Returns: - List of subscription dicts that are due - """ - now = time.time() - due = [] - - # Get all daily/weekly subscriptions - rows = self._db.execute(""" - SELECT * FROM subscriptions - WHERE sub_type IN ('daily', 'weekly') AND enabled = 1 - """).fetchall() - - current_minutes = int(current_time_hhmm[:2]) * 60 + int(current_time_hhmm[2:]) - - for row in rows: - sub = self._row_to_dict(row) - schedule_time = sub.get("schedule_time") - if not schedule_time: - continue - - schedule_minutes = int(schedule_time[:2]) * 60 + int(schedule_time[2:]) - - # 5-minute matching window - if abs(schedule_minutes - current_minutes) > 5: - continue - - sub_type = sub["sub_type"] - last_sent = sub.get("last_sent", 0) or 0 - - if sub_type == "daily": - # Don't fire if sent within last 23 hours - if now - last_sent < 23 * 3600: - continue - due.append(sub) - - elif sub_type == "weekly": - # Check day matches - schedule_day = sub.get("schedule_day", "").lower() - if schedule_day != current_day.lower(): - continue - # Don't fire if sent within last 6 days - if now - last_sent < 6 * 24 * 3600: - continue - due.append(sub) - - return due - - def get_alert_subscribers(self, scope_type: str = None, scope_value: str = None) -> list[dict]: - """Get users subscribed to alerts matching a scope. - - Args: - scope_type: "mesh", "region", or "node" - scope_value: Region name or node identifier - - Returns: - List of subscription dicts where scope matches - """ - # Get all alert subscriptions - rows = self._db.execute(""" - SELECT * FROM subscriptions - WHERE sub_type = 'alerts' AND enabled = 1 - """).fetchall() - - matching = [] - for row in rows: - sub = self._row_to_dict(row) - sub_scope = sub.get("scope_type", "mesh") - sub_value = sub.get("scope_value") - - # Mesh scope gets ALL alerts - if sub_scope == "mesh": - matching.append(sub) - # Region scope gets alerts for that region - elif sub_scope == "region" and scope_type == "region": - if sub_value and scope_value and sub_value.lower() == scope_value.lower(): - matching.append(sub) - # Node scope gets alerts for that node - elif sub_scope == "node" and scope_type == "node": - if sub_value and scope_value and sub_value.lower() == scope_value.lower(): - matching.append(sub) - - return matching - - def mark_sent(self, subscription_id: int): - """Update last_sent timestamp to now.""" - self._db.execute( - "UPDATE subscriptions SET last_sent = ? WHERE id = ?", - (time.time(), subscription_id) - ) - self._db.commit() - - def get_all_subs(self) -> list[dict]: - """Get all subscriptions (for admin view).""" - rows = self._db.execute( - "SELECT * FROM subscriptions WHERE enabled = 1 ORDER BY user_id, created_at" - ).fetchall() - return [self._row_to_dict(r) for r in rows] - - def close(self): - """Close database connection.""" - if self._db: - self._db.close() - self._db = None