SSE Event IDs and last-event-id — Tracking Events and Recovering from Disconnections

import time
import json
from flask import Flask, Response, stream_with_context, request

app = Flask(__name__)

# In-memory event buffer (use Redis in production)
event_buffer = []
event_counter = 0

@app.route('/events/notifications')
def notification_stream():
    last_event_id = request.headers.get('Last-Event-ID')
    def generate():
        global event_counter

        # Replay missed events if Last-Event-ID is provided
        if last_event_id:
            last_id = int(last_event_id)
            missed = [e for e in event_buffer if e['id'] > last_id]
            for event in missed:
                yield f"id: {event['id']}\n"
                yield f"event: {event['type']}\n"
                yield f"data: {json.dumps(event['data'])}\n\n"

        # Send new events
        while True:
            event_counter += 1
            notification = {
                'id': event_counter,
                'type': 'notification',
                'data': {
                    'message': f'Event #{event_counter}',
                    'timestamp': time.time()
                }
            }

            event_buffer.append(notification)
            # Keep only last 1000 events
            if len(event_buffer) > 1000:
                event_buffer.pop(0)

            yield f"id: {notification['id']}\n"
            yield f"event: {notification['type']}\n"
            yield f"data: {json.dumps(notification['data'])}\n\n"

            time.sleep(2)

    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream'
    )

class ResilientEventSource {
  constructor(url, options = {}) {
    this.url = url;
    this.options = options;
    this.lastEventId = localStorage.getItem('sse_last_event_id');
    this.connect();
  }

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

    this.eventSource = new EventSource(url);

    this.eventSource.onopen = () => {
      console.log('SSE connection established');
    };

    this.eventSource.onmessage = (event) => {
      // Store the last event ID
      if (event.lastEventId) {
        this.lastEventId = event.lastEventId;
        localStorage.setItem('sse_last_event_id', event.lastEventId);
      }
      this.handleEvent(event);
    };

    this.eventSource.onerror = (error) => {
      console.warn('SSE connection error, will auto-reconnect');
      // EventSource automatically reconnects
      // The browser sends Last-Event-ID header automatically
    };

    this.eventSource.addEventListener('notification', (event) => {
      const data = JSON.parse(event.data);
      this.showNotification(data);
    });
  }

  handleEvent(event) {
    // Custom event handling
  }

  showNotification(data) {
    const container = document.getElementById('notifications');
    const div = document.createElement('div');
    div.textContent = data.message;
    container.prepend(div);
  }

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

// Usage
const sse = new ResilientEventSource('/events/notifications');

import redis
import json
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def store_event(event_id, event_type, data):
    """Store event in Redis for replay"""
    event = {
        'id': event_id,
        'type': event_type,
        'data': data,
        'timestamp': time.time()
    }
    r.xadd('sse:events', event, maxlen=1000)
    return event_id

def get_events_since(last_id):
    """Get all events after a given ID"""
    events = r.xrange('sse:events', min=f'({last_id}', max='+')
    return [
        {'id': e[0].decode(), 'type': e[1][b'type'].decode(), 'data': json.loads(e[1][b'data'])}
        for e in events
    ]