Server-Sent Events and Long Polling

const source = new EventSource('/events?rooms=general', { withCredentials: false });

retry: 3000

: keepalive

id: 41
event: message
data: {"room":"general","from":"ana","text":"hello"}

id: 42
event: mention
data: {"room":"general","from":"bo","text":"@ana lunch?"}

GET /events?rooms=general HTTP/1.1
Accept: text/event-stream
Last-Event-ID: 41

// Chatterbox notification feed over Server-Sent Events, plain node:http, no dependencies.
import { createServer } from 'node:http';
import { readFile } from 'node:fs/promises';

const PORT = Number(process.env.PORT ?? 8080);
const KEEPALIVE_MS = Number(process.env.KEEPALIVE_MS ?? 15000); // below any proxy idle timeout
const BUFFER_SIZE = 500; // how many events can be replayed
const SECRET = 'change-me'; // placeholder session value

const log = []; // replay buffer: { id, type, data }
let nextId = 1;
const clients = new Set(); // { res, rooms: Set<string> | null }

function format(ev) {
  // data is single-line JSON, so one data: line is enough
  return `id: ${ev.id}\nevent: ${ev.type}\ndata: ${ev.data}\n\n`;
}

function publish(type, payload) {
  const ev = { id: nextId++, type, data: JSON.stringify(payload) };
  log.push(ev);
  if (log.length > BUFFER_SIZE) log.shift();
  for (const c of clients) {
    if (!c.rooms || c.rooms.has(payload.room)) c.res.write(format(ev));
  }
  return ev.id;
}

function authorized(req) {
  const bearer = req.headers.authorization === `Bearer ${SECRET}`;
  const cookie = (req.headers.cookie ?? '').split(/;\s*/).includes(`session=${SECRET}`);
  return bearer || cookie;
}

function readJson(req) {
  return new Promise((resolve, reject) => {
    let body = '';
    req.setEncoding('utf8');
    req.on('data', (chunk) => {
      body += chunk;
      if (body.length > 16384) reject(new Error('too_large'));
    });
    req.on('end', () => {
      try { resolve(JSON.parse(body)); } catch (e) { reject(e); }
    });
  });
}

const server = createServer(async (req, res) => {
  const url = new URL(req.url, 'http://localhost');

  if (req.method === 'GET' && url.pathname === '/') {
    res.writeHead(200, { 'Content-Type': 'text/html; charset=utf-8', 'Set-Cookie': `session=${SECRET}; HttpOnly; SameSite=Lax; Path=/` });
    res.end(await readFile(new URL('./index.html', import.meta.url)));
    return;
  }

  if (req.method === 'GET' && url.pathname === '/events') {
    if (!authorized(req)) { res.writeHead(401).end(); return; }

    const rooms = url.searchParams.get('rooms');
    const client = { res, rooms: rooms ? new Set(rooms.split(',')) : null };

    res.writeHead(200, {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      'X-Accel-Buffering': 'no', // tell nginx not to buffer this response
    });
    res.flushHeaders(); // send the headers now so the client fires "open"
    res.write('retry: 3000\n\n'); // reconnection delay for this client

    // Resume: the header (automatic reconnect) or a query parameter (first load after a page refresh)
    const raw = req.headers['last-event-id'] ?? url.searchParams.get('lastEventId');
    if (raw) {
      const lastId = Number(raw);
      const oldest = log.length ? log[0].id : nextId;
      if (!Number.isInteger(lastId) || lastId < oldest - 1) {
        // the gap is larger than the buffer: tell the client to reload its state
        res.write(`event: reset\ndata: ${JSON.stringify({ oldest })}\n\n`);
      } else {
        for (const ev of log) {
          if (ev.id > lastId && (!client.rooms || client.rooms.has(JSON.parse(ev.data).room))) res.write(format(ev));
        }
      }
    }

    clients.add(client);
    const timer = setInterval(() => res.write(': keepalive\n\n'), KEEPALIVE_MS);
    req.on('close', () => { clearInterval(timer); clients.delete(client); });
    return;
  }

  if (req.method === 'POST' && url.pathname === '/messages') {
    if (!authorized(req)) { res.writeHead(401).end(); return; }
    try {
      const { room, text, from } = await readJson(req);
      if (!/^[a-z0-9-]{1,64}$/.test(room ?? '') || typeof text !== 'string' || !text) throw new Error('bad_request');
      const id = publish('message', { room, from: from ?? 'anonymous', text });
      res.writeHead(202, { 'Content-Type': 'application/json' }).end(JSON.stringify({ ok: true, id }));
    } catch {
      res.writeHead(400, { 'Content-Type': 'application/json' }).end(JSON.stringify({ ok: false, code: 'bad_request' }));
    }
    return;
  }

  res.writeHead(404).end();
});

