Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
Commit13cbd17unknown_key

Add Redis SSE fan-out for multi-instance deploys — falls back to in-memory

Add Redis SSE fan-out for multi-instance deploys — falls back to in-memory

When REDIS_URL (or VALKEY_URL) is set, src/lib/sse.ts uses two
Bun.RedisClient instances — a dedicated subscriber and a publisher —
so SSE events (live comments, PR live view, push watch) reach all
Fly.io instances behind the load balancer rather than just the process
that owns the connection.

Architecture:
- Subscriber client enters pub/sub mode and routes Redis channel
  messages into the existing local Map<topic,Set<Handler>> fan-out.
- Publisher client sends PUBLISH; the bounced-back message does local
  delivery so no double-deliver occurs.
- Redis subscriptions are lazily acquired per-topic (SUBSCRIBE on first
  local handler, UNSUBSCRIBE on last) and re-issued automatically on
  reconnect via the onconnect hook.
- If Redis is unavailable at publish time, the event falls back to
  local-only delivery so in-process SSE streams never break.
- When REDIS_URL is unset the original in-memory path is used unchanged.

Also adds config.redisUrl getter to src/lib/config.ts.

Public API (publish / subscribe / topicSubscriberCount) is unchanged;
no callers required modification.  Zero new npm dependencies — uses
Bun.RedisClient which ships built-in with Bun 1.x.

https://claude.ai/code/session_01DzJMTFASjMHt2f5ze4cNLR
Claude committed on June 6, 2026Parent: 30639af
2 files changed+210−1613cbd1768287cbf5c1d22a7de96f59240e26ca83
2 changed files+210−16
Modifiedsrc/lib/config.ts+9−0View fileUnifiedSplit
102102 get webauthnRpName() {
103103 return process.env.WEBAUTHN_RP_NAME || "gluecron";
104104 },
105 /**
106 * Redis / Valkey connection URL for cross-instance SSE fan-out.
107 * When set, `src/lib/sse.ts` uses Redis pub/sub so SSE events reach all
108 * server instances behind the load balancer. Falls back to in-process
109 * delivery when unset.
110 */
111 get redisUrl() {
112 return process.env.REDIS_URL || process.env.VALKEY_URL || "";
113 },
105114};
Modifiedsrc/lib/sse.ts+201−16View fileUnifiedSplit
11/**
2 * In-process topic-based pub/sub broadcaster for Server-Sent Events.
2 * Topic-based pub/sub broadcaster for Server-Sent Events.
33 *
4 * Module-level `Map<topic, Set<handler>>`. `publish` iterates subscribers
5 * synchronously (fire-and-forget). Handlers are expected not to throw — we
6 * swallow exceptions defensively so one misbehaving subscriber cannot take
7 * down the publisher or starve its peers.
4 * Two modes, selected at startup:
85 *
9 * TODO(scale): this is deliberately single-process / in-memory. Horizontally
10 * scaled deploys (multiple Bun instances behind a load balancer) will need
11 * a cross-node fanout layer — likely Redis pub/sub or NATS — that feeds this
12 * local broadcaster on each node. Until then, SSE subscribers only receive
13 * events published by the same process handling their connection.
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
1447 */
1548
1649export type SSEEvent = {
2154
2255type Handler = (event: SSEEvent) => void;
2356
57// ---------------------------------------------------------------------------
58// In-memory local state (shared by both modes)
59// ---------------------------------------------------------------------------
60
2461const topics = new Map<string, Set<Handler>>();
2562
26/**
27 * Publish an event to every subscriber of `topic`. No-op if the topic has
28 * no subscribers. Handler exceptions are caught and swallowed so a single
29 * broken subscriber cannot break fanout for its peers.
30 */
31export function publish(topic: string, event: SSEEvent): void {
63/** Fan out an event to every local handler registered for `topic`. */
64function localDeliver(topic: string, event: SSEEvent): void {
3265 const subs = topics.get(topic);
3366 if (!subs || subs.size === 0) return;
3467 for (const handler of subs) {
4073 }
4174}
4275
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. */
81let 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.
86let redisPub: InstanceType<typeof Bun.RedisClient> | null = null;
87let redisSub: InstanceType<typeof Bun.RedisClient> | null = null;
88
89/** Redis channels (topics) the subscriber client is currently subscribed to. */
90const redisSubscribed = new Set<string>();
91
92/** Initialise the Redis pub and sub clients (called once, lazily). */
93function 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 */
129function 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 */
142function 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 */
154function 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
166let redisInitialised = false;
167
168function 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.
184maybeInitRedis();
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 */
201export 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
43218/**
44219 * Register a handler for `topic`. Returns a cleanup function that removes
45220 * the handler (and drops the topic's entry when its last subscriber leaves).
53228 subs = new Set<Handler>();
54229 topics.set(topic, subs);
55230 }
231 const isFirst = subs.size === 0;
56232 subs.add(handler);
57233
234 // Ensure the Redis subscriber is tracking this channel.
235 if (redisMode && isFirst) {
236 redisEnsureSubscribed(topic);
237 }
238
58239 return () => {
59240 const current = topics.get(topic);
60241 if (!current) return;
61242 current.delete(handler);
62243 if (current.size === 0) {
63244 topics.delete(topic);
245 // Release the Redis subscription when no local handlers remain.
246 if (redisMode) {
247 redisEnsureUnsubscribed(topic);
248 }
64249 }
65250 };
66251}
67252