Refactor event stream to use EventSource API

src/tools/events.ts

@@ -11,61 +11,37 @@ interface Listener {
 
 const _listeners = new Set<Listener>()
 
-let _abort: AbortController | undefined
-let _connected = false
+let _source: EventSource | undefined
 
 function ensureConnection() {
-  if (_connected) return
-  _connected = true
+  if (_source) return
   const url = `${process.env.TOES_URL}/api/events/stream`
-  _abort = new AbortController()
+  _source = new EventSource(url)
 
-  fetch(url, { signal: _abort.signal })
-    .then(async (res) => {
-      const reader = res.body!.getReader()
-      const decoder = new TextDecoder()
-      let buf = ''
+  _source.onerror = () => {
+    if (_source?.readyState === EventSource.CLOSED) {
+      console.warn('[toes] Event stream closed unexpectedly')
+      _source = undefined
+    }
+  }
 
-      while (true) {
-        const { done, value } = await reader.read()
-        if (done) break
-        buf += decoder.decode(value, { stream: true })
-
-        const lines = buf.split('\n')
-        buf = lines.pop()!
-        for (const line of lines) {
-          if (!line.startsWith('data: ')) continue
-          const payload = line.slice(6)
-          if (!payload) continue
-          try {
-            const event: ToesEvent = JSON.parse(payload)
-            _listeners.forEach(l => {
-              if (l.types.includes(event.type)) l.callback(event)
-            })
-          } catch (e) {
-            console.warn('[toes] Failed to parse event:', e)
-          }
-        }
-      }
-    })
-    .catch((e) => {
-      if (e.name === 'AbortError') return
-      console.warn('[toes] Event stream error, reconnecting...', e.message)
-    })
-    .finally(() => {
-      _connected = false
-      if (_listeners.size > 0) {
-        setTimeout(ensureConnection, 1000)
-      }
-    })
+  _source.onmessage = (e) => {
+    try {
+      const event: ToesEvent = JSON.parse(e.data)
+      _listeners.forEach(l => {
+        if (l.types.includes(event.type)) l.callback(event)
+      })
+    } catch {
+      // Ignore malformed events
+    }
+  }
 }
 
 function closeConnection() {
-  if (_abort) {
-    _abort.abort()
-    _abort = undefined
+  if (_source) {
+    _source.close()
+    _source = undefined
   }
-  _connected = false
 }
 
 export function on(type: ToesEventType | ToesEventType[], callback: EventCallback): () => void {