server.listen(PORT, () => console.log(`Chatterbox SSE feed on http://localhost:${PORT}`));

<!doctype html>
<meta charset="utf-8">
<title>Chatterbox notifications</title>
<h1>Chatterbox notifications</h1>
<p id="status">connecting</p>
<ul id="feed"></ul>
<form id="form">
  <input id="text" placeholder="Say something" required>
  <button>Send</button>
</form>
<script>
  const status = document.querySelector('#status');
  const feed = document.querySelector('#feed');

  // Same origin, so the HttpOnly session cookie is sent automatically.
  const source = new EventSource('/events?rooms=general');

  source.addEventListener('open', () => { status.textContent = 'live'; });

  source.addEventListener('error', () => {
    // readyState CONNECTING (0) means the browser is retrying; CLOSED (2) means it gave up
    status.textContent = source.readyState === EventSource.CONNECTING ? 'reconnecting' : 'closed';
  });

  source.addEventListener('message', (event) => {
    const note = JSON.parse(event.data);
    const li = document.createElement('li');
    li.textContent = `#${event.lastEventId} [${note.room}] ${note.from}: ${note.text}`;
    feed.append(li);
  });

  source.addEventListener('reset', () => {
    // the server could not replay the gap: drop local state and reload it from the history API
    feed.replaceChildren();
  });

  document.querySelector('#form').addEventListener('submit', async (e) => {
    e.preventDefault();
    const text = document.querySelector('#text');
    await fetch('/messages', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ room: 'general', from: 'ana', text: text.value }),
    });
    text.value = '';
  });
</script>

node sse-server.mjs &
for i in 1 2 3; do
  curl -s -X POST -H 'Authorization: Bearer change-me' -H 'Content-Type: application/json' \
    -d "{\"room\":\"general\",\"from\":\"ana\",\"text\":\"msg $i\"}" localhost:8080/messages
done
curl -s -N -m 2 -H 'Authorization: Bearer change-me' -H 'Last-Event-ID: 1' 'localhost:8080/events?rooms=general'

{"ok":true,"id":1}
{"ok":true,"id":2}
{"ok":true,"id":3}
retry: 3000

id: 2
event: message
data: {"room":"general","from":"ana","text":"msg 2"}

id: 3
event: message
data: {"room":"general","from":"ana","text":"msg 3"}

: keepalive

location /events {
    proxy_pass http://127.0.0.1:8080;
    proxy_http_version 1.1;
    proxy_set_header Connection "";   # keep the upstream connection reusable
    proxy_buffering off;              # the application also sends X-Accel-Buffering: no
    proxy_cache off;
    gzip off;
    proxy_read_timeout 1h;            # default is 60 s; see below
}

// SSE over fetch(): custom headers, reconnect with Last-Event-ID. Runs in Node 18+ and in modern browsers.

export class SseParser {
  #buf = '';
  #first = true;
  #type = '';
  #data = [];
  #id = ''; // last event ID, persists across events
  retry = null;

