const { EventEmitter } = require('events');
class NotificationBus extends EventEmitter {
constructor(bufferSize = 100) {
super();
this.setMaxListeners(0);
this.seq = 0;
this.recent = [];
this.bufferSize = bufferSize;
}
publish(userId, type, payload) {
const event = { id: ++this.seq, userId, type, payload, ts: Date.now() };
this.recent.push(event);
if (this.recent.length > this.bufferSize) this.recent.shift();
this.emit(`user:${userId}`, event);
return event;
}
subscribe(userId, listener) {
const channel = `user:${userId}`;
this.on(channel, listener);
return () => this.off(channel, listener);
}
replaySince(userId, lastId) {
if (!lastId) return [];
return this.recent.filter((e) => e.userId === userId && e.id > lastId);
}
}
module.exports = new NotificationBus();
const express = require('express');
const bus = require('./notificationBus');
const router = express.Router();
function write(res, event) {
res.write(`id: ${event.id}\n`);
res.write(`event: notification\n`);
res.write(`data: ${JSON.stringify(event)}\n\n`);
}
router.get('/events', (req, res) => {
const userId = req.user.id;
res.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache, no-transform',
Connection: 'keep-alive',
'X-Accel-Buffering': 'no',
});
res.write('retry: 3000\n\n');
const lastId = Number(req.headers['last-event-id']) || 0;
for (const missed of bus.replaySince(userId, lastId)) write(res, missed);
const unsubscribe = bus.subscribe(userId, (event) => write(res, event));
const ping = setInterval(() => res.write(':ping\n\n'), 25000);
req.on('close', () => {
clearInterval(ping);
unsubscribe();
res.end();
});
});
// Example trigger: any part of the app can push to a user's feed.
router.post('/notify/:userId', express.json(), (req, res) => {
const event = bus.publish(req.params.userId, req.body.type, req.body.payload);
res.status(202).json({ id: event.id });
});
module.exports = router;
import { useEffect, useState } from 'react';
export function useNotifications(url = '/events') {
const [items, setItems] = useState([]);
const [connected, setConnected] = useState(false);
useEffect(() => {
const source = new EventSource(url, { withCredentials: true });
source.onopen = () => setConnected(true);
source.onerror = () => setConnected(false); // browser auto-reconnects
source.addEventListener('notification', (e) => {
const event = JSON.parse(e.data);
setItems((prev) => [event, ...prev].slice(0, 50));
});
return () => source.close();
}, [url]);
return { items, connected };
}
This snippet shows how to push live notifications from a Node/Express server to browsers using Server-Sent Events (SSE), a one-way streaming protocol built on a long-lived HTTP response. Unlike WebSockets, SSE runs over plain HTTP, carries a simple text/event-stream format, and gets automatic reconnection from the browser for free — making it ideal for a notification feed where only the server needs to push.
The notificationBus file is a tiny in-process pub/sub built on Node's EventEmitter. Each connected client registers a listener keyed by userId, and publish fans an event out to every listener for that user. It also keeps a small ring buffer of recent events in recent so a client that reconnects can replay anything it missed. Every event carries a monotonically increasing id, which is the key to reliable delivery.
The sseRouter file is where the protocol details live. GET /events sets the SSE headers — Content-Type: text/event-stream, no-cache, and keep-alive — and disables buffering with X-Accel-Buffering so proxies flush each chunk immediately. The write helper serializes each event into the wire format: an id: line, an event: line, and one or more data: lines terminated by a blank line. On connect it reads the standard Last-Event-ID header and calls bus.replaySince so no notification is dropped across a reconnect. A setInterval sends comment-only :ping lines to keep intermediaries from timing the connection out, and the req.on('close') handler unsubscribes and clears the timer to prevent leaks.
The useNotifications hook wraps the browser EventSource. It listens for the named notification event, parses the JSON payload, and prepends it to state. Because EventSource reconnects on its own and replays Last-Event-ID automatically, the hook stays small — it only tracks a connected flag via onopen/onerror and tears the stream down on unmount.
The trade-offs: SSE is unidirectional and, over HTTP/1.1, subject to the ~6-connection-per-origin limit, so it suits notifications far better than chat. The EventEmitter bus works only within a single process; scaling horizontally requires backing publish with Redis pub/sub or similar so every instance sees every event.
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)
class Comment < ApplicationRecord
belongs_to :article
belongs_to :author, class_name: "User"
validates :body, presence: true, length: { maximum: 2_000 }Live comments with model broadcasts + turbo_stream_from
import React from "react";
type FallbackProps = {
error: Error;
reset: () => void;
};React Error Boundary + error reporting hook
Share this code
Here's the card — post it anywhere.
