Merge pull request #49 from AnsoCode/feature/ws-notification-push

feat(notifications): replace polling with WebSocket push

CHANGELOG.md

@@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+- **Changed:** Notification delivery replaced polling with WebSocket push — `EditorLayout` no longer runs a `setInterval` to fetch `/api/notifications` every 5 seconds. A persistent `ws://host/ws/notifications` connection is opened on mount; the backend pushes each new alert in real-time as it is dispatched. Auto-reconnects with a 5-second back-off on close.
+- **Added:** `NotificationService.setBroadcaster()` — injectable callback wired to a `notificationSubscribers` Set in `index.ts`; every authenticated WS client subscribed to `/ws/notifications` receives a `{ type: 'notification', payload: NotificationHistory }` message the moment `dispatchAlert` fires.
+- **Changed:** `DatabaseService.addNotificationHistory` now returns the full inserted `NotificationHistory` record (including auto-generated `id` and `is_read: false`) instead of `void`, enabling the broadcaster to push the complete record to clients.
+- **Added:** `/ws/notifications` WebSocket upgrade path in `index.ts` — handled before the remote-node proxy path so it is always local. JWT auth verified manually (same pattern as all other WS paths per Directive 8).
 - **Security:** `GET /api/settings` no longer leaks `auth_username`, `auth_password_hash`, or `auth_jwt_secret` to the frontend — these keys are stripped from the response before sending.
 - **Security:** `POST /api/settings` now enforces a strict allowlist of writable keys — attempts to write auth credential keys (`auth_jwt_secret`, `auth_password_hash`, etc.) or arbitrary unknown keys are rejected with a 400 error.
 - **Added:** `PATCH /api/settings` bulk-update endpoint — accepts a partial settings object, validates all values via a Zod schema (type checking, range enforcement, URL format validation), and persists changes atomically in a single SQLite transaction. Replaces the N+1 per-key POST loop.

backend/src/index.ts

@@ -418,6 +418,17 @@ const wss = new WebSocket.Server({ noServer: true });
 
 let terminalWs: WebSocket | null = null;
 
+// Notification push — set of authenticated browser clients subscribed to real-time alerts
+const notificationSubscribers = new Set<WebSocket>();
+NotificationService.getInstance().setBroadcaster((notification) => {
+  const msg = JSON.stringify({ type: 'notification', payload: notification });
+  for (const ws of notificationSubscribers) {
+    if (ws.readyState === WebSocket.OPEN) {
+      ws.send(msg);
+    }
+  }
+});
+
 // Handle WebSocket upgrade with JWT authentication
 server.on('upgrade', async (req, socket, head) => {
   // Parse cookies from the upgrade request
@@ -450,6 +461,18 @@ server.on('upgrade', async (req, socket, head) => {
     const parsedUrl = new URL(url, `http://${req.headers.host || 'localhost'}`);
     const pathname = parsedUrl.pathname;
 
+    // Notification push channel — always local, never proxied to remote nodes
+    if (pathname === '/ws/notifications') {
+      const notifWss = new WebSocket.Server({ noServer: true });
+      notifWss.handleUpgrade(req, socket, head, (ws) => {
+        notifWss.close();
+        notificationSubscribers.add(ws);
+        ws.on('close', () => notificationSubscribers.delete(ws));
+        ws.on('error', () => { notificationSubscribers.delete(ws); ws.terminate(); });
+      });
+      return;
+    }
+
     // Resolve node context from query param
     const nodeIdParam = parsedUrl.searchParams.get('nodeId');
     const nodeId = nodeIdParam ? parseInt(nodeIdParam, 10) : NodeRegistry.getInstance().getDefaultNodeId();

backend/src/services/DatabaseService.ts

@@ -310,9 +310,9 @@ export class DatabaseService {
         }));
     }
 
-    public addNotificationHistory(notification: Omit<NotificationHistory, 'id' | 'is_read'>): void {
+    public addNotificationHistory(notification: Omit<NotificationHistory, 'id' | 'is_read'>): NotificationHistory {
         const stmt = this.db.prepare('INSERT INTO notification_history (level, message, timestamp, is_read) VALUES (?, ?, ?, 0)');
-        stmt.run(notification.level, notification.message, notification.timestamp);
+        const result = stmt.run(notification.level, notification.message, notification.timestamp);
 
         this.db.exec(`
       DELETE FROM notification_history
@@ -320,6 +320,14 @@ export class DatabaseService {
         SELECT id FROM notification_history ORDER BY timestamp DESC LIMIT 100
       )
     `);
+
+        return {
+            id: result.lastInsertRowid as number,
+            level: notification.level,
+            message: notification.message,
+            timestamp: notification.timestamp,
+            is_read: false,
+        };
     }
 
     public markAllNotificationsRead(): void {

backend/src/services/NotificationService.ts

@@ -1,8 +1,9 @@
-import { DatabaseService } from './DatabaseService';
+import { DatabaseService, NotificationHistory } from './DatabaseService';
 
 export class NotificationService {
     private static instance: NotificationService;
     private dbService: DatabaseService;
+    private broadcaster: ((notification: NotificationHistory) => void) | null = null;
 
     private constructor() {
         this.dbService = DatabaseService.getInstance();
@@ -15,14 +16,24 @@ export class NotificationService {
         return NotificationService.instance;
     }
 
+    /** Wire up the WebSocket push function after the WS server is initialised. */
+    public setBroadcaster(fn: (notification: NotificationHistory) => void): void {
+        this.broadcaster = fn;
+    }
+
     public async dispatchAlert(level: 'info' | 'warning' | 'error', message: string) {
-        // 1. Log to history
+        // 1. Log to history and get the full inserted record (with id)
-        this.dbService.addNotificationHistory({
+        const notification = this.dbService.addNotificationHistory({
             level,
             message,
             timestamp: Date.now()
         });
 
+        // 2. Push to connected browser clients via WebSocket
+        if (this.broadcaster) {
+            this.broadcaster(notification);
+        }
+
         // 2. Fetch enabled agents
         const agents = this.dbService.getEnabledAgents();
         if (agents.length === 0) {

frontend/src/components/EditorLayout.tsx

@@ -162,11 +162,48 @@ export default function EditorLayout() {
     }
   };
 
-  // Notification polling - independent of active node, runs once on mount
+  // Notification WS push — load history once on mount, then receive live updates
   useEffect(() => {
     fetchNotifications();
-    const notificationInterval = setInterval(fetchNotifications, 5000);
+
-    return () => clearInterval(notificationInterval);
+    const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
+    const wsBase = `${wsProtocol}//${window.location.host}`;
+    let ws: WebSocket | null = null;
+    let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+    let isMounted = true;
+
+    const connect = () => {
+      ws = new WebSocket(`${wsBase}/ws/notifications`);
+
+      ws.onmessage = (event) => {
+        try {
+          const msg = JSON.parse(event.data);
+          if (msg.type === 'notification' && msg.payload) {
+            setNotifications(prev => [msg.payload, ...prev]);
+          }
+        } catch (e) {
+          console.error('[WS notifications] parse error', e);
+        }
+      };
+
+      ws.onclose = () => {
+        if (isMounted) {
+          reconnectTimer = setTimeout(connect, 5000);
+        }
+      };
+
+      ws.onerror = () => {
+        ws?.close();
+      };
+    };
+
+    connect();
+
+    return () => {
+      isMounted = false;
+      if (reconnectTimer) clearTimeout(reconnectTimer);
+      ws?.close();
+    };
   }, []); // eslint-disable-line react-hooks/exhaustive-deps
 
   // Re-fetch stacks whenever the active node changes (or becomes available on mount).