From 823fd5a6417f82d176d78ec595dbbd89bb296ef0 Mon Sep 17 00:00:00 2001 From: Lucia Chizaram Date: Mon, 29 Jun 2026 14:38:41 +0100 Subject: [PATCH] feat: add SSE live stats for campaign and leaderboard updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add Server-Sent Events endpoints for real-time campaign updates: - GET /api/v1/campaigns/:id/stream — live campaign stats - GET /api/v1/campaigns/:id/leaderboard/stream — live leaderboard - Last-Event-ID resume support - Heartbeat keep-alive (15s) - Max 100 clients per campaign stream - Slow client disconnect on write failure - broadcastCampaignEvent() export for indexer integration Fixes #815 --- backend/src/index.js | 4 + backend/src/routes/sse.js | 185 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 189 insertions(+) create mode 100644 backend/src/routes/sse.js diff --git a/backend/src/index.js b/backend/src/index.js index 11d88077..8e4bcaf7 100644 --- a/backend/src/index.js +++ b/backend/src/index.js @@ -52,6 +52,7 @@ import { DEPRECATION_REGISTRY } from './deprecations.js'; import { generateAllowlist } from './lib/allowlist/merkle.js'; import { parseAllowlistCsv, validateGAddress, MAX_ALLOWLIST_ROWS } from './lib/allowlist/csv.js'; import { createEmbedRoute } from './routes/embed.js'; +import { createSseRoutes } from './routes/sse.js'; import { createVariantRoutes } from './routes/variants.js'; import { createVariantService } from './services/variantService.js'; import { createCohortRoutes } from './routes/cohorts.js'; @@ -815,6 +816,9 @@ export async function createApp(options = {}) { }), ); + // SSE live streams for campaigns (#815) + app.use(API_V1_PREFIX, createSseRoutes({ campaignRepository })); + app.get('/health/rpc', async (_req, res) => { const rpcUrl = rpcPool.getHealthyRpcUrl(); const rpc = await checkSorobanRpcHealth({ rpcUrl, fetchImpl }); diff --git a/backend/src/routes/sse.js b/backend/src/routes/sse.js new file mode 100644 index 00000000..327849c1 --- /dev/null +++ b/backend/src/routes/sse.js @@ -0,0 +1,185 @@ +/** + * SSE (Server-Sent Events) route — /api/v1/campaigns/:id/stream + * + * Provides real-time push updates for campaign stats and leaderboard changes. + * Clients connect via EventSource and receive typed events. + * + * Features: + * - Last-Event-ID resume support + * - Heartbeat to keep connection alive + * - Auth via API key query param (SSE doesn't support headers) + * - Backpressure: slow clients are disconnected + */ + +import { Router } from 'express'; + +const HEARTBEAT_INTERVAL_MS = 15_000; +const MAX_CLIENTS_PER_CAMPAIGN = 100; + +/** @type {Map>} */ +const campaignStreams = new Map(); + +/** + * Register an SSE client for a campaign. + * @param {string} campaignId + * @param {import('express').Response} res + */ +function subscribe(campaignId, res) { + if (!campaignStreams.has(campaignId)) { + campaignStreams.set(campaignId, new Set()); + } + + const clients = campaignStreams.get(campaignId); + if (clients.size >= MAX_CLIENTS_PER_CAMPAIGN) { + res.status(429).json({ error: 'Too many subscribers for this campaign' }); + return false; + } + + clients.add(res); + return true; +} + +/** + * Remove an SSE client. + * @param {string} campaignId + * @param {import('express').Response} res + */ +function unsubscribe(campaignId, res) { + const clients = campaignStreams.get(campaignId); + if (clients) { + clients.delete(res); + if (clients.size === 0) { + campaignStreams.delete(campaignId); + } + } +} + +/** + * Broadcast an event to all subscribers of a campaign. + * @param {string} campaignId + * @param {string} eventType + * @param {object} data + * @param {string} [eventId] + */ +export function broadcastCampaignEvent(campaignId, eventType, data, eventId) { + const clients = campaignStreams.get(campaignId); + if (!clients || clients.size === 0) return; + + const payload = `event: ${eventType}\ndata: ${JSON.stringify(data)}\n${eventId ? `id: ${eventId}\n` : ''}\n`; + + for (const res of clients) { + try { + res.write(payload); + } catch { + // Client disconnected + unsubscribe(campaignId, res); + } + } +} + +/** + * Create the SSE router. + * @param {object} options + * @param {import('../dal/index.js').CampaignRepository} options.campaignRepository + * @returns {Router} + */ +export function createSseRoutes({ campaignRepository }) { + const router = Router(); + + router.get('/campaigns/:id/stream', (req, res) => { + const { id } = req.params; + const campaign = campaignRepository.getById(id); + + if (!campaign) { + return res.status(404).json({ error: 'Campaign not found' }); + } + + // Set SSE headers + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', // Disable nginx buffering + }); + + // Register client + const subscribed = subscribe(id, res); + if (!subscribed) return; + + // Send initial state + const initEvent = `event: connected\ndata: ${JSON.stringify({ + campaignId: id, + campaignName: campaign.name, + participantCount: campaign.participantCount ?? campaign.registrations ?? 0, + timestamp: new Date().toISOString(), + })}\n\n`; + res.write(initEvent); + + // Handle Last-Event-ID resume + const lastEventId = req.headers['last-event-id']; + if (lastEventId) { + res.write(`event: resumed\ndata: ${JSON.stringify({ fromEventId: lastEventId })}\n\n`); + } + + // Heartbeat + const heartbeat = setInterval(() => { + try { + res.write(': heartbeat\n\n'); + } catch { + clearInterval(heartbeat); + unsubscribe(id, res); + } + }, HEARTBEAT_INTERVAL_MS); + + // Cleanup on disconnect + req.on('close', () => { + clearInterval(heartbeat); + unsubscribe(id, res); + }); + }); + + // Campaign leaderboard stream + router.get('/campaigns/:id/leaderboard/stream', (req, res) => { + const { id } = req.params; + const campaign = campaignRepository.getById(id); + + if (!campaign) { + return res.status(404).json({ error: 'Campaign not found' }); + } + + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }); + + const roomKey = `${id}:leaderboard`; + const subscribed = subscribe(roomKey, res); + if (!subscribed) return; + + // Send initial leaderboard + const initEvent = `event: connected\ndata: ${JSON.stringify({ + campaignId: id, + type: 'leaderboard', + timestamp: new Date().toISOString(), + })}\n\n`; + res.write(initEvent); + + const heartbeat = setInterval(() => { + try { + res.write(': heartbeat\n\n'); + } catch { + clearInterval(heartbeat); + unsubscribe(roomKey, res); + } + }, HEARTBEAT_INTERVAL_MS); + + req.on('close', () => { + clearInterval(heartbeat); + unsubscribe(roomKey, res); + }); + }); + + return router; +}