import { Response } from 'express';
type Client = { id: number; res: Response };
const clients = new Map<string, Set<Client>>();
let nextId = 1;
export function subscribe(userId: string, res: Response): () => void {
const client: Client = { id: nextId++, res };
const set = clients.get(userId) ?? new Set<Client>();
set.add(client);
clients.set(userId, set);
return () => {
const current = clients.get(userId);
if (!current) return;
current.delete(client);
if (current.size === 0) clients.delete(userId);
};
}
function writeEvent(res: Response, event: string, data: unknown, id: number): void {
res.write(`id: ${id}\n`);
res.write(`event: ${event}\n`);
res.write(`data: ${JSON.stringify(data)}\n\n`);
}
export function publish(userId: string, event: string, data: unknown): void {
const set = clients.get(userId);
if (!set) return;
for (const client of set) {
writeEvent(client.res, event, data, client.id);
}
}
import { Router, Request, Response } from 'express';
import { subscribe, publish } from './sseHub';
const router = Router();
router.get('/events', (req: Request, res: Response) => {
const userId = String(req.query.userId ?? req.header('x-user-id') ?? '');
if (!userId) {
res.status(401).end();
return;
}
res.status(200).set({
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
'X-Accel-Buffering': 'no',
});
res.flushHeaders();
res.write('retry: 5000\n\n');
const heartbeat = setInterval(() => {
res.write(': keep-alive\n\n');
}, 15000);
const unsubscribe = subscribe(userId, res);
req.on('close', () => {
clearInterval(heartbeat);
unsubscribe();
res.end();
});
});
router.post('/emit', (req: Request, res: Response) => {
const { userId, message } = req.body as { userId: string; message: string };
publish(userId, 'notification', { message, at: Date.now() });
res.status(202).json({ delivered: true });
});
export default router;
import { useEffect, useRef, useState } from 'react';
type Notification = { message: string; at: number };
type Status = 'connecting' | 'open' | 'error';
export function useServerEvents(userId: string) {
const [latest, setLatest] = useState<Notification | null>(null);
const [status, setStatus] = useState<Status>('connecting');
const sourceRef = useRef<EventSource | null>(null);
useEffect(() => {
if (!userId) return;
const source = new EventSource(`/api/events?userId=${encodeURIComponent(userId)}`);
sourceRef.current = source;
source.onopen = () => setStatus('open');
source.onerror = () => setStatus('error');
source.addEventListener('notification', (evt) => {
try {
setLatest(JSON.parse((evt as MessageEvent).data) as Notification);
} catch {
// ignore malformed frames
}
});
return () => {
source.close();
sourceRef.current = null;
};
}, [userId]);
return { latest, status };
}
Server-Sent Events (SSE) is a one-way streaming protocol built on top of a long-lived HTTP response: the server keeps the connection open and pushes text/event-stream frames, while the browser's native EventSource handles parsing and automatic reconnection. It is a good fit for push notifications, live counters, or activity feeds where traffic flows server-to-client only and the operational cost of WebSockets is not justified.
In sseHub.ts the process keeps an in-memory registry of open connections in a Map keyed by user id. subscribe records the Response object and returns an unsubscribe function so callers can clean up on disconnect. publish serializes a payload into the SSE wire format inside writeEvent — a event: line, an incrementing id: line used for resumption, and a data: line — then flushes it to every response registered for that user. Because SSE frames are terminated by a blank line, the trailing \n\n is essential; omitting it means the browser buffers the event indefinitely.
notifications.ts wires the hub into an Express route. It sets the streaming headers (Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive) and calls flushHeaders() so the client receives the response head immediately rather than waiting for the first body chunk. A setInterval heartbeat writes a comment line (:\n\n) every 15 seconds to defeat idle proxies and load balancers that would otherwise kill a quiet connection. The req.on('close') handler clears the timer and unsubscribes, which is the single most common source of leaks in naive SSE code.
useServerEvents.ts is the client half. It opens an EventSource, attaches a named listener for the notification event, and exposes the latest parsed message plus a connection status. EventSource reconnects on its own, so the hook only needs to reflect state and close the socket on unmount. A caveat worth noting: browsers cap concurrent EventSource connections per domain over HTTP/1.1, and this in-memory hub does not survive multiple server instances, so a production deployment would back publish with Redis pub/sub.
Related snips
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
import axios, { AxiosError } from 'axios'
import { v4 as uuidv4 } from 'uuid'
const api = axios.create({
baseURL: import.meta.env.VITE_API_URL || 'http://localhost:3000/api/v1',
timeout: 15000,Axios API client with interceptors
import { create } from 'zustand'
import { persist } from 'zustand/middleware'
interface FilterState {
search: string
category: string | nullZustand for lightweight state management
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
import React from "react";
type FallbackProps = {
error: Error;
reset: () => void;
};React Error Boundary + error reporting hook
const withBundleAnalyzer = require('@next/bundle-analyzer')({
enabled: process.env.ANALYZE === 'true',
openAnalyzer: true,
});
/** @type {import('next').NextConfig} */Next.js bundle analyzer for targeted performance work
Share this code
Here's the card — post it anywhere.