  feed(chunk) {
    this.#buf += chunk;
    if (this.#first && this.#buf.length > 0) {
      if (this.#buf[0] === '') this.#buf = this.#buf.slice(1); // one leading BOM
      this.#first = false;
    }
    const events = [];
    const re = /\r\n|\r|\n/g;
    let start = 0;
    let m;
    while ((m = re.exec(this.#buf)) !== null) {
      // a lone CR at the very end may be the first half of CRLF: wait for the next chunk
      if (m[0] === '\r' && m.index === this.#buf.length - 1) break;
      const line = this.#buf.slice(start, m.index);
      start = m.index + m[0].length;
      const ev = this.#line(line);
      if (ev) events.push(ev);
    }
    this.#buf = this.#buf.slice(start); // an unterminated tail stays buffered
    return events;
  }

  #line(line) {
    if (line === '') { // blank line: dispatch
      if (this.#data.length === 0) { this.#type = ''; return null; }
      const ev = { type: this.#type || 'message', data: this.#data.join('\n'), lastEventId: this.#id };
      this.#type = ''; this.#data = [];
      return ev;
    }
    if (line.startsWith(':')) return null; // comment
    const i = line.indexOf(':');
    const field = i === -1 ? line : line.slice(0, i);
    let value = i === -1 ? '' : line.slice(i + 1);
    if (value.startsWith(' ')) value = value.slice(1);
    if (field === 'event') this.#type = value;
    else if (field === 'data') this.#data.push(value);
    else if (field === 'id' && !value.includes('\0')) this.#id = value;
    else if (field === 'retry' && /^\d+$/.test(value)) this.retry = Number(value);
    return null;
  }
}

export function subscribe(url, { token, onEvent, onState = () => {}, idleMs = 45000, signal }) {
  let lastId = '';
  let retryMs = 3000;
  let failures = 0;

  const run = async () => {
    while (!signal?.aborted) {
      const ac = new AbortController();
      const stop = () => ac.abort();
      signal?.addEventListener('abort', stop, { once: true });
      let idle;
      const arm = () => { clearTimeout(idle); idle = setTimeout(stop, idleMs); }; // watchdog for silent drops
      try {
        arm();
        const res = await fetch(url, {
          headers: {
            Accept: 'text/event-stream',
            Authorization: `Bearer ${token}`, // impossible with EventSource
            ...(lastId && { 'Last-Event-ID': lastId }),
          },
          cache: 'no-store',
          signal: ac.signal,
        });
        if (res.status === 401 || res.status === 403) { onState('unauthorized'); return; } // do not retry
        if (!res.ok || !res.headers.get('content-type')?.startsWith('text/event-stream')) throw new Error(`HTTP ${res.status}`);
        onState('open');
        failures = 0;
        const parser = new SseParser();
        const reader = res.body.getReader();
        const decoder = new TextDecoder('utf-8');
        for (;;) {
          const { value, done } = await reader.read();
          if (done) break;
          arm();
          for (const ev of parser.feed(decoder.decode(value, { stream: true }))) {
            if (ev.lastEventId) lastId = ev.lastEventId;
            onEvent(ev);
          }
          if (parser.retry !== null) retryMs = parser.retry;
        }
      } catch (err) {
        if (signal?.aborted) return;
      } finally {
        clearTimeout(idle);
        signal?.removeEventListener('abort', stop);
      }
      failures++;
      onState('reconnecting');
      const backoff = Math.min(retryMs * 2 ** (failures - 1), 30000);
      await new Promise((r) => setTimeout(r, backoff * (0.5 + Math.random() / 2))); // jitter
    }
  };
  return run();
}

// Command-line demo: node sse-fetch-client.mjs
if (import.meta.url === `file://${process.argv[1]}`) {
  const ac = new AbortController();
  const max = Number(process.env.MAX_EVENTS ?? Infinity);
  let seen = 0;
  await subscribe(process.env.FEED_URL ?? 'http://localhost:8080/events', {
    token: process.env.TOKEN ?? 'change-me',
    signal: ac.signal,
    onState: (s) => console.log(`state: ${s}`),
    onEvent: (ev) => {
      console.log(`${ev.type} id=${ev.lastEventId} ${ev.data}`);
      if (++seen >= max) ac.abort();
    },
  });
}

$ MAX_EVENTS=3 node sse-fetch-client.mjs      # three messages are posted after it starts
state: open
message id=1 {"room":"general","from":"ana","text":"msg 1"}
message id=2 {"room":"general","from":"ana","text":"msg 2"}
message id=3 {"room":"general","from":"ana","text":"msg 3"}

// Cursor-based long polling for Chatterbox rooms, plain node:http.
import { createServer } from 'node:http';

const PORT = Number(process.env.PORT ?? 8081);
const MAX_HOLD_MS = Number(process.env.MAX_HOLD_MS ?? 25000); // under a 30-60 s proxy idle timeout
const KEEP = 1000; // events kept for cursors

const events = []; // { seq, room, from, text }
let seq = 0;
const waiters = new Set(); // { since, room, send, timer }

function after(since, room) {
  return events.filter((e) => e.seq > since && (!room || e.room === room)).slice(0, 100);
}

const server = createServer((req, res) => {
  const url = new URL(req.url, 'http://localhost');

  if (req.method === 'POST' && url.pathname === '/messages') {
    let body = '';
    req.on('data', (c) => { body += c; });
    req.on('end', () => {
      try {
        const { room, from, text } = JSON.parse(body);
        const ev = { seq: ++seq, room, from, text };
        events.push(ev);
        if (events.length > KEEP) events.shift();
        for (const w of [...waiters]) { // wake every parked request whose room matches
          if (!w.room || w.room === room) w.send();
        }
        res.writeHead(202, { 'Content-Type': 'application/json' }).end(JSON.stringify({ seq: ev.seq }));
      } catch {
        res.writeHead(400).end();
      }
    });
    return;
  }

  if (req.method === 'GET' && url.pathname === '/poll') {
    const since = Number(url.searchParams.get('since') ?? 0);
    const room = url.searchParams.get('room');
    const wait = Math.min(Number(url.searchParams.get('wait') ?? MAX_HOLD_MS), MAX_HOLD_MS);
    const headers = { 'Cache-Control': 'no-store' };

    if (!Number.isInteger(since) || since < 0 || since > seq) { res.writeHead(400, headers).end(); return; }
    if (events.length && since < events[0].seq - 1) {
      // the cursor fell off the retained window: the client must resync from the history API
      res.writeHead(410, headers).end();
      return;
    }

    const reply = () => {
      const batch = after(since, room);
      if (batch.length === 0) return false;
      res.writeHead(200, { ...headers, 'Content-Type': 'application/json' });
      res.end(JSON.stringify({ events: batch, cursor: batch[batch.length - 1].seq }));
      return true;
    };
    if (reply()) return; // something is already waiting: answer immediately

    const waiter = {
      room,
      send: () => { if (reply()) cleanup(); },
      timer: setTimeout(() => { cleanup(); res.writeHead(204, headers).end(); }, wait),
    };
    const cleanup = () => { clearTimeout(waiter.timer); waiters.delete(waiter); };
    waiters.add(waiter);
    req.on('close', cleanup); // the client went away while parked
    return;
  }

  res.writeHead(404).end();
});

server.listen(PORT, () => console.log(`Chatterbox long-poll on http://localhost:${PORT}`));

// Long-poll client with cursor, jitter and exponential backoff. Node 18+ or any modern browser.
const BASE = process.env.POLL_URL ?? 'http://localhost:8081';

async function resync() {
  // Real code would call the history API and return the newest seq; here we restart at 0.
  return 0;
}

export async function pollLoop({ room, onEvents, signal }) {
  let cursor = 0;
  let failures = 0;
  while (!signal?.aborted) {
    try {
      const res = await fetch(`${BASE}/poll?room=${encodeURIComponent(room)}&since=${cursor}&wait=25000`, {
        cache: 'no-store',
        // slightly longer than the server hold, so the server normally answers first
        signal: AbortSignal.any([AbortSignal.timeout(35000), ...(signal ? [signal] : [])]),
      });
      if (res.status === 204) { // timeout, nothing new: ask again after a small random pause
        failures = 0;
        await sleep(Math.random() * 500);
        continue;
      }
      if (res.status === 410) { cursor = await resync(); continue; }
      if (!res.ok) throw new Error(`HTTP ${res.status}`);
      const { events, cursor: next } = await res.json();
      cursor = next; // advance only after the batch was parsed
      failures = 0;
      onEvents(events);
    } catch (err) {
      if (signal?.aborted) return;
      failures++;
      const base = Math.min(1000 * 2 ** (failures - 1), 30000);
      await sleep(base * (0.5 + Math.random() / 2)); // exponential backoff with jitter
    }
  }
}

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

if (import.meta.url === `file://${process.argv[1]}`) {
  const ac = new AbortController();
  let n = 0;
  await pollLoop({
    room: 'general',
    signal: ac.signal,
    onEvents: (events) => {
      for (const e of events) console.log(`seq=${e.seq} ${e.from}: ${e.text}`);
      n += events.length;
      if (n >= Number(process.env.MAX_EVENTS ?? Infinity)) ac.abort();
    },
  });
}

$ curl -s -o /dev/null -w '%{http_code} %{time_total}\n' 'localhost:8081/poll?since=0&room=general'
204 3.003614
$ curl -s -X POST -d '{"room":"general","from":"ana","text":"one"}' localhost:8081/messages
{"seq":1}
$ curl -s 'localhost:8081/poll?since=0&room=general'
{"events":[{"seq":1,"room":"general","from":"ana","text":"one"}],"cursor":1}
$ curl -s -w ' %{http_code} %{time_total}\n' 'localhost:8081/poll?since=1&room=general'   # a post arrives 1 s later
{"events":[{"seq":2,"room":"general","from":"bo","text":"two"}],"cursor":2} 200 1.011529
$ curl -s -o /dev/null -w '%{http_code}\n' 'localhost:8081/poll?since=99'
400
$ node poll-client.mjs        # a message is posted after the first 204
seq=1 ana: one
seq=2 bo: two
seq=3 cy: three