Skip to content

feat(api): adopt redis streams for realtime #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Rieranthony merged 3 commits into from
Sep 20, 2025
Merged

feat(api): adopt redis streams for realtime #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7c635ef
feat(api): adopt redis streams for realtime
Rieranthony Sep 20, 2025
462bbb9
Update apps/api/src/ws/realtime-pubsub.ts
Rieranthony Sep 20, 2025
000037e
Update apps/api/src/ws/socket.ts
Rieranthony Sep 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions apps/api/src/env.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,11 @@ export const env = {
"UPSTASH_REDIS_REST_URL",
"http://localhost:8079"
),
UPSTASH_REDIS_REST_TOKEN: getEnvVariable(
"UPSTASH_REDIS_REST_TOKEN",
"example_token"
),
UPSTASH_REDIS_REST_TOKEN: getEnvVariable(
"UPSTASH_REDIS_REST_TOKEN",
"example_token"
),
Comment on lines +38 to +41

@coderabbitai coderabbitai Bot Sep 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Don’t default sensitive tokens; fail fast instead.

Defaulting UPSTASH token to "example_token" can mask prod misconfig. Return empty and validate at startup.

-        UPSTASH_REDIS_REST_TOKEN: getEnvVariable(
-                "UPSTASH_REDIS_REST_TOKEN",
-                "example_token"
-        ),
+        UPSTASH_REDIS_REST_TOKEN: getEnvVariable("UPSTASH_REDIS_REST_TOKEN", ""),

Add a one-time validation (e.g., in server bootstrap):

// env-validate.ts
import { env } from "@api/env";
if (env.NODE_ENV === "production") {
  if (!env.UPSTASH_REDIS_REST_TOKEN) {
    throw new Error("UPSTASH_REDIS_REST_TOKEN is required in production");
  }
}
🤖 Prompt for AI Agents 
In apps/api/src/env.ts around lines 38-41, stop defaulting
UPSTASH_REDIS_REST_TOKEN to "example_token" — remove the hardcoded default so
the env var is returned empty/undefined when not set; then add a one-time
startup validation (e.g., in server bootstrap or a new env-validate module) that
throws an error when NODE_ENV === "production" and UPSTASH_REDIS_REST_TOKEN is
not present, causing the process to fail fast with a clear message.

REDIS_URL: getEnvVariable("REDIS_URL", "redis://localhost:6379"),

@coderabbitai coderabbitai Bot Sep 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Make REDIS_URL required in production (no silent localhost).

Avoid connecting to local Redis in prod when unset.

-        REDIS_URL: getEnvVariable("REDIS_URL", "redis://localhost:6379"),
+        REDIS_URL: getEnvVariable("REDIS_URL", process.env.NODE_ENV === "production" ? "" : "redis://localhost:6379"),

And validate on bootstrap:

import { env } from "@api/env";
if (env.NODE_ENV === "production" && !env.REDIS_URL) {
  throw new Error("REDIS_URL is required in production");
}
🤖 Prompt for AI Agents 
In apps/api/src/env.ts around line 42, the REDIS_URL currently falls back to
"redis://localhost:6379" which allows silent local connections in production;
remove the localhost default so REDIS_URL is read as required (call
getEnvVariable without the localhost default or mark it required), and add a
bootstrap validation on startup (e.g., after importing env) that throws an error
if NODE_ENV === "production" and env.REDIS_URL is falsy to fail fast in prod
when REDIS_URL is not provided.

BETTER_AUTH_URL: getEnvVariable("BETTER_AUTH_URL"),
BETTER_AUTH_SECRET: getEnvVariable("BETTER_AUTH_SECRET"),
API_KEY_SECRET: getEnvVariable("API_KEY_SECRET"),
Expand Down
110 changes: 110 additions & 0 deletions apps/api/src/ws/connection-registry.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
import type { RealtimeEvent } from "@cossistant/types/realtime-events";
import type { ServerWebSocket } from "bun";
import type { DispatchOptions } from "./router";

export type RawSocket = ServerWebSocket & { connectionId?: string };

export type LocalConnectionRecord = {
socket: RawSocket;
websiteId?: string;
organizationId?: string;
userId?: string;
visitorId?: string;
};

export const localConnections = new Map<string, LocalConnectionRecord>();

function createExcludePredicate(
options?: DispatchOptions
): ((connectionId: string) => boolean) | undefined {
if (!options?.exclude) {
return;
}

const excludeIds = Array.isArray(options.exclude)
? new Set(options.exclude)
: new Set<string>([options.exclude]);

return (connectionId: string) => excludeIds.has(connectionId);
}

function sendEventToSocket(
record: LocalConnectionRecord,
serializedEvent: string
): void {
try {
record.socket.send(serializedEvent);
} catch (error) {
console.error("[WebSocket] Failed to send event:", error);
}
}

export function dispatchEventToLocalConnection(
connectionId: string,
event: RealtimeEvent
): void {
const connection = localConnections.get(connectionId);
if (!connection) {
return;
}

const serializedEvent = JSON.stringify(event);
sendEventToSocket(connection, serializedEvent);
}

export function dispatchEventToLocalVisitor(
visitorId: string,
event: RealtimeEvent,
options?: DispatchOptions
): void {
const shouldExclude = createExcludePredicate(options);
const serializedEvent = JSON.stringify(event);

for (const [connectionId, connection] of localConnections) {
if (connection.visitorId !== visitorId) {
continue;
}

if (shouldExclude?.(connectionId)) {
continue;
}

console.log("[WebSocket] Dispatching visitor event", {
visitorId,
connectionId,
eventType: event.type,
});
sendEventToSocket(connection, serializedEvent);
Comment on lines +72 to +77

@coderabbitai coderabbitai Bot Sep 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Replace console logging with centralized logger.

Guidelines prohibit console*. Use a structured logger (e.g., pino) or existing project logger.

-    console.log("[WebSocket] Dispatching visitor event", {
+    // logger.info({ visitorId, connectionId, eventType: event.type }, "Dispatching visitor event");
       visitorId,
       connectionId,
       eventType: event.type,
     });
-    console.log("[WebSocket] Dispatching website event", {
+    // logger.info({ websiteId, connectionId, eventType: event.type }, "Dispatching website event");
       websiteId,
       connectionId,
       eventType: event.type,
     });

Also applies to: 103-107

🤖 Prompt for AI Agents 
In apps/api/src/ws/connection-registry.ts around lines 72-77 (and also update
similar usage at 103-107), replace the ad-hoc console.log calls with the
project's centralized structured logger: import the existing logger (e.g., from
the project's logger module), call the appropriate log level (info/debug) on
that logger, and pass the same structured payload (visitorId, connectionId,
eventType and any context) as metadata rather than using console; remove the
console.* lines and ensure TypeScript types and imports compile after swapping
to the project logger.

}
}

export function dispatchEventToLocalWebsite(
websiteId: string,
event: RealtimeEvent,
options?: DispatchOptions
): void {
const shouldExclude = createExcludePredicate(options);
const serializedEvent = JSON.stringify(event);

for (const [connectionId, connection] of localConnections) {
if (connection.websiteId !== websiteId) {
continue;
}

// Only dashboard/user connections should receive website events
if (!connection.userId) {
continue;
}

if (shouldExclude?.(connectionId)) {
continue;
}

console.log("[WebSocket] Dispatching website event", {
websiteId,
connectionId,
eventType: event.type,
});
sendEventToSocket(connection, serializedEvent);
}
}
Loading
Loading