mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
feat(dashboard): Phase E — Activity Log (per-mesh broadcast feed); remove subscription backend
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) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson
parent
c7b3a68f57
commit
9e6a3715ed
19 changed files with 312 additions and 1537 deletions
|
|
@ -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() {
|
|||
<Route path="/" element={<Dashboard />} />
|
||||
<Route path="/environment" element={<Environment />} />
|
||||
<Route path="/config" element={<Config />} />
|
||||
<Route path="/alerts" element={<Alerts />} />
|
||||
<Route path="/alerts" element={<ActivityLog />} />
|
||||
<Route path="/activity" element={<ActivityLog />} />
|
||||
<Route path="/notifications" element={<Notifications />} />
|
||||
<Route path="/reference" element={<Reference />} />
|
||||
<Route path="/adapter-config" element={<AdapterConfig />} />
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
],
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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<AlertHistoryResponse | AlertHistoryItem[]>(`/api/alerts/history?${params.toString()}`)
|
||||
}
|
||||
|
||||
export async function fetchSubscriptions(): Promise<Subscription[]> {
|
||||
return fetchJson<Subscription[]>('/api/subscriptions')
|
||||
export async function fetchActivity(limit = 100): Promise<ActivityEntry[]> {
|
||||
return fetchJson<ActivityEntry[]>(`/api/activity?limit=${limit}`)
|
||||
}
|
||||
|
||||
export async function fetchEnvStatus(): Promise<EnvStatus> {
|
||||
|
|
|
|||
180
work/dashboard-frontend/src/pages/ActivityLog.tsx
Normal file
180
work/dashboard-frontend/src/pages/ActivityLog.tsx
Normal file
|
|
@ -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<ActivityEntry[]>([])
|
||||
const [loading, setLoading] = useState(true)
|
||||
const [error, setError] = useState<string | null>(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 (
|
||||
<div className="flex items-center justify-center h-64">
|
||||
<div className="text-slate-400">Loading activity…</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
if (error) {
|
||||
return (
|
||||
<div className="flex items-center justify-center h-64">
|
||||
<div className="text-red-400">Error: {error}</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="space-y-4">
|
||||
<div className="bg-bg-card border border-border">
|
||||
<div className="p-4 border-b border-border flex items-center gap-2">
|
||||
<Activity size={14} className="text-[#f59e0b]" />
|
||||
<h2 className="text-sm font-medium text-slate-300">
|
||||
Activity Log
|
||||
</h2>
|
||||
<span className="text-xs text-slate-500 ml-auto">
|
||||
{entries.length} recent broadcast{entries.length === 1 ? '' : 's'} · newest first
|
||||
</span>
|
||||
</div>
|
||||
|
||||
{entries.length === 0 ? (
|
||||
<div className="flex items-center gap-2 text-slate-500 p-8">
|
||||
<Radio size={18} />
|
||||
<span>No outbound broadcasts recorded yet.</span>
|
||||
</div>
|
||||
) : (
|
||||
<ul className="divide-y divide-border">
|
||||
{entries.map((e) => {
|
||||
const badge = transportBadge(e.transport)
|
||||
return (
|
||||
<li key={e.id} className="p-4 hover:bg-bg-hover transition-colors">
|
||||
<div className="flex items-start gap-3">
|
||||
{/* status indicator */}
|
||||
<div className="pt-0.5">
|
||||
{e.success === 1 ? (
|
||||
<CheckCircle size={16} className="text-green-500" />
|
||||
) : e.success === 0 ? (
|
||||
<MinusCircle size={16} className="text-amber-500" />
|
||||
) : (
|
||||
<MinusCircle size={16} className="text-slate-600" />
|
||||
)}
|
||||
</div>
|
||||
|
||||
<div className="flex-1 min-w-0">
|
||||
{/* meta row: mesh badge, channel, family, status */}
|
||||
<div className="flex items-center flex-wrap gap-2 mb-1">
|
||||
<span className={`text-xs px-2 py-0.5 rounded-full ${badge.cls}`}>
|
||||
{badge.label}
|
||||
</span>
|
||||
<span className="text-xs px-2 py-0.5 rounded-full bg-bg-hover text-slate-400 border border-border">
|
||||
{channelLabel(e.channel)}
|
||||
</span>
|
||||
<span className="text-xs px-2 py-0.5 rounded-full bg-[#f59e0b]/10 text-[#f59e0b]">
|
||||
{familyLabel(e.source_event_table)}
|
||||
</span>
|
||||
{e.success === 1 && (
|
||||
<span className="text-xs text-green-500">Sent</span>
|
||||
)}
|
||||
{e.success === 0 && (
|
||||
<span className="text-xs text-amber-500">Skip</span>
|
||||
)}
|
||||
{(e.success === null || e.success === undefined) && (
|
||||
<span className="text-xs text-slate-500">—</span>
|
||||
)}
|
||||
</div>
|
||||
|
||||
{/* message text */}
|
||||
<div className="text-sm text-slate-200 break-words whitespace-pre-wrap">
|
||||
{e.text || <span className="text-slate-500 italic">(no text)</span>}
|
||||
</div>
|
||||
|
||||
{/* timestamp */}
|
||||
<div className="flex items-center gap-1 mt-1.5 text-xs text-slate-500 font-mono">
|
||||
<Clock size={12} />
|
||||
{formatSentAt(e.sent_at)}
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</li>
|
||||
)
|
||||
})}
|
||||
</ul>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
|
@ -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<string, typeof Bell> = {
|
||||
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 (
|
||||
<div className={`p-4 ${styles.bg} border-l-4 ${styles.border}`}>
|
||||
<div className="flex items-start gap-3">
|
||||
<Icon size={20} className={styles.iconColor} />
|
||||
<div className="flex-1 min-w-0">
|
||||
<div className="flex items-center gap-2 mb-1">
|
||||
<span className={`text-xs px-2 py-0.5 rounded-full ${styles.badge}`}>
|
||||
{alert.severity?.toUpperCase()}
|
||||
</span>
|
||||
<span className="text-xs text-slate-500">{alert.type}</span>
|
||||
</div>
|
||||
<div className="text-sm text-slate-200">{alert.message}</div>
|
||||
<div className="flex items-center gap-4 mt-2 text-xs text-slate-500">
|
||||
<span className="flex items-center gap-1">
|
||||
<Clock size={12} />
|
||||
{alert.timestamp ? formatTimeAgo(alert.timestamp) : 'Just now'}
|
||||
</span>
|
||||
{alert.scope_value && (
|
||||
<span>{alert.scope_type}: {alert.scope_value}</span>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
<button
|
||||
onClick={() => onAcknowledge(alert)}
|
||||
className="px-3 py-1 text-xs text-slate-400 hover:text-slate-200 border border-border rounded hover:bg-bg-hover transition-colors"
|
||||
>
|
||||
Acknowledge
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
// 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 (
|
||||
<div className="bg-bg-card border border-border">
|
||||
{/* Filters */}
|
||||
<div className="p-4 border-b border-border flex items-center gap-4">
|
||||
<div className="flex items-center gap-2">
|
||||
<Filter size={14} className="text-slate-400" />
|
||||
<span className="text-sm text-slate-400">Filter:</span>
|
||||
</div>
|
||||
<select
|
||||
value={typeFilter}
|
||||
onChange={(e) => onTypeFilterChange(e.target.value)}
|
||||
className="bg-bg border border-border rounded px-3 py-1.5 text-sm text-slate-200 focus:outline-none focus:border-[#f59e0b]"
|
||||
>
|
||||
{alertTypes.map((t) => (
|
||||
<option key={t} value={t}>
|
||||
{t === 'all' ? 'All Types' : t.replace(/_/g, ' ')}
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
<select
|
||||
value={severityFilter}
|
||||
onChange={(e) => onSeverityFilterChange(e.target.value)}
|
||||
className="bg-bg border border-border rounded px-3 py-1.5 text-sm text-slate-200 focus:outline-none focus:border-[#f59e0b]"
|
||||
>
|
||||
{severities.map((s) => (
|
||||
<option key={s} value={s}>
|
||||
{s === 'all' ? 'All Severities' : s.charAt(0).toUpperCase() + s.slice(1)}
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
</div>
|
||||
|
||||
{/* Table */}
|
||||
<div className="overflow-x-auto">
|
||||
<table className="w-full">
|
||||
<thead>
|
||||
<tr className="border-b border-border">
|
||||
<th className="text-left text-xs font-medium text-slate-400 p-4">Time</th>
|
||||
<th className="text-left text-xs font-medium text-slate-400 p-4">Type</th>
|
||||
<th className="text-left text-xs font-medium text-slate-400 p-4">Severity</th>
|
||||
<th className="text-left text-xs font-medium text-slate-400 p-4">Message</th>
|
||||
<th className="text-left text-xs font-medium text-slate-400 p-4">Duration</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{history.length > 0 ? (
|
||||
history.map((item, i) => {
|
||||
const styles = getSeverityStyles(item.severity)
|
||||
return (
|
||||
<tr key={item.id || i} className="border-b border-border hover:bg-bg-hover">
|
||||
<td className="p-4 text-sm text-slate-400 font-mono whitespace-nowrap">
|
||||
{formatDateTime(item.timestamp)}
|
||||
</td>
|
||||
<td className="p-4 text-sm text-slate-300">
|
||||
{item.type.replace(/_/g, ' ')}
|
||||
</td>
|
||||
<td className="p-4">
|
||||
<span className={`text-xs px-2 py-0.5 rounded-full ${styles.badge}`}>
|
||||
{item.severity}
|
||||
</span>
|
||||
</td>
|
||||
<td className="p-4 text-sm text-slate-200 max-w-md truncate">
|
||||
{item.message}
|
||||
</td>
|
||||
<td className="p-4 text-sm text-slate-400 font-mono">
|
||||
{item.duration ? formatDuration(item.duration) : '-'}
|
||||
</td>
|
||||
</tr>
|
||||
)
|
||||
})
|
||||
) : (
|
||||
<tr>
|
||||
<td colSpan={5} className="p-8 text-center text-slate-500">
|
||||
No alert history available
|
||||
</td>
|
||||
</tr>
|
||||
)}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
{/* Pagination */}
|
||||
{totalPages > 1 && (
|
||||
<div className="p-4 border-t border-border flex items-center justify-between">
|
||||
<span className="text-sm text-slate-400">
|
||||
Page {page} of {totalPages}
|
||||
</span>
|
||||
<div className="flex items-center gap-2">
|
||||
<button
|
||||
onClick={() => onPageChange(page - 1)}
|
||||
disabled={page <= 1}
|
||||
className="p-2 text-slate-400 hover:text-slate-200 disabled:opacity-50 disabled:cursor-not-allowed"
|
||||
>
|
||||
<ChevronLeft size={16} />
|
||||
</button>
|
||||
<button
|
||||
onClick={() => onPageChange(page + 1)}
|
||||
disabled={page >= totalPages}
|
||||
className="p-2 text-slate-400 hover:text-slate-200 disabled:opacity-50 disabled:cursor-not-allowed"
|
||||
>
|
||||
<ChevronRight size={16} />
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
// 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 (
|
||||
<div className="p-4 bg-bg-hover border border-border">
|
||||
<div className="flex items-center gap-3">
|
||||
<div className="w-10 h-10 bg-[#f59e0b]/10 flex items-center justify-center">
|
||||
<Icon size={18} className="text-[#f59e0b]" />
|
||||
</div>
|
||||
<div className="flex-1">
|
||||
<div className="text-sm text-slate-200 font-medium">
|
||||
{subscription.sub_type.charAt(0).toUpperCase() + subscription.sub_type.slice(1)}
|
||||
{subscription.scope_type !== 'mesh' && subscription.scope_value && (
|
||||
<span className="text-slate-400 font-normal ml-2">
|
||||
({subscription.scope_type}: {subscription.scope_value})
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
<div className="text-xs text-slate-500 mt-0.5">
|
||||
{formatSchedule()} • {resolveNodeName(subscription.user_id)}
|
||||
</div>
|
||||
</div>
|
||||
<div className={`w-2 h-2 rounded-full ${subscription.enabled ? 'bg-green-500' : 'bg-slate-500'}`} />
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
export default function Alerts() {
|
||||
const [activeAlerts, setActiveAlerts] = useState<Alert[]>([])
|
||||
const [history, setHistory] = useState<AlertHistoryItem[]>([])
|
||||
const [subscriptions, setSubscriptions] = useState<Subscription[]>([])
|
||||
const [nodes, setNodes] = useState<Node[]>([])
|
||||
const [loading, setLoading] = useState(true)
|
||||
const [error, setError] = useState<string | null>(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<Set<string>>(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 (
|
||||
<div className="flex items-center justify-center h-64">
|
||||
<div className="text-slate-400">Loading alerts...</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
if (error) {
|
||||
return (
|
||||
<div className="flex items-center justify-center h-64">
|
||||
<div className="text-red-400">Error: {error}</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="space-y-6">
|
||||
{/* Active Alerts */}
|
||||
<div className="bg-bg-card border border-border p-6">
|
||||
<h2 className="text-sm font-medium text-slate-400 mb-4 flex items-center gap-2">
|
||||
<AlertTriangle size={14} />
|
||||
Active Alerts ({visibleAlerts.length})
|
||||
</h2>
|
||||
{visibleAlerts.length > 0 ? (
|
||||
<div className="space-y-3">
|
||||
{visibleAlerts.map((alert, i) => (
|
||||
<ActiveAlertCard
|
||||
key={`${alert.type}-${alert.timestamp}-${i}`}
|
||||
alert={alert}
|
||||
onAcknowledge={handleAcknowledge}
|
||||
/>
|
||||
))}
|
||||
</div>
|
||||
) : (
|
||||
<div className="flex items-center gap-2 text-slate-500 py-8">
|
||||
<CheckCircle size={20} className="text-green-500" />
|
||||
<span>No active alerts — all systems nominal</span>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
|
||||
{/* Alert History */}
|
||||
<div>
|
||||
<h2 className="text-sm font-medium text-slate-400 mb-4 flex items-center gap-2">
|
||||
<Clock size={14} />
|
||||
Alert History
|
||||
</h2>
|
||||
<AlertHistoryTable
|
||||
history={history}
|
||||
typeFilter={typeFilter}
|
||||
severityFilter={severityFilter}
|
||||
onTypeFilterChange={(v) => {
|
||||
setTypeFilter(v)
|
||||
setPage(1)
|
||||
}}
|
||||
onSeverityFilterChange={(v) => {
|
||||
setSeverityFilter(v)
|
||||
setPage(1)
|
||||
}}
|
||||
page={page}
|
||||
totalPages={totalPages}
|
||||
onPageChange={setPage}
|
||||
/>
|
||||
</div>
|
||||
|
||||
{/* Subscriptions */}
|
||||
<div className="bg-bg-card border border-border p-6">
|
||||
<h2 className="text-sm font-medium text-slate-400 mb-4 flex items-center gap-2">
|
||||
<Users size={14} />
|
||||
Mesh Subscriptions ({subscriptions.length})
|
||||
</h2>
|
||||
{subscriptions.length > 0 ? (
|
||||
<div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-3">
|
||||
{subscriptions.map((sub) => (
|
||||
<SubscriptionCard key={sub.id} subscription={sub} nodes={nodes} />
|
||||
))}
|
||||
</div>
|
||||
) : (
|
||||
<div className="text-slate-500 py-4">
|
||||
<p>No active subscriptions.</p>
|
||||
<p className="text-xs mt-2">
|
||||
Manage subscriptions via <code className="text-[#f59e0b]">!subscribe</code> on mesh. Broadcasts arrive with one of three prefixes — <strong>New:</strong> (first sight), <strong>Update:</strong> (material change), or <strong>Active:</strong> (clock-driven reminder while the event is still live). See <a href="/reference#broadcast-types" className="text-[#f59e0b] hover:underline">Broadcast Types</a> and <a href="/reference#reminders" className="text-[#f59e0b] hover:underline">Reminder System</a> in Reference.
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
|
@ -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)' },
|
||||
|
|
|
|||
|
|
@ -1186,18 +1186,6 @@ export default function Reference() {
|
|||
]}
|
||||
/>
|
||||
|
||||
<SectionHeader>Subscription Commands</SectionHeader>
|
||||
<RefTable
|
||||
headers={['Command', 'What It Does']}
|
||||
rows={[
|
||||
[<Mono>!subscribe</Mono>, 'Lists all alert categories you can subscribe to'],
|
||||
[<Mono>!subscribe fire_proximity</Mono>, 'Subscribe to a specific category'],
|
||||
[<Mono>!subscribe all</Mono>, 'Subscribe to everything'],
|
||||
[<Mono>!unsubscribe fire_proximity</Mono>, 'Unsubscribe from a category'],
|
||||
[<Mono>!subscriptions</Mono>, "Shows what you're currently subscribed to"],
|
||||
]}
|
||||
/>
|
||||
|
||||
<SectionHeader>Conversational</SectionHeader>
|
||||
<p>
|
||||
Bang commands are the short, predictable interface. For anything that
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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|<category> [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 <category> - 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 <category> - 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|<category>|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|<category>|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, <category>, 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
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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 = []
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
||||
|
|
|
|||
9
work/meshai/persistence/migrations/v20.sql
Normal file
9
work/meshai/persistence/migrations/v20.sql
Normal file
|
|
@ -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;
|
||||
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue