Replace EventSource with fetch-based SSE with reconnect

src/tools/events.ts

@@ -11,23 +11,41 @@ interface Listener {
 
 const _listeners = new Set<Listener>()
 
-let _source: EventSource | undefined
+let _abort: AbortController | undefined
 
 function ensureConnection() {
-  if (_source) return
+  if (_abort) return
   const url = `${process.env.TOES_URL}/api/events/stream`
-  _source = new EventSource(url)
+  _abort = new AbortController()
+  connect(url, _abort.signal)
+}
 
-  _source.onerror = () => {
-    if (_source?.readyState === EventSource.CLOSED) {
-      console.warn('[toes] Event stream closed unexpectedly')
-      _source = undefined
+function closeConnection() {
+  if (_abort) {
+    _abort.abort()
+    _abort = undefined
   }
 }
 
-  _source.onmessage = (e) => {
+async function connect(url: string, signal: AbortSignal) {
+  while (!signal.aborted) {
     try {
-      const event: ToesEvent = JSON.parse(e.data)
+      const res = await fetch(url, { signal })
+      if (!res.ok || !res.body) throw new Error(`SSE ${res.status}`)
+      const reader = res.body.getReader()
+      const decoder = new TextDecoder()
+      let buf = ''
+      while (true) {
+        const { done, value } = await reader.read()
+        if (done) break
+        buf += decoder.decode(value, { stream: true })
+        const parts = buf.split('\n\n')
+        buf = parts.pop()!
+        for (const part of parts) {
+          const line = part.split('\n').find(l => l.startsWith('data:'))
+          if (!line) continue
+          try {
+            const event: ToesEvent = JSON.parse(line.slice(5).trim())
             _listeners.forEach(l => {
               if (l.types.includes(event.type)) l.callback(event)
             })
@@ -36,11 +54,11 @@ function ensureConnection() {
           }
         }
       }
-
-function closeConnection() {
-  if (_source) {
-    _source.close()
-    _source = undefined
+    } catch (err) {
+      if (signal.aborted) return
+      console.warn('[toes] Event stream error, reconnecting...')
+    }
+    if (!signal.aborted) await new Promise(r => setTimeout(r, 2000))
   }
 }