feat(realtime): add dedicated Redis endpoint

.env.example

@@ -91,7 +91,9 @@ DATABASE_URL=postgresql://chatbotx:secretkey@localhost:5432/chatbotx?schema=publ
 # Redis
 # ─────────────────────────────────────────────
 REDIS_URL=redis://localhost:6379
-# Durable realtime stream records use REDIS_URL directly.
+# Optional dedicated Redis for realtime streams and revoke markers. Set this
+# identically on builder, worker, and realtime; it falls back to REDIS_URL.
+# REDIS_REALTIME_URL=redis://redis-realtime:6379
 
 # Stored media
 # Re-encode received photos (JPEG) and contact avatars (WebP) to cut
@@ -235,8 +237,8 @@ REALTIME_BROADCAST_SECRET=insecure-default-realtime-broadcast-secret-change-me
 # re-checked membership/permissions) — see docs/realtime.md's close-code table.
 # REALTIME_CONNECTION_LIFETIME_MS=1800000
 
-# Redis Streams connection used by the realtime gateway and event publishers —
-# reuses the REDIS_URL set above; there is no separate env var for it.
+# Realtime publishers and the gateway use REDIS_REALTIME_URL when it is set.
+# Switching it drops up to five minutes of replay history and active revoke markers.
 #
 # The gateway reports agent presence to the builder with the shared secret above.
 # See docs/realtime.md for stream retention, replay, and close-code semantics.

apps/realtime/__tests__/gateway-config.test.ts

@@ -0,0 +1,34 @@
+import { describe, expect, test } from "vitest"
+import { resolveRealtimeGatewayConfig } from "../src/gateway-config"
+
+const gatewayEnvironment: NodeJS.ProcessEnv = {
+  NEXT_PUBLIC_BUILDER_URL: "http://builder.test",
+  REALTIME_BROADCAST_SECRET: "gateway-test-secretgateway-test-secret",
+  REDIS_URL: "redis://localhost:6379",
+}
+
+describe("resolveRealtimeGatewayConfig", () => {
+  test("uses REDIS_URL when REDIS_REALTIME_URL is unset", () => {
+    expect(resolveRealtimeGatewayConfig(gatewayEnvironment).redisUrl).toBe(
+      "redis://localhost:6379",
+    )
+  })
+
+  test("uses REDIS_REALTIME_URL when it is set", () => {
+    expect(
+      resolveRealtimeGatewayConfig({
+        ...gatewayEnvironment,
+        REDIS_REALTIME_URL: "redis://localhost:6380",
+      }).redisUrl,
+    ).toBe("redis://localhost:6380")
+  })
+
+  test("rejects an invalid REDIS_REALTIME_URL", () => {
+    expect(() =>
+      resolveRealtimeGatewayConfig({
+        ...gatewayEnvironment,
+        REDIS_REALTIME_URL: "not-a-url",
+      }),
+    ).toThrow()
+  })
+})

apps/realtime/src/gateway-config.ts

@@ -38,6 +38,7 @@ const gatewayEnv = (runtimeEnv: NodeJS.ProcessEnv) =>
         .default(1000),
       REALTIME_SHUTDOWN_DRAIN_MS: z.coerce.number().int().min(0).default(5000),
       NEXT_PUBLIC_BUILDER_URL: z.url(),
+      REDIS_REALTIME_URL: z.url().optional(),
       REDIS_URL: z.url(),
     },
     runtimeEnv,
@@ -82,7 +83,7 @@ export const resolveRealtimeGatewayConfig = (
     port: environment.PORT,
     presenceReportCoalesceMs: PRESENCE_REPORT_COALESCE_MS,
     presenceReportIntervalMs: PRESENCE_REPORT_INTERVAL_MS,
-    redisUrl: environment.REDIS_URL,
+    redisUrl: environment.REDIS_REALTIME_URL ?? environment.REDIS_URL,
     shutdownDrainMs: environment.REALTIME_SHUTDOWN_DRAIN_MS,
     secret: environment.REALTIME_BROADCAST_SECRET,
   }

docs/realtime.md

@@ -47,6 +47,12 @@ REALTIME_HOST=0.0.0.0
 REDIS_URL=redis://localhost:6379
 ```
 
+To isolate realtime streams and member-revoke markers, set
+`REDIS_REALTIME_URL` identically on Builder, Worker, and the realtime gateway.
+It falls back to `REDIS_URL`; setting it on only producers or only the gateway
+silently splits the stream. A cutover drops up to five minutes of replay history
+and active revoke markers.
+
 Start Redis and Builder with Docker, then run `pnpm --filter realtime dev`.
 Use `REALTIME_HOST`, not `HOSTNAME`: Docker owns `HOSTNAME`.
 

docs/scaling-messaging.md

@@ -14,6 +14,7 @@ retention-formula worked example.
 | Sequence scheduler | `REDIS_SEQUENCE_URL` → `REDIS_URL` | `redis:8-alpine` or Valkey 8+ | zset + Redlock coordination across 256 hash buckets. |
 | Cache / distributed lock / MAC bloom filter | `REDIS_CACHE_URL` → `REDIS_URL` | `redis:8-alpine`, or Dragonfly, or Valkey **with** `valkey-bloom` loaded | `packages/redis/src/bloom-filter.ts` issues `BF.RESERVE`/`BF.ADD`. Plain `valkey/valkey` rejects these commands and silently breaks MAC counting. |
 | Event streams (`events:message`, `events:analytics-dashboard`, `events:error-log`, `flow:events`) | `REDIS_QUEUE_URL` → `REDIS_URL` (shares hot) | `redis:8-alpine` or Valkey 8+ until tier-2 | Redis Streams with consumer groups; ceiling is `MAXLEN` retention, not throughput. |
+| Realtime streams + revoke markers | `REDIS_REALTIME_URL` → `REDIS_URL` | `redis:8-alpine` or Valkey 8+ | Streams without consumer groups; capped at five-minute retention, so an eviction-tolerant configuration is acceptable. |
 
 The `sequenceScheduler` BullMQ queue always uses `sequenceConnections`
 (`packages/redis/src/connections/sequence-connection.ts`) — it is never affected by

packages/business/__tests__/realtime-redis-url.test.ts

@@ -0,0 +1,33 @@
+import { afterEach, describe, expect, test, vi } from "vitest"
+import { resolveRealtimeRedisUrl } from "../src/platform/settings"
+
+const REDIS_URL = "redis://localhost:6379"
+const REDIS_REALTIME_URL = "redis://localhost:6380"
+
+afterEach(() => {
+  vi.unstubAllEnvs()
+})
+
+describe("resolveRealtimeRedisUrl", () => {
+  test("uses REDIS_URL when REDIS_REALTIME_URL is unset", () => {
+    vi.stubEnv("REDIS_URL", REDIS_URL)
+    vi.stubEnv(
+      "REALTIME_BROADCAST_SECRET",
+      "test-secret-test-secret-test-secret",
+    )
+    vi.stubEnv("REDIS_REALTIME_URL", undefined)
+
+    expect(resolveRealtimeRedisUrl()).toBe(REDIS_URL)
+  })
+
+  test("uses REDIS_REALTIME_URL when it is set", () => {
+    vi.stubEnv("REDIS_URL", REDIS_URL)
+    vi.stubEnv(
+      "REALTIME_BROADCAST_SECRET",
+      "test-secret-test-secret-test-secret",
+    )
+    vi.stubEnv("REDIS_REALTIME_URL", REDIS_REALTIME_URL)
+
+    expect(resolveRealtimeRedisUrl()).toBe(REDIS_REALTIME_URL)
+  })
+})

packages/business/src/integration-context/keys.ts

@@ -11,6 +11,7 @@ export const integrationContextEnv = () =>
       NEXT_PUBLIC_STORAGE_URL: z.url().optional(),
       FORCE_PUBLIC_HTTPS: z.stringbool().optional().default(false),
       REALTIME_BROADCAST_SECRET: z.string().min(32),
+      REDIS_REALTIME_URL: z.url().optional(),
       REDIS_URL: z.url(),
     },
     runtimeEnv: process.env,

packages/business/src/platform/settings.ts

@@ -221,9 +221,14 @@ export const resolveWorkspaceAppUrl = async (args: {
 export const resolveBroadcastSecret = (): string =>
   integrationContextEnv().REALTIME_BROADCAST_SECRET
 
-/** Resolve the validated Redis endpoint used by the realtime stream publisher. */
-export const resolveRealtimeRedisUrl = (): string =>
-  integrationContextEnv().REDIS_URL
+/**
+ * Resolve the validated Redis endpoint used by realtime publishers, preferring
+ * the optional dedicated REDIS_REALTIME_URL.
+ */
+export const resolveRealtimeRedisUrl = (): string => {
+  const environment = integrationContextEnv()
+  return environment.REDIS_REALTIME_URL ?? environment.REDIS_URL
+}
 
 /**
  * Resolve tenant settings for a reseller/owner directly, without a request