fix: use native EventSource for SSE instead of fetch with manual parsing

apps/pay/src/monitor.ts

@@ -56,62 +56,33 @@ async function refreshMaps() {
 }
 
 // ── SSE streams per chain ───────────────────────────────────────────
-const activeStreams = new Map<string, AbortController>();
+const activeStreams = new Map<string, EventSource>();
 
 /** Start a raw SSE stream for a chain — receives ALL txs and blocks. */
 function startChainStream(chain: string) {
   if (activeStreams.has(chain)) return;
 
-  const ac = new AbortController();
-  activeStreams.set(chain, ac);
-
   const query = { crypto: chain };
   const q = Buffer.from(JSON.stringify(query)).toString("base64");
   const url = `${SOCK_API}/sse?q=${q}`;
 
-  connectSSE(chain, url, ac.signal);
+  const es = new EventSource(url);
-}
+  activeStreams.set(chain, es);
 
-async function connectSSE(chain: string, url: string, signal: AbortSignal) {
+  es.onmessage = (e) => {
-  while (!signal.aborted) {
     try {
-      const res = await fetch(url, { signal });
+      const event = JSON.parse(e.data);
-      if (!res.ok || !res.body) {
+      if (event.type === "block") {
-        console.error(`SSE ${chain}: HTTP ${res.status}`);
+        handleBlockEvent(chain, event).catch(() => {});
-        await sleep(5000);
+      } else {
-        continue;
+        handleTxEvent(chain, event).catch(() => {});
       }
+    } catch {}
+  };
 
-      const reader = res.body.getReader();
+  es.onerror = () => {
-      const decoder = new TextDecoder();
+    console.error(`SSE ${chain}: connection error (will auto-reconnect)`);
-      let buffer = "";
+  };
-
-      while (!signal.aborted) {
-        const { done, value } = await reader.read();
-        if (done) break;
-
-        buffer += decoder.decode(value, { stream: true });
-        const lines = buffer.split("\n");
-        buffer = lines.pop() ?? "";
-
-        for (const line of lines) {
-          if (!line.startsWith("data: ")) continue;
-          try {
-            const event = JSON.parse(line.slice(6));
-            if (event.type === "block") {
-              await handleBlockEvent(chain, event);
-            } else {
-              await handleTxEvent(chain, event);
-            }
-          } catch {}
-        }
-      }
-    } catch (e: any) {
-      if (signal.aborted) return;
-      console.error(`SSE ${chain} error:`, e.message);
-    }
-    if (!signal.aborted) await sleep(3000);
-  }
 }
 
 /** Handle a tx event — compare all outputs against our active addresses. */
@@ -237,9 +208,9 @@ async function syncStreams() {
     if (COINS[chain]) startChainStream(chain);
   }
 
-  for (const [chain, ac] of activeStreams) {
+  for (const [chain, es] of activeStreams) {
     if (!chainsNeeded.has(chain)) {
-      ac.abort();
+      es.close();
       activeStreams.delete(chain);
     }
   }
@@ -365,8 +336,4 @@ export async function expireProPlans() {
   if (result.count > 0) {
     console.log(`Downgraded ${result.count} expired pro accounts to free`);
   }
-}
+}
-
-function sleep(ms: number) {
-  return new Promise(resolve => setTimeout(resolve, ms));
-}

apps/web/src/views/checkout.ejs

@@ -136,7 +136,6 @@
   let paymentId = null;
   let pollInterval = null;
   let countdownInterval = null;
-  let sseAbort = null;
 
   // Fetch available coins on load
   (async () => {
@@ -248,55 +247,36 @@
   let watchedAddress = null;
   let watchedTxids = [];
 
-  /** Listen to raw SSE for this coin, match tx outputs against our address locally.
+  /** Listen to raw SSE via EventSource for this coin, match tx outputs locally.
    *  Tracks multiple txids (user may send across several transactions).
    *  On block, checks if any of our txids got confirmed. */
+  let eventSource = null;
+
   function watchAddress(coin, address) {
-    if (sseAbort) sseAbort.abort();
+    if (eventSource) { eventSource.close(); eventSource = null; }
-    sseAbort = new AbortController();
     watchedAddress = address;
     watchedTxids = [];
 
-    // Raw stream for this coin — all txs and blocks, no query filter
     const query = { crypto: coin };
     const q = btoa(JSON.stringify(query));
     const url = `${SOCK_API}/sse?q=${q}`;
 
-    (async function connect() {
+    eventSource = new EventSource(url);
-      while (!sseAbort.signal.aborted) {
-        try {
-          const res = await fetch(url, { signal: sseAbort.signal });
-          if (!res.ok || !res.body) { await new Promise(r => setTimeout(r, 3000)); continue; }
 
-          const reader = res.body.getReader();
+    eventSource.onmessage = (e) => {
-          const decoder = new TextDecoder();
+      try {
-          let buffer = '';
+        const event = JSON.parse(e.data);
-
+        if (event.type === 'block') {
-          while (!sseAbort.signal.aborted) {
+          onBlock(event);
-            const { done, value } = await reader.read();
+        } else {
-            if (done) break;
+          onTx(event);
-            buffer += decoder.decode(value, { stream: true });
-            const lines = buffer.split('\n');
-            buffer = lines.pop() ?? '';
-
-            for (const line of lines) {
-              if (!line.startsWith('data: ')) continue;
-              try {
-                const event = JSON.parse(line.slice(6));
-                if (event.type === 'block') {
-                  onBlock(event);
-                } else {
-                  onTx(event);
-                }
-              } catch {}
-            }
-          }
-        } catch (e) {
-          if (sseAbort.signal.aborted) return;
         }
-        if (!sseAbort.signal.aborted) await new Promise(r => setTimeout(r, 3000));
+      } catch {}
-      }
+    };
-    })();
+
+    eventSource.onerror = () => {
+      // EventSource auto-reconnects
+    };
   }
 
   /** Check if any tx output matches our payment address. */
@@ -352,14 +332,14 @@
       } else if (data.status === 'paid') {
         clearInterval(pollInterval);
         clearInterval(countdownInterval);
-        if (sseAbort) { sseAbort.abort(); sseAbort = null; }
+        if (eventSource) { eventSource.close(); eventSource = null; }
         document.getElementById('pay-status-section').classList.add('hidden');
         document.getElementById('pay-success').classList.remove('hidden');
         setTimeout(() => { window.location.href = '/dashboard/settings'; }, 3000);
       } else if (data.status === 'expired') {
         clearInterval(pollInterval);
         clearInterval(countdownInterval);
-        if (sseAbort) { sseAbort.abort(); sseAbort = null; }
+        if (eventSource) { eventSource.close(); eventSource = null; }
         document.getElementById('pay-status-section').classList.add('hidden');
         document.getElementById('pay-expired').classList.remove('hidden');
       }
@@ -396,7 +376,7 @@
   }
 
   function resetCheckout() {
-    if (sseAbort) { sseAbort.abort(); sseAbort = null; }
+    if (eventSource) { eventSource.close(); eventSource = null; }
     if (pollInterval) { clearInterval(pollInterval); pollInterval = null; }
     if (countdownInterval) { clearInterval(countdownInterval); countdownInterval = null; }
     watchedAddress = null;