Refactor SSE event handling and consolidate notification contexts

Author: chriswritescode-dev

backend/package.json

@@ -19,6 +19,7 @@
     "@opencode-manager/shared": "workspace:*",
     "archiver": "^7.0.1",
     "dotenv": "^17.2.3",
+    "eventsource": "^4.1.0",
     "hono": "^4.10.1",
     "strip-json-comments": "^3.1.1",
     "zod": "^4.1.12"
@@ -28,6 +29,7 @@
     "@types/archiver": "^7.0.0",
     "@types/better-sqlite3": "^7.6.13",
     "@types/bun": "latest",
+    "@types/eventsource": "^3.0.0",
     "@vitest/ui": "^3.2.4",
     "eslint": "^9.39.1",
     "typescript-eslint": "^8.45.0",

backend/src/index.ts

@@ -13,6 +13,8 @@ import { createFileRoutes } from './routes/files'
 import { createProvidersRoutes } from './routes/providers'
 import { createOAuthRoutes } from './routes/oauth'
 import { createTitleRoutes } from './routes/title'
+import { createSSERoutes } from './routes/sse'
+import { sseAggregator } from './services/sse-aggregator'
 import { ensureDirectoryExists, writeFileContent, fileExists, readFileContent } from './services/file-operations'
 import { SettingsService } from './services/settings'
 import { opencodeServerManager } from './services/opencode-single-server'
@@ -230,6 +232,7 @@ app.route('/api/providers', createProvidersRoutes())
 app.route('/api/oauth', createOAuthRoutes())
 app.route('/api/tts', createTTSRoutes(db))
 app.route('/api/generate-title', createTitleRoutes())
+app.route('/api/sse', createSSERoutes())
 
 app.all('/api/opencode/*', async (c) => {
   const request = c.req.raw
@@ -302,10 +305,12 @@ const shutdown = async (signal: string) => {
 
   logger.info(`${signal} received, shutting down gracefully...`)
   try {
+    sseAggregator.shutdown()
+    logger.info('SSE Aggregator stopped')
     await opencodeServerManager.stop()
     logger.info('OpenCode server stopped')
   } catch (error) {
-    logger.error('Error stopping OpenCode server:', error)
+    logger.error('Error during shutdown:', error)
   }
   process.exit(0)
 }

backend/src/routes/sse.ts

@@ -0,0 +1,78 @@
+import { Hono } from 'hono'
+import { streamSSE } from 'hono/streaming'
+import { sseAggregator } from '../services/sse-aggregator'
+import { SSESubscribeSchema } from '@opencode-manager/shared/schemas'
+import { logger } from '../utils/logger'
+
+export function createSSERoutes() {
+  const app = new Hono()
+
+  app.get('/stream', async (c) => {
+    const directoriesParam = c.req.query('directories')
+    const directories = directoriesParam ? directoriesParam.split(',').filter(Boolean) : []
+
+    return streamSSE(c, async (stream) => {
+      const clientId = `client_${Date.now()}_${Math.random().toString(36).slice(2)}`
+
+      const cleanup = sseAggregator.addClient(
+        clientId,
+        (event, data) => {
+          stream.writeSSE({ event, data })
+        },
+        directories
+      )
+
+      stream.onAbort(() => {
+        cleanup()
+      })
+
+      try {
+        await stream.writeSSE({
+          event: 'connected',
+          data: JSON.stringify({ clientId, directories, ...sseAggregator.getConnectionStatus() })
+        })
+      } catch (err) {
+        logger.error(`Failed to send SSE connected event for ${clientId}:`, err)
+      }
+
+      await new Promise(() => {})
+    })
+  })
+
+  app.post('/subscribe', async (c) => {
+    const body = await c.req.json()
+    const result = SSESubscribeSchema.safeParse(body)
+    if (!result.success) {
+      return c.json({ success: false, error: 'Invalid request', details: result.error.issues }, 400)
+    }
+    const success = sseAggregator.addDirectories(result.data.clientId, result.data.directories)
+    if (!success) {
+      return c.json({ success: false, error: 'Client not found' }, 404)
+    }
+    return c.json({ success: true })
+  })
+
+  app.post('/unsubscribe', async (c) => {
+    const body = await c.req.json()
+    const result = SSESubscribeSchema.safeParse(body)
+    if (!result.success) {
+      return c.json({ success: false, error: 'Invalid request', details: result.error.issues }, 400)
+    }
+    const success = sseAggregator.removeDirectories(result.data.clientId, result.data.directories)
+    if (!success) {
+      return c.json({ success: false, error: 'Client not found' }, 404)
+    }
+    return c.json({ success: true })
+  })
+
+  app.get('/status', (c) => {
+    return c.json({
+      ...sseAggregator.getConnectionStatus(),
+      clients: sseAggregator.getClientCount(),
+      directories: sseAggregator.getActiveDirectories(),
+      activeSessions: sseAggregator.getActiveSessions()
+    })
+  })
+
+  return app
+}