Sse Last Event Id

const http = require('http');

// Event store
const eventStore = [];
const MAX_EVENTS = 5000;
let globalId = 0;

// Generate events periodically
function generateEvent() {
  globalId++;
  const event = {
    id: `evt-${globalId}`,
    data: {
      id: globalId,
      type: 'update',
      value: Math.random(),
      timestamp: Date.now(),
    },
  };
  eventStore.push(event);
  if (eventStore.length > MAX_EVENTS) {
    eventStore.shift();
  }
}

setInterval(generateEvent, 500);

http.createServer((req, res) => {
  if (req.url !== '/events') {
    res.writeHead(404);
    res.end();
    return;
  }

  // Read Last-Event-ID from header
  const lastEventId = req.headers['last-event-id'];
  const lastSeq = lastEventId 
    ? parseInt(lastEventId.replace('evt-', '')) 
    : 0;

  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
  });

  // Replay missed events
  const missedEvents = eventStore.filter(e => {
    const eventSeq = parseInt(e.id.replace('evt-', ''));
    return eventSeq > lastSeq;
  });

  if (missedEvents.length > 0) {
    console.log(`Replaying ${missedEvents.length} missed events`);
    missedEvents.forEach(event => {
      res.write(`id: ${event.id}\ndata: ${JSON.stringify(event.data)}\n\n`);
    });
  }

  // Subscribe to new events
  const listener = (event) => {
    res.write(`id: ${event.id}\ndata: ${JSON.stringify(event.data)}\n\n`);
  };

  // In production, use an event emitter instead
  const interval = setInterval(() => {
    const recentEvents = eventStore.filter(e => e.id.startsWith('evt-'));
    if (recentEvents.length > 0) {
      const lastEvent = recentEvents[recentEvents.length - 1];
      const eventSeq = parseInt(lastEvent.id.replace('evt-', ''));
      if (eventSeq > lastSeq) {
        lastSeq = eventSeq;
        res.write(`id: ${lastEvent.id}\ndata: ${JSON.stringify(lastEvent.data)}\n\n`);
      }
    }
  }, 500);

  req.on('close', () => {
    clearInterval(interval);
  });
}).listen(3000);

class ReliableEventSource {
  constructor(url) {
    this.url = url;
    this.lastEventId = localStorage.getItem('sse-last-id') || null;
    this.listeners = new Map();
    this.connect();
  }

  connect() {
    const url = this.lastEventId
      ? `${this.url}?lastId=${this.lastEventId}`
      : this.url;

    this.source = new EventSource(url);

    this.source.onopen = () => {
      console.log('Connected, last ID:', this.lastEventId);
      this.emit('connected');
    };

    this.source.onmessage = (event) => {
      // Track last event ID
      if (event.lastEventId) {
        this.lastEventId = event.lastEventId;
        localStorage.setItem('sse-last-id', event.lastEventId);
      }
      this.emit('message', JSON.parse(event.data));
    };

    this.source.onerror = () => {
      // EventSource will auto-reconnect with Last-Event-ID header
      this.emit('reconnecting');
    };
  }

  on(event, callback) {
    if (!this.listeners.has(event)) {
      this.listeners.set(event, new Set());
    }
    this.listeners.get(event).add(callback);
    return () => this.listeners.get(event)?.delete(callback);
  }

  emit(event, data) {
    this.listeners.get(event)?.forEach(cb => cb(data));
  }

  clearRecovery() {
    this.lastEventId = null;
    localStorage.removeItem('sse-last-id');
  }

  close() {
    this.source.close();
  }
}

// Usage with persistent recovery
const sse = new ReliableEventSource('/events');

sse.on('message', (data) => {
  updateDashboard(data);
});

sse.on('connected', () => {
  updateStatus('connected');
});

sse.on('reconnecting', () => {
  updateStatus('reconnecting');
});

// Manual recovery reset
document.getElementById('refresh-btn').onclick = () => {
  sse.clearRecovery();
  sse.close();
  window.location.reload();
};

// Server with ordered event delivery
const http = require('http');
const { EventEmitter } = require('events');

class OrderedSSEServer {
  constructor() {
    this.emitter = new EventEmitter();
    this.sequenceMap = new Map(); // connection -> lastSeq
    this.globalSeq = 0;
  }

  handleConnection(req, res) {
    const lastEventId = parseInt(
      req.headers['last-event-id']?.replace('seq-', '')
    ) || 0;

    const connId = `${Date.now()}-${Math.random()}`;
    this.sequenceMap.set(connId, lastEventId);

    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
    });

    // Send missed events from buffer
    const buffer = this.getEventBuffer(lastEventId);
    buffer.forEach(entry => {
      res.write(
        `id: seq-${entry.seq}\n` +
        `event: ${entry.event}\n` +
        `data: ${JSON.stringify(entry.data)}\n` +
        '\n'
      );
    });

    // Subscribe to new events with ordering
    const listener = (eventData) => {
      const currentSeq = this.sequenceMap.get(connId) || 0;
      const nextSeq = currentSeq + 1;

      const message = 
        `id: seq-${nextSeq}\n` +
        `event: ${eventData.event}\n` +
        `data: ${JSON.stringify(eventData.data)}\n` +
        '\n';

      res.write(message);
      this.sequenceMap.set(connId, nextSeq);
    };

    this.emitter.on('event', listener);

    req.on('close', () => {
      this.emitter.off('event', listener);
      this.sequenceMap.delete(connId);
    });
  }

  broadcast(event, data) {
    this.globalSeq++;
    const eventData = { seq: this.globalSeq, event, data };
    this.eventBuffer.push(eventData);
    if (this.eventBuffer.length > 1000) {
      this.eventBuffer.shift();
    }
    this.emitter.emit('event', eventData);
  }

  getEventBuffer(sinceSeq) {
    return this.eventBuffer.filter(e => e.seq > sinceSeq);
  }
}

const sseServer = new OrderedSSEServer();
http.createServer((req, res) => sseServer.handleConnection(req, res))
  .listen(3000);

// Broadcast events from anywhere
setInterval(() => {
  sseServer.broadcast('update', {
    value: Math.random(),
    time: Date.now(),
  });
}, 1000);