feat: instant payment detection via Freedom.st SSE on backend and frontend

apps/pay/src/monitor.ts

@@ -1,9 +1,147 @@
-/// Background job: poll pending payments and activate plans on confirmation.
+/// Payment monitor: uses Freedom.st SSE for instant tx detection,
+/// with periodic polling as fallback for confirmations and expiry.
 import sql from "./db";
 import { getAddressInfo } from "./freedom";
 import { COINS } from "./plans";
 
-/** Check all pending/confirming payments against the blockchain. */
+const SOCK_API = process.env.FREEDOM_SOCK ?? "https://sock-v1.freedom.st";
+
+// ── In-memory address lookup for SSE matching ─────────────────────────
+// Maps address → payment row for all pending/confirming payments.
+let addressMap = new Map<string, any>();
+
+/** Refresh the address map from DB. Called periodically. */
+async function refreshAddressMap() {
+  const pending = await sql`
+    SELECT * FROM payments
+    WHERE status IN ('pending', 'confirming')
+    AND expires_at >= now()
+  `;
+  const map = new Map<string, any>();
+  for (const p of pending) map.set(p.address, p);
+  addressMap = map;
+}
+
+// ── SSE listeners per chain ───────────────────────────────────────────
+const activeStreams = new Map<string, AbortController>();
+
+/** Start an SSE stream for a chain, watching for txs to any of our addresses. */
+function startChainStream(chain: string) {
+  if (activeStreams.has(chain)) return;
+
+  const ac = new AbortController();
+  activeStreams.set(chain, ac);
+
+  // No address filter — we watch ALL txs on this chain and match server-side
+  // against our address map. This is necessary because addresses change as
+  // new payments are created.
+  const query = { chain };
+  const q = Buffer.from(JSON.stringify(query)).toString("base64");
+  const url = `${SOCK_API}/sse?q=${q}`;
+
+  connectSSE(chain, url, ac.signal);
+}
+
+async function connectSSE(chain: string, url: string, signal: AbortSignal) {
+  while (!signal.aborted) {
+    try {
+      const res = await fetch(url, { signal });
+      if (!res.ok || !res.body) {
+        console.error(`SSE ${chain}: HTTP ${res.status}`);
+        await sleep(5000);
+        continue;
+      }
+
+      const reader = res.body.getReader();
+      const decoder = new TextDecoder();
+      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));
+            await handleTxEvent(chain, event);
+          } catch {}
+        }
+      }
+    } catch (e: any) {
+      if (signal.aborted) return;
+      console.error(`SSE ${chain} error:`, e.message);
+    }
+    // Reconnect after brief pause
+    if (!signal.aborted) await sleep(3000);
+  }
+}
+
+/** Handle a transaction event from SSE. Check if any output matches a pending payment address. */
+async function handleTxEvent(chain: string, event: any) {
+  const tx = event.data || event;
+  const outputs = tx.out || tx.outputs || tx.vout || [];
+  const txid = tx.txid || tx.hash || null;
+
+  for (const out of outputs) {
+    // Freedom.st tx format: out[].script.address
+    const addr = out?.script?.address || out?.address || out?.scriptPubKey?.address;
+    if (!addr) continue;
+
+    const payment = addressMap.get(addr);
+    if (!payment) continue;
+    if (payment.coin !== chain) continue;
+
+    console.log(`SSE: tx ${txid} detected for payment ${payment.id} (${chain})`);
+
+    const coin = COINS[chain];
+    if (!coin) continue;
+
+    if (coin.confirmations === 0) {
+      // 0-conf: activate immediately
+      await activatePayment(payment, txid);
+      addressMap.delete(addr);
+    } else {
+      // 1+ conf: mark as confirming, polling will handle confirmation
+      if (payment.status === "pending") {
+        await sql`UPDATE payments SET status = 'confirming', txid = ${txid} WHERE id = ${payment.id}`;
+        payment.status = "confirming";
+        payment.txid = txid;
+        console.log(`Payment ${payment.id} now confirming (waiting for block)`);
+      }
+    }
+  }
+}
+
+/** Start SSE streams for all chains that have pending payments. */
+async function syncStreams() {
+  // Determine which chains have pending payments
+  const chainsNeeded = new Set<string>();
+  for (const p of addressMap.values()) {
+    chainsNeeded.add(p.coin);
+  }
+
+  // Start streams for chains we need
+  for (const chain of chainsNeeded) {
+    if (COINS[chain]) startChainStream(chain);
+  }
+
+  // Stop streams for chains with no pending payments
+  for (const [chain, ac] of activeStreams) {
+    if (!chainsNeeded.has(chain)) {
+      ac.abort();
+      activeStreams.delete(chain);
+    }
+  }
+}
+
+// ── Polling fallback (for confirmations and missed txs) ───────────────
+
+/** Poll all pending/confirming payments. Handles confirmations and expiry. */
 export async function checkPayments() {
   // Expire stale payments
   await sql`
@@ -12,22 +150,57 @@ export async function checkPayments() {
     AND expires_at < now()
   `;
 
+  // Refresh address map and sync SSE streams
+  await refreshAddressMap();
+  await syncStreams();
+
+  // Poll confirming payments for block confirmation
+  const confirming = await sql`
+    SELECT * FROM payments
+    WHERE status = 'confirming'
+    AND expires_at >= now()
+  `;
+
+  for (const payment of confirming) {
+    try {
+      await checkConfirmation(payment);
+    } catch (e) {
+      console.error(`Error checking confirmation for payment ${payment.id}:`, e);
+    }
+  }
+
+  // Also poll any pending payments as fallback (SSE might have missed them)
   const pending = await sql`
     SELECT * FROM payments
-    WHERE status IN ('pending', 'confirming')
+    WHERE status = 'pending'
     AND expires_at >= now()
   `;
 
   for (const payment of pending) {
     try {
-      await checkPayment(payment);
+      await checkPending(payment);
     } catch (e) {
       console.error(`Error checking payment ${payment.id}:`, e);
     }
   }
 }
 
-async function checkPayment(payment: any) {
+async function checkConfirmation(payment: any) {
+  const info = await getAddressInfo(payment.address);
+  if (info.error) return;
+
+  const expectedSats = cryptoToSats(payment.coin, payment.amount_crypto);
+  const threshold = expectedSats * 0.995;
+  const confirmed = sumReceived(info, true);
+
+  if (confirmed >= threshold) {
+    const txid = payment.txid || findTxid(info);
+    await activatePayment(payment, txid);
+    addressMap.delete(payment.address);
+  }
+}
+
+async function checkPending(payment: any) {
   const info = await getAddressInfo(payment.address);
   if (info.error) return;
 
@@ -35,65 +208,52 @@ async function checkPayment(payment: any) {
   if (!coin) return;
 
   const expectedSats = cryptoToSats(payment.coin, payment.amount_crypto);
-
-  // Sum confirmed received
-  const confirmedReceived = sumReceived(info, true);
-  // Sum all received (including unconfirmed)
-  const totalReceived = sumReceived(info, false);
-
-  // Allow 0.5% tolerance for rounding
   const threshold = expectedSats * 0.995;
 
-  if (coin.confirmations === 0) {
+  const confirmed = sumReceived(info, true);
-    // 0-conf coins (BCH, XEC): accept as soon as seen in mempool
+  const total = sumReceived(info, false);
-    if (totalReceived >= threshold) {
+
-      const txid = findTxid(info);
+  if (coin.confirmations === 0 && total >= threshold) {
-      await activatePayment(payment, txid);
+    await activatePayment(payment, findTxid(info));
-    }
+    addressMap.delete(payment.address);
-  } else {
+  } else if (coin.confirmations > 0 && confirmed >= threshold) {
-    // 1-conf coins: need confirmed balance
+    await activatePayment(payment, findTxid(info));
-    if (confirmedReceived >= threshold) {
+    addressMap.delete(payment.address);
-      const txid = findTxid(info);
+  } else if (total >= threshold && payment.status === "pending") {
-      await activatePayment(payment, txid);
+    await sql`UPDATE payments SET status = 'confirming' WHERE id = ${payment.id}`;
-    } else if (totalReceived >= threshold && payment.status === "pending") {
-      // Seen in mempool but not yet confirmed
-      await sql`UPDATE payments SET status = 'confirming' WHERE id = ${payment.id}`;
-    }
   }
 }
 
+// ── Helpers ───────────────────────────────────────────────────────────
+
 function sumReceived(info: any, confirmedOnly: boolean): number {
-  // Freedom.st /address response has balance fields
-  // The response includes received/balance in the coin's base unit (satoshis)
   if (confirmedOnly) {
     return Number(info.balance?.confirmed ?? info.confirmed ?? info.received ?? 0);
   }
-  // Total = confirmed + unconfirmed
   const confirmed = Number(info.balance?.confirmed ?? info.confirmed ?? info.received ?? 0);
   const unconfirmed = Number(info.balance?.unconfirmed ?? info.unconfirmed ?? 0);
   return confirmed + unconfirmed;
 }
 
 function findTxid(info: any): string | null {
-  // Try to get the first txid from transaction history
   if (info.txs?.length) return info.txs[0].txid || info.txs[0].hash || null;
   if (info.transactions?.length) return info.transactions[0].txid || info.transactions[0] || null;
   return null;
 }
 
-/** Convert a decimal crypto amount string to satoshis/base units. */
 function cryptoToSats(coin: string, amount: string): number {
-  // XEC uses 100 sats per coin, everything else uses 1e8
   const multiplier = coin === "xec" ? 100 : 1e8;
   return Math.round(parseFloat(amount) * multiplier);
 }
 
 async function activatePayment(payment: any, txid: string | null) {
-  await sql`
+  const [updated] = await sql`
     UPDATE payments
     SET status = 'paid', paid_at = now(), txid = ${txid}
     WHERE id = ${payment.id} AND status != 'paid'
+    RETURNING id
   `;
+  if (!updated) return; // Already activated by another path
 
   if (payment.plan === "lifetime") {
     await sql`
@@ -101,7 +261,6 @@ async function activatePayment(payment: any, txid: string | null) {
       WHERE id = ${payment.account_id}
     `;
   } else {
-    // Pro: extend from current expiry or now
     const [account] = await sql`
       SELECT plan, plan_expires_at FROM accounts WHERE id = ${payment.account_id}
     `;
@@ -132,3 +291,7 @@ export async function expireProPlans() {
     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

@@ -127,6 +127,7 @@
 
 <script>
   const PAY_API = '<%= it.payApi || "" %>';
+  const SOCK_API = 'https://sock-v1.freedom.st';
 
   let selectedPlan = null;
   let selectedCoin = null;
@@ -135,6 +136,7 @@
   let paymentId = null;
   let pollInterval = null;
   let countdownInterval = null;
+  let sseAbort = null;
 
   // Fetch available coins on load
   (async () => {
@@ -234,8 +236,65 @@
     updateCountdown(expiresAt);
     countdownInterval = setInterval(() => updateCountdown(expiresAt), 1000);
 
-    // Start polling
+    // Start SSE stream for instant tx detection on the client
-    pollInterval = setInterval(() => pollStatus(), 5000);
+    watchAddress(data.coin, data.address);
+
+    // Poll as fallback (for confirmations and in case SSE drops)
+    pollInterval = setInterval(() => pollStatus(), 10000);
+  }
+
+  /** Subscribe to Freedom.st SSE filtered by our payment address for instant detection. */
+  function watchAddress(coin, address) {
+    if (sseAbort) sseAbort.abort();
+    sseAbort = new AbortController();
+
+    const query = { chain: coin, "data.out.script.address": address };
+    const q = btoa(JSON.stringify(query));
+    const url = `${SOCK_API}/sse?q=${q}`;
+
+    (async function connect() {
+      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();
+          const decoder = new TextDecoder();
+          let buffer = '';
+
+          while (!sseAbort.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));
+                onTxDetected(event);
+              } catch {}
+            }
+          }
+        } catch (e) {
+          if (sseAbort.signal.aborted) return;
+        }
+        if (!sseAbort.signal.aborted) await new Promise(r => setTimeout(r, 3000));
+      }
+    })();
+  }
+
+  function onTxDetected(event) {
+    // Transaction matched our address filter — payment detected instantly
+    console.log('TX detected via SSE:', event);
+    const statusEl = document.getElementById('pay-status');
+    statusEl.innerHTML = `
+      <span class="w-2 h-2 rounded-full bg-blue-500 animate-pulse"></span>
+      <span class="text-blue-400">Transaction detected, waiting for confirmation...</span>
+    `;
+    // Immediately poll the pay API to get authoritative status update
+    pollStatus();
   }
 
   function updateCountdown(expiresAt) {
@@ -261,12 +320,14 @@
       } else if (data.status === 'paid') {
         clearInterval(pollInterval);
         clearInterval(countdownInterval);
+        if (sseAbort) { sseAbort.abort(); sseAbort = 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; }
         document.getElementById('pay-status-section').classList.add('hidden');
         document.getElementById('pay-expired').classList.remove('hidden');
       }
@@ -282,6 +343,9 @@
   }
 
   function resetCheckout() {
+    if (sseAbort) { sseAbort.abort(); sseAbort = null; }
+    if (pollInterval) { clearInterval(pollInterval); pollInterval = null; }
+    if (countdownInterval) { clearInterval(countdownInterval); countdownInterval = null; }
     document.getElementById('step-select').classList.remove('hidden');
     document.getElementById('step-pay').classList.add('hidden');
     document.getElementById('pay-success').classList.add('hidden');