1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

const events = new EventSource('/api/events');
 
// Track connection state
let connectionAttempts = 0;
let lastConnectedAt = null;
 
events.onopen = () => {
    connectionAttempts++;
    lastConnectedAt = new Date();
    console.log(`Connected (attempt #${connectionAttempts}) at ${lastConnectedAt}`);
    
    // Reset reconnection UI
    hideReconnectionBanner();
};
 
events.onerror = (error) => {
    // Check the readyState to understand what's happening
    switch (events.readyState) {
        case EventSource.CONNECTING:
            // Actively trying to reconnect
            console.log('Reconnecting...');
            showReconnectionBanner('Reconnecting to server...');
            break;
            
        case EventSource.OPEN:
            // Error during open connection (rare)
            console.error('Error while connected:', error);
            break;
            
        case EventSource.CLOSED:
            // Permanently closed - won't retry
            console.error('Connection permanently closed');
            showReconnectionBanner('Connection failed. Please refresh the page.');
            break;
    }
};
 
// Note: EventSource doesn't expose the actual retry interval
// or provide hooks into the retry mechanism directly.
// For more control, see custom retry strategies below.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# Initial connection request
GET /api/events HTTP/1.1
Host: example.com
Accept: text/event-stream
 
# Server response - streaming events with IDs
HTTP/1.1 200 OK
Content-Type: text/event-stream
 
id: 1001
event: message
data: {"text": "Hello"}
 
id: 1002
event: message
data: {"text": "World"}
 
id: 1003
event: notification
data: {"type": "alert"}
 
# Connection drops here...
 
# Reconnection request - note the Last-Event-ID header!
GET /api/events HTTP/1.1
Host: example.com
Accept: text/event-stream
Last-Event-ID: 1003
 
# Server should resume from 1004 onwards
HTTP/1.1 200 OK
Content-Type: text/event-stream
 
id: 1004
event: message
data: {"text": "First message after reconnect"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

const express = require('express');
const app = express();
 
// In-memory message store (use Redis/DB in production)
const messageStore = new Map(); // id -> { event, data, timestamp }
let currentEventId = 0;
 
// Store messages for recovery
function storeMessage(event, data) {
    const id = String(++currentEventId);
    messageStore.set(id, { id, event, data, timestamp: Date.now() });
    
    // Cleanup old messages (keep last 1000 or last hour)
    pruneOldMessages();
    
    return id;
}
 
function pruneOldMessages() {
    const ONE_HOUR = 60 * 60 * 1000;
    const cutoff = Date.now() - ONE_HOUR;
    
    for (const [id, msg] of messageStore) {
        if (msg.timestamp < cutoff) {
            messageStore.delete(id);
        }
    }
}
 
// Get messages after a given ID
function getMessagesSince(lastEventId) {
    if (!lastEventId) return [];
    
    const messages = [];
    let foundLast = false;
    
    for (const [id, msg] of messageStore) {
        if (foundLast) {
            messages.push(msg);
        }
        if (id === lastEventId) {
            foundLast = true;
        }
    }
    
    return messages;
}
 
// SSE endpoint with recovery support
app.get('/api/events', (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    
    // Check for recovery request
    const lastEventId = req.headers['last-event-id'];
    
    if (lastEventId) {
        console.log(`Client reconnecting, last ID: ${lastEventId}`);
        
        // Send missed messages
        const missedMessages = getMessagesSince(lastEventId);
        console.log(`Sending ${missedMessages.length} missed messages`);
        
        for (const msg of missedMessages) {
            res.write(`id: ${msg.id}\n`);
            res.write(`event: ${msg.event}\n`);
            res.write(`data: ${JSON.stringify(msg.data)}\n\n`);
        }
    }
    
    // Set recommended retry interval
    res.write('retry: 5000\n\n');
    
    // Send connection confirmation
    res.write(': connected\n\n');
    
    // Subscribe to future events
    const clientId = registerClient(res);
    
    req.on('close', () => {
        unregisterClient(clientId);
    });
});
 
// When broadcasting, store and include ID
function broadcast(event, data) {
    const id = storeMessage(event, data);
    const message = `id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
    
    for (const client of clients.values()) {
        client.write(message);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

HTTP/1.1 200 OK
Content-Type: text/event-stream
 
retry: 10000
: Client should wait 10 seconds before reconnecting
 
data: First message
 
: During high load, increase retry interval
retry: 30000
 
data: Server under load, backing off
 
: Back to normal
retry: 5000
 
data: Load normalized

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

interface RetryConfig {
    initialDelay: number;      // Start delay (ms)
    maxDelay: number;          // Maximum delay (ms)
    backoffMultiplier: number; // Delay multiplier per attempt
    jitterFactor: number;      // Random jitter (0-1)
    maxRetries: number;        // Max attempts (0 = infinite)
}
 
const DEFAULT_RETRY_CONFIG: RetryConfig = {
    initialDelay: 1000,
    maxDelay: 60000,
    backoffMultiplier: 2,
    jitterFactor: 0.1,
    maxRetries: 0,  // Infinite
};
 
class ResilientEventSource {
    private url: string;
    private eventSource: EventSource | null = null;
    private config: RetryConfig;
    private retryCount = 0;
    private currentDelay: number;
    private retryTimeout: ReturnType<typeof setTimeout> | null = null;
    private isIntentionallyClosed = false;
    private lastEventId: string | null = null;
    
    // Event handlers
    onmessage: ((event: MessageEvent) => void) | null = null;
    onerror: ((error: Event, retryInfo: RetryInfo) => void) | null = null;
    onopen: (() => void) | null = null;
    onreconnecting: ((info: RetryInfo) => void) | null = null;
    
    constructor(url: string, config: Partial<RetryConfig> = {}) {
        this.url = url;
        this.config = { ...DEFAULT_RETRY_CONFIG, ...config };
        this.currentDelay = this.config.initialDelay;
        this.connect();
    }
    
    private connect() {
        if (this.isIntentionallyClosed) return;
        
        // Include last event ID in URL if reconnecting
        let connectionUrl = this.url;
        if (this.lastEventId) {
            const separator = this.url.includes('?') ? '&' : '?';
            connectionUrl = `${this.url}${separator}lastEventId=${this.lastEventId}`;
        }
        
        this.eventSource = new EventSource(connectionUrl);
        
        this.eventSource.onopen = () => {
            // Reset retry state on successful connection
            this.retryCount = 0;
            this.currentDelay = this.config.initialDelay;
            this.onopen?.();
        };
        
        this.eventSource.onmessage = (event) => {
            if (event.lastEventId) {
                this.lastEventId = event.lastEventId;
            }
            this.onmessage?.(event);
        };
        
        this.eventSource.onerror = (error) => {
            if (this.isIntentionallyClosed) return;
            
            // Close the failed connection
            this.eventSource?.close();
            this.eventSource = null;
            
            // Check retry limits
            if (this.config.maxRetries > 0 && 
                this.retryCount >= this.config.maxRetries) {
                this.onerror?.(error, {
                    attempt: this.retryCount,
                    willRetry: false,
                    nextRetryIn: 0,
                });
                return;
            }
            
            // Calculate next delay with jitter
            const jitter = this.currentDelay * this.config.jitterFactor * Math.random();
            const delay = Math.min(
                this.currentDelay + jitter,
                this.config.maxDelay
            );
            
            this.retryCount++;
            
            const retryInfo: RetryInfo = {
                attempt: this.retryCount,
                willRetry: true,
                nextRetryIn: delay,
            };
            
            this.onreconnecting?.(retryInfo);
            this.onerror?.(error, retryInfo);
            
            // Schedule retry
            this.retryTimeout = setTimeout(() => {
                this.currentDelay = Math.min(
                    this.currentDelay * this.config.backoffMultiplier,
                    this.config.maxDelay
                );
                this.connect();
            }, delay);
        };
    }
    
    addEventListener(event: string, handler: (e: MessageEvent) => void) {
        this.eventSource?.addEventListener(event, (e) => {
            if ((e as MessageEvent).lastEventId) {
                this.lastEventId = (e as MessageEvent).lastEventId;
            }
            handler(e as MessageEvent);
        });
    }
    
    close() {
        this.isIntentionallyClosed = true;
        if (this.retryTimeout) {
            clearTimeout(this.retryTimeout);
        }
        this.eventSource?.close();
    }
    
    // Force immediate reconnection
    reconnect() {
        this.eventSource?.close();
        this.currentDelay = this.config.initialDelay;
        this.retryCount = 0;
        this.connect();
    }
    
    get readyState(): number {
        return this.eventSource?.readyState ?? EventSource.CLOSED;
    }
}
 
interface RetryInfo {
    attempt: number;
    willRetry: boolean;
    nextRetryIn: number;
}
 
// Usage
const events = new ResilientEventSource('/api/events', {
    initialDelay: 1000,
    maxDelay: 30000,
    backoffMultiplier: 1.5,
    jitterFactor: 0.2,
    maxRetries: 10,
});
 
events.onopen = () => console.log('Connected!');
 
events.onmessage = (e) => console.log('Message:', e.data);
 
events.onreconnecting = (info) => {
    console.log(`Reconnecting (attempt ${info.attempt}) in ${info.nextRetryIn}ms`);
    showReconnectionUI(info);
};
 
events.onerror = (error, info) => {
    if (!info.willRetry) {
        console.error('Max retries exceeded, giving up');
        showPermanentErrorUI();
    }
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

class AuthAwareEventSource {
    private events: EventSource | null = null;
    private url: string;
    private getToken: () => Promise<string>;
    private refreshToken: () => Promise<string>;
    
    constructor(
        url: string, 
        tokenProvider: { get: () => Promise<string>; refresh: () => Promise<string> }
    ) {
        this.url = url;
        this.getToken = tokenProvider.get;
        this.refreshToken = tokenProvider.refresh;
    }
    
    async connect() {
        const token = await this.getToken();
        const urlWithToken = `${this.url}?token=${encodeURIComponent(token)}`;
        
        this.events = new EventSource(urlWithToken);
        
        this.events.onerror = async (error) => {
            if (this.events?.readyState === EventSource.CLOSED) {
                // Connection failed - might be auth issue
                await this.handleConnectionFailure();
            } else {
                // Reconnecting - EventSource handles automatically
                console.log('SSE reconnecting...');
            }
        };
    }
    
    private async handleConnectionFailure() {
        try {
            // Try refreshing the token
            console.log('Connection failed, attempting token refresh...');
            const newToken = await this.refreshToken();
            
            // Retry connection with new token
            await this.connect();
            console.log('Reconnected with refreshed token');
        } catch (refreshError) {
            console.error('Token refresh failed:', refreshError);
            this.onAuthFailure?.();
        }
    }
    
    onAuthFailure: (() => void) | null = null;
    
    close() {
        this.events?.close();
    }
}
 
// Usage with token refresh
const events = new AuthAwareEventSource('/api/events', {
    get: () => Promise.resolve(localStorage.getItem('accessToken') || ''),
    refresh: async () => {
        const response = await fetch('/auth/refresh', {
            method: 'POST',
            credentials: 'include',
        });
        const { accessToken } = await response.json();
        localStorage.setItem('accessToken', accessToken);
        return accessToken;
    },
});
 
events.onAuthFailure = () => {
    // Redirect to login
    window.location.href = '/login?returnTo=' + encodeURIComponent(window.location.href);
};
 
await events.connect();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

class StreamCoordinator {
    private streams: Map<string, EventSource> = new Map();
    private lastEventIds: Map<string, string> = new Map();
    private reconnectInProgress = false;
    
    addStream(name: string, url: string, handlers: StreamHandlers) {
        const fullUrl = this.buildUrl(url, name);
        const stream = new EventSource(fullUrl);
        
        stream.onmessage = (event) => {
            if (event.lastEventId) {
                this.lastEventIds.set(name, event.lastEventId);
            }
            handlers.onMessage(event);
        };
        
        stream.onerror = () => {
            this.handleStreamError(name, url, handlers);
        };
        
        this.streams.set(name, stream);
    }
    
    private buildUrl(baseUrl: string, streamName: string): string {
        const lastId = this.lastEventIds.get(streamName);
        if (lastId) {
            const sep = baseUrl.includes('?') ? '&' : '?';
            return `${baseUrl}${sep}lastEventId=${lastId}`;
        }
        return baseUrl;
    }
    
    private async handleStreamError(
        name: string, 
        url: string, 
        handlers: StreamHandlers
    ) {
        // Close failed stream
        this.streams.get(name)?.close();
        this.streams.delete(name);
        
        // Coordinate with other streams - don't spam reconnections
        if (this.reconnectInProgress) {
            console.log(`Stream '${name}' waiting for coordinated reconnect`);
            return;
        }
        
        // Check if this is a systemic failure
        const healthyStreams = Array.from(this.streams.values())
            .filter(s => s.readyState === EventSource.OPEN);
        
        if (healthyStreams.length === 0) {
            // All streams failed - likely server-side issue
            console.log('All streams failed, coordinated reconnection...');
            await this.coordinatedReconnect();
        } else {
            // Just this stream failed - reconnect independently
            console.log(`Stream '${name}' reconnecting`);
            setTimeout(() => {
                this.addStream(name, url, handlers);
            }, 1000 + Math.random() * 2000); // Jittered delay
        }
    }
    
    private async coordinatedReconnect() {
        this.reconnectInProgress = true;
        
        // Close all streams
        for (const stream of this.streams.values()) {
            stream.close();
        }
        
        // Wait for server to recover
        await this.waitForServerHealth();
        
        // Reconnect all streams with jitter
        this.reconnectInProgress = false;
        // ... reconnection logic
    }
    
    private async waitForServerHealth(): Promise<void> {
        // Exponential backoff health check
        let delay = 1000;
        while (delay < 60000) {
            try {
                const response = await fetch('/api/health');
                if (response.ok) return;
            } catch {}
            await new Promise(r => setTimeout(r, delay));
            delay *= 2;
        }
    }
}
 
interface StreamHandlers {
    onMessage: (event: MessageEvent) => void;
    onError?: (error: Event) => void;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

// For high-volume streams, store processed checkpoints
// This allows recovery even after page reload
 
class CheckpointedEventSource {
    private events: EventSource | null = null;
    private checkpoint: string | null = null;
    private storageKey: string;
    
    constructor(private url: string, streamId: string) {
        this.storageKey = `sse_checkpoint_${streamId}`;
        this.checkpoint = this.loadCheckpoint();
    }
    
    private loadCheckpoint(): string | null {
        try {
            return localStorage.getItem(this.storageKey);
        } catch {
            return null;
        }
    }
    
    private saveCheckpoint(id: string) {
        try {
            localStorage.setItem(this.storageKey, id);
            this.checkpoint = id;
        } catch {
            // localStorage might be full or disabled
        }
    }
    
    connect(handlers: {
        onMessage: (event: MessageEvent) => Promise<void>; // Async for processing
        onError?: (error: Event) => void;
    }) {
        let connectionUrl = this.url;
        if (this.checkpoint) {
            const sep = this.url.includes('?') ? '&' : '?';
            connectionUrl = `${this.url}${sep}afterId=${this.checkpoint}`;
        }
        
        this.events = new EventSource(connectionUrl);
        
        this.events.onmessage = async (event) => {
            // Process message
            await handlers.onMessage(event);
            
            // Only checkpoint after successful processing
            if (event.lastEventId) {
                this.saveCheckpoint(event.lastEventId);
            }
        };
        
        this.events.onerror = (error) => {
            handlers.onError?.(error);
            // EventSource will auto-reconnect, and we'll send the checkpoint
        };
    }
    
    // Allow manual checkpoint clear (e.g., when user logs out)
    clearCheckpoint() {
        try {
            localStorage.removeItem(this.storageKey);
            this.checkpoint = null;
        } catch {}
    }
    
    close() {
        this.events?.close();
    }
}
 
// Server must handle 'afterId' parameter for recovery
// GET /api/events?afterId=12345
 
// Usage
const events = new CheckpointedEventSource('/api/events', 'main-stream');
 
events.connect({
    onMessage: async (event) => {
        const data = JSON.parse(event.data);
        await processMessageWithRetry(data); // Your processing logic
        // Checkpoint saved after this returns successfully
    },
    onError: (error) => {
        showReconnectionNotice();
    },
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

const express = require('express');
const Redis = require('ioredis');
 
const app = express();
const redis = new Redis();
 
// Constants
const MESSAGE_TTL_SECONDS = 3600; // 1 hour recovery window
const MAX_STORED_MESSAGES = 10000;
const KEEPALIVE_INTERVAL_MS = 25000;
 
// Reconnection surge protection
class ConnectionThrottler {
    constructor(maxConnectionsPerSecond = 100) {
        this.maxPerSecond = maxConnectionsPerSecond;
        this.currentSecond = 0;
        this.connectionsThisSecond = 0;
    }
    
    canConnect() {
        const now = Math.floor(Date.now() / 1000);
        if (now !== this.currentSecond) {
            this.currentSecond = now;
            this.connectionsThisSecond = 0;
        }
        
        if (this.connectionsThisSecond >= this.maxPerSecond) {
            return false;
        }
        
        this.connectionsThisSecond++;
        return true;
    }
}
 
const throttler = new ConnectionThrottler(100);
 
// Store message in Redis with automatic expiration
async function storeMessage(event, data) {
    const id = await redis.incr('sse:event_counter');
    const message = JSON.stringify({ id, event, data, timestamp: Date.now() });
    
    await redis.multi()
        .zadd('sse:messages', id, message)
        .expire('sse:messages', MESSAGE_TTL_SECONDS * 2)
        .exec();
    
    // Trim to max size
    await redis.zremrangebyrank('sse:messages', 0, -MAX_STORED_MESSAGES - 1);
    
    return String(id);
}
 
// Get messages after a given ID
async function getMessagesSince(lastEventId) {
    if (!lastEventId) return [];
    
    // Get all messages with score > lastEventId
    const messages = await redis.zrangebyscore(
        'sse:messages',
        Number(lastEventId) + 1,
        '+inf'
    );
    
    return messages.map(m => JSON.parse(m));
}
 
// SSE endpoint
app.get('/api/events', async (req, res) => {
    // Surge protection
    if (!throttler.canConnect()) {
        res.status(503)
           .set('Retry-After', '2')
           .send('Too many connections, try again shortly');
        return;
    }
    
    // Set SSE headers
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache, no-store, must-revalidate');
    res.setHeader('Connection', 'keep-alive');
    res.setHeader('X-Accel-Buffering', 'no');
    
    // Handle Last-Event-ID for recovery
    const lastEventId = req.headers['last-event-id'];
    
    if (lastEventId) {
        console.log(`Recovery request from ID: ${lastEventId}`);
        
        const missedMessages = await getMessagesSince(lastEventId);
        console.log(`Sending ${missedMessages.length} missed messages`);
        
        for (const msg of missedMessages) {
            res.write(`id: ${msg.id}\nevent: ${msg.event}\ndata: ${JSON.stringify(msg.data)}\n\n`);
        }
    }
    
    // Suggest retry interval
    res.write('retry: 5000\n\n');
    
    // Connection confirmation
    res.write(': connected\n\n');
    
    // Keepalive
    const keepaliveTimer = setInterval(() => {
        if (!res.writableEnded) {
            res.write(': keepalive\n\n');
        }
    }, KEEPALIVE_INTERVAL_MS);
    
    // Subscribe to Redis pub/sub for new messages
    const subscriber = new Redis();
    subscriber.subscribe('sse:broadcast');
    
    subscriber.on('message', (channel, message) => {
        if (!res.writableEnded) {
            res.write(message);
        }
    });
    
    // Cleanup on disconnect
    req.on('close', () => {
        clearInterval(keepaliveTimer);
        subscriber.unsubscribe();
        subscriber.disconnect();
    });
});
 
// Broadcast to all clients
async function broadcast(event, data) {
    const id = await storeMessage(event, data);
    const message = `id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
    
    await redis.publish('sse:broadcast', message);
}
 
// Graceful shutdown
process.on('SIGTERM', async () => {
    console.log('Shutting down, closing SSE connections...');
    // The framework will handle closing connections
    // Clients will receive close event and reconnect
    process.exit(0);
});
 
app.listen(3000);