CodeIssuesDiscussionsWikiPull RequestsProjectsCommitsActionsReleasesContributorsPulse● GatesSecuritySettingsDeploymentsPipelineInsightsAgents✨ Explain✨ Ask AI✨ Workspace✨ Spec✨ Tests▓ Debt Map✨ NL Search🏛 Archaeology
Blame · Line-by-line history
sse.ts
Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.
| febd4f0 | 1 | /** |
| 13cbd17 | 2 | * Topic-based pub/sub broadcaster for Server-Sent Events. |
| febd4f0 | 3 | * |
| 13cbd17 | 4 | * Two modes, selected at startup: |
| febd4f0 | 5 | * |
| 13cbd17 | 6 | * IN-MEMORY (no REDIS_URL set) |
| 7 | * Module-level `Map<topic, Set<handler>>`. `publish` iterates | |
| 8 | * subscribers synchronously (fire-and-forget). Handlers are expected | |
| 9 | * not to throw — we swallow exceptions defensively so one misbehaving | |
| 10 | * subscriber cannot take down the publisher or starve its peers. | |
| 11 | * | |
| 12 | * REDIS PUB/SUB (REDIS_URL is set) | |
| 13 | * Two dedicated `Bun.RedisClient` instances — one for subscribing | |
| 14 | * (which enters pub/sub mode and can't issue normal commands) and one | |
| 15 | * for publishing. Both are lazy-connected and auto-reconnect on | |
| 16 | * disconnect via the built-in `autoReconnect` option. | |
| 17 | * | |
| 18 | * Architecture on each server instance: | |
| 19 | * • Local `topics` map tracks handlers just like the in-memory path. | |
| 20 | * • When the first handler subscribes to a topic, the Redis subscriber | |
| 21 | * client issues SUBSCRIBE for that channel. | |
| 22 | * • When the last handler unsubscribes, UNSUBSCRIBE is issued. | |
| 23 | * • Incoming Redis messages fan out to all local handlers via | |
| 24 | * `localDeliver`. | |
| 25 | * • `publish` serialises the event to JSON and calls PUBLISH on the | |
| 26 | * publisher client. It does NOT also call `localDeliver` — the | |
| 27 | * Redis message that bounces back from the broker does that, so | |
| 28 | * every instance (including the publisher) receives it exactly once | |
| 29 | * through the subscriber path. | |
| 30 | * | |
| 31 | * Error handling: Redis errors are caught and logged to stderr; they | |
| 32 | * never throw into callers. If Redis is temporarily unavailable, | |
| 33 | * `publish` silently drops the cross-instance delivery (local handlers | |
| 34 | * on the same instance still receive the event because `localDeliver` | |
| 35 | * is always called, see NOTE below). | |
| 36 | * | |
| 37 | * NOTE on publish-with-no-Redis fallback: | |
| 38 | * When REDIS_URL is set but the broker is unreachable at publish time, | |
| 39 | * we fall back to local-only delivery so SSE streams on the same process | |
| 40 | * are never broken. This matches the in-memory semantics exactly on | |
| 41 | * single-instance deploys and degrades gracefully on multi-instance ones. | |
| 42 | * | |
| 43 | * Public API (unchanged from the original in-memory implementation): | |
| 44 | * publish(topic, event) → void | |
| 45 | * subscribe(topic, cb) → () => void (cleanup / unsubscribe) | |
| 46 | * topicSubscriberCount(topic) → number | |
| febd4f0 | 47 | */ |
| 48 | ||
| 49 | export type SSEEvent = { | |
| 50 | event?: string; | |
| 51 | data: unknown; | |
| 52 | id?: string; | |
| 53 | }; | |
| 54 | ||
| 55 | type Handler = (event: SSEEvent) => void; | |
| 56 | ||
| 13cbd17 | 57 | // --------------------------------------------------------------------------- |
| 58 | // In-memory local state (shared by both modes) | |
| 59 | // --------------------------------------------------------------------------- | |
| 60 | ||
| febd4f0 | 61 | const topics = new Map<string, Set<Handler>>(); |
| 62 | ||
| 13cbd17 | 63 | /** Fan out an event to every local handler registered for `topic`. */ |
| 64 | function localDeliver(topic: string, event: SSEEvent): void { | |
| febd4f0 | 65 | const subs = topics.get(topic); |
| 66 | if (!subs || subs.size === 0) return; | |
| 67 | for (const handler of subs) { | |
| 68 | try { | |
| 69 | handler(event); | |
| 70 | } catch { | |
| 71 | // Swallow — handlers are fire-and-forget and must not disrupt fanout. | |
| 72 | } | |
| 73 | } | |
| 74 | } | |
| 75 | ||
| 13cbd17 | 76 | // --------------------------------------------------------------------------- |
| 77 | // Redis layer (only initialised when REDIS_URL is present) | |
| 78 | // --------------------------------------------------------------------------- | |
| 79 | ||
| 80 | /** True when REDIS_URL is configured and the Redis layer has been set up. */ | |
| 81 | let redisMode = false; | |
| 82 | ||
| 83 | // Bun.RedisClient instances — typed loosely to avoid issues when running in | |
| 84 | // environments where the Bun global is absent (e.g., pure Node test runners). | |
| 85 | // We access them through the `Bun` global at runtime so no import is needed. | |
| 86 | let redisPub: InstanceType<typeof Bun.RedisClient> | null = null; | |
| 87 | let redisSub: InstanceType<typeof Bun.RedisClient> | null = null; | |
| 88 | ||
| 89 | /** Redis channels (topics) the subscriber client is currently subscribed to. */ | |
| 90 | const redisSubscribed = new Set<string>(); | |
| 91 | ||
| 92 | /** Initialise the Redis pub and sub clients (called once, lazily). */ | |
| 93 | function initRedis(url: string): void { | |
| 94 | if (redisMode) return; | |
| 95 | redisMode = true; | |
| 96 | ||
| 97 | const opts = { | |
| 98 | autoReconnect: true, | |
| 99 | maxRetries: Infinity as unknown as number, | |
| 100 | enableOfflineQueue: true, | |
| 101 | }; | |
| 102 | ||
| 103 | redisPub = new Bun.RedisClient(url, opts); | |
| 104 | redisSub = new Bun.RedisClient(url, opts); | |
| 105 | ||
| 106 | // Log disconnections to stderr but do not crash. | |
| 107 | redisPub.onclose = (err: Error) => { | |
| 108 | if (err) console.error("[sse] Redis pub connection closed:", err.message); | |
| 109 | }; | |
| 110 | redisSub.onclose = (err: Error) => { | |
| 111 | if (err) console.error("[sse] Redis sub connection closed:", err.message); | |
| 112 | }; | |
| 113 | ||
| 114 | // When the subscriber reconnects it must re-subscribe to all tracked | |
| 115 | // channels because Redis discards subscriptions on disconnect. | |
| 116 | redisSub.onconnect = function (this: InstanceType<typeof Bun.RedisClient>) { | |
| 117 | for (const ch of redisSubscribed) { | |
| 118 | this.subscribe(ch, redisMessageHandler).catch((e: unknown) => { | |
| 119 | console.error("[sse] Redis re-subscribe failed for", ch, e); | |
| 120 | }); | |
| 121 | } | |
| 122 | }; | |
| 123 | } | |
| 124 | ||
| 125 | /** | |
| 126 | * Called by the Redis subscriber client for every incoming message. | |
| 127 | * The `channel` is the topic name; `message` is a JSON-encoded SSEEvent. | |
| 128 | */ | |
| 129 | function redisMessageHandler(message: string, channel: string): void { | |
| 130 | try { | |
| 131 | const event = JSON.parse(message) as SSEEvent; | |
| 132 | localDeliver(channel, event); | |
| 133 | } catch { | |
| 134 | // Malformed payload — ignore. | |
| 135 | } | |
| 136 | } | |
| 137 | ||
| 138 | /** | |
| 139 | * Ensure the Redis subscriber is subscribed to `topic`. | |
| 140 | * Called when the first local handler registers for a topic. | |
| 141 | */ | |
| 142 | function redisEnsureSubscribed(topic: string): void { | |
| 143 | if (redisSubscribed.has(topic) || !redisSub) return; | |
| 144 | redisSubscribed.add(topic); | |
| 145 | redisSub.subscribe(topic, redisMessageHandler).catch((e: unknown) => { | |
| 146 | console.error("[sse] Redis subscribe failed for", topic, e); | |
| 147 | }); | |
| 148 | } | |
| 149 | ||
| 150 | /** | |
| 151 | * Remove the Redis subscription for `topic` when the last local handler | |
| 152 | * leaves. Silently ignored if not subscribed. | |
| 153 | */ | |
| 154 | function redisEnsureUnsubscribed(topic: string): void { | |
| 155 | if (!redisSubscribed.has(topic) || !redisSub) return; | |
| 156 | redisSubscribed.delete(topic); | |
| 157 | redisSub.unsubscribe(topic).catch((e: unknown) => { | |
| 158 | console.error("[sse] Redis unsubscribe failed for", topic, e); | |
| 159 | }); | |
| 160 | } | |
| 161 | ||
| 162 | // --------------------------------------------------------------------------- | |
| 163 | // Lazy Redis initialisation guard | |
| 164 | // --------------------------------------------------------------------------- | |
| 165 | ||
| 166 | let redisInitialised = false; | |
| 167 | ||
| 168 | function maybeInitRedis(): void { | |
| 169 | if (redisInitialised) return; | |
| 170 | redisInitialised = true; | |
| 171 | const url = process.env.REDIS_URL || process.env.VALKEY_URL; | |
| 172 | if (url) { | |
| 173 | try { | |
| 174 | initRedis(url); | |
| 175 | } catch (e) { | |
| 176 | console.error("[sse] Failed to initialise Redis clients:", e); | |
| 177 | redisMode = false; | |
| 178 | } | |
| 179 | } | |
| 180 | } | |
| 181 | ||
| 182 | // Initialise eagerly at module load so reconnect logic is set up before the | |
| 183 | // first request arrives. The clients themselves are lazy-connected by Bun. | |
| 184 | maybeInitRedis(); | |
| 185 | ||
| 186 | // --------------------------------------------------------------------------- | |
| 187 | // Public API | |
| 188 | // --------------------------------------------------------------------------- | |
| 189 | ||
| 190 | /** | |
| 191 | * Publish an event to every subscriber of `topic`. | |
| 192 | * | |
| 193 | * In Redis mode the event is serialised to JSON and PUBLISH-ed to the Redis | |
| 194 | * channel; the subscriber client on every instance (including this one) | |
| 195 | * receives the message and fans it out to local handlers. If Redis is | |
| 196 | * unavailable the event is delivered locally so SSE streams on this process | |
| 197 | * are never silently broken. | |
| 198 | * | |
| 199 | * In in-memory mode subscribers are called synchronously. | |
| 200 | */ | |
| 201 | export function publish(topic: string, event: SSEEvent): void { | |
| 202 | if (redisMode && redisPub) { | |
| 203 | const payload = JSON.stringify(event); | |
| 204 | redisPub.publish(topic, payload).catch((e: unknown) => { | |
| 205 | // Redis unavailable — fall back to local delivery so in-process SSE | |
| 206 | // streams remain functional on single-instance deploys. | |
| 207 | console.error("[sse] Redis publish failed, delivering locally:", e); | |
| 208 | localDeliver(topic, event); | |
| 209 | }); | |
| 210 | // Do NOT call localDeliver here: the Redis subscriber path does it so | |
| 211 | // the publisher instance doesn't double-deliver. | |
| 212 | return; | |
| 213 | } | |
| 214 | // In-memory path. | |
| 215 | localDeliver(topic, event); | |
| 216 | } | |
| 217 | ||
| febd4f0 | 218 | /** |
| 219 | * Register a handler for `topic`. Returns a cleanup function that removes | |
| 220 | * the handler (and drops the topic's entry when its last subscriber leaves). | |
| 221 | */ | |
| 222 | export function subscribe( | |
| 223 | topic: string, | |
| 224 | handler: Handler | |
| 225 | ): () => void { | |
| 226 | let subs = topics.get(topic); | |
| 227 | if (!subs) { | |
| 228 | subs = new Set<Handler>(); | |
| 229 | topics.set(topic, subs); | |
| 230 | } | |
| 13cbd17 | 231 | const isFirst = subs.size === 0; |
| febd4f0 | 232 | subs.add(handler); |
| 233 | ||
| 13cbd17 | 234 | // Ensure the Redis subscriber is tracking this channel. |
| 235 | if (redisMode && isFirst) { | |
| 236 | redisEnsureSubscribed(topic); | |
| 237 | } | |
| 238 | ||
| febd4f0 | 239 | return () => { |
| 240 | const current = topics.get(topic); | |
| 241 | if (!current) return; | |
| 242 | current.delete(handler); | |
| 243 | if (current.size === 0) { | |
| 244 | topics.delete(topic); | |
| 13cbd17 | 245 | // Release the Redis subscription when no local handlers remain. |
| 246 | if (redisMode) { | |
| 247 | redisEnsureUnsubscribed(topic); | |
| 248 | } | |
| febd4f0 | 249 | } |
| 250 | }; | |
| 251 | } | |
| 252 | ||
| 253 | /** Number of active subscribers on a topic (0 if unknown). */ | |
| 254 | export function topicSubscriberCount(topic: string): number { | |
| 255 | return topics.get(topic)?.size ?? 0; | |
| 256 | } |