Repository navigation
feat(api): adopt redis streams for realtime - #13
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
WalkthroughAdds a Redis-backed real-time pub/sub system and a local WebSocket connection registry, wires socket dispatch to publish via Redis streams, and initializes consumers to deliver events to targeted connections/visitors/websites. Introduces REDIS_URL env var. Minor formatting in router files and test updates with pub/sub mocks. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor Client
participant WS Server as WS Server (socket.ts)
participant Pub as Redis Publisher (realtime-pubsub)
participant Stream as Redis Stream
participant Sub as Redis Consumer (realtime-pubsub)
participant Registry as Connection Registry
participant Socket as Target Socket(s)
Client->>WS Server: Send event
WS Server->>Pub: publishTo(Connection|Visitor|Website)
Pub->>Stream: XADD envelope
loop Long-poll consume
Sub->>Stream: XREAD BLOCK
Stream-->>Sub: Envelope batch
Sub->>Registry: dispatchEventTo(Local Connection/Visitor/Website)
Registry->>Socket: send(event)
Socket-->>Registry: ack/close or error
end
sequenceDiagram
autonumber
participant WS Server as WS Server (startup)
participant Sub as Redis Consumer
participant Registry as Local Dispatchers
WS Server->>Registry: build dispatchers (connection/visitor/website)
WS Server->>Sub: initializeRealtimePubSub(dispatchers)
Sub->>Sub: Load last cursor, start loop
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🧹 Nitpick comments (9)
apps/api/src/ws/connection-registry.ts (1)
31-39: Harden send path against closed sockets.Sending on a closed/closing socket will throw repeatedly. Add a readyState check and prune dead connections on failure.
Bun’s ServerWebSocket exposes a readyState-compatible API; verify constants (e.g., WebSocket.OPEN) in your runtime.
function sendEventToSocket( record: LocalConnectionRecord, serializedEvent: string ): void { try { - record.socket.send(serializedEvent); + // Avoid sending to dead sockets; prune on failure + // If Bun doesn't expose WebSocket.OPEN, compare to numeric 1. + // eslint-disable-next-line @typescript-eslint/ban-ts-comment + // @ts-ignore - tolerate env differences + if (typeof record.socket.readyState === "number" && record.socket.readyState !== 1) { + return; + } + record.socket.send(serializedEvent); } catch (error) { - console.error("[WebSocket] Failed to send event:", error); + // TODO: route to central logger; avoid console + // logger.error({ err: error }, "[WebSocket] Failed to send event"); } }apps/api/src/ws/socket.test.ts (1)
57-62: Mock publish functions with call-tracking to assert behavior.Using mock.fn() enables verifying that publish functions are invoked as expected.
-mock.module("./realtime-pubsub", () => ({ - initializeRealtimePubSub: () => {}, - publishToConnection: () => Promise.resolve(), - publishToVisitor: () => Promise.resolve(), - publishToWebsite: () => Promise.resolve(), -})); +mock.module("./realtime-pubsub", () => { + const initializeRealtimePubSub = mock.fn(() => {}); + const publishToConnection = mock.fn(() => Promise.resolve()); + const publishToVisitor = mock.fn(() => Promise.resolve()); + const publishToWebsite = mock.fn(() => Promise.resolve()); + return { + initializeRealtimePubSub, + publishToConnection, + publishToVisitor, + publishToWebsite, + }; +});Please add assertions in relevant tests to ensure the publish functions are called with the expected envelopes.
apps/api/src/ws/realtime-pubsub.ts (1)
165-201: Minor: drop unreachable default and make the switch exhaustive.You already narrow target.type; assigning to
neveris good, but the default block can be removed in favor of a compile‑time exhaustive check.- default: { - const exhaustiveCheck: never = target; - console.error( - "[RealtimePubSub] Unsupported dispatch target", - exhaustiveCheck - ); - } + // exhaustive by design; adding a new target will surface a TS error hereapps/api/src/ws/socket.ts (6)
70-106: Return‑path for publish failures and explicit fire‑and‑forgetThe wrappers intentionally fire‑and‑forget. Prefix with
voidto satisfy promise‑handling linters and includeevent.typefor observability.export const sendEventToConnection: ConnectionDispatcher = ( connectionId, event ) => { - publishToConnection(connectionId, event).catch((error) => { + void publishToConnection(connectionId, event).catch((error) => { console.error("[RealtimePubSub] Failed to publish connection event", { connectionId, + eventType: event.type, error, }); }); }; export const sendEventToVisitor: VisitorDispatcher = ( visitorId, event, options ) => { - publishToVisitor(visitorId, event, options).catch((error) => { + void publishToVisitor(visitorId, event, options).catch((error) => { console.error("[RealtimePubSub] Failed to publish visitor event", { visitorId, + eventType: event.type, error, }); }); }; export const sendEventToWebsite: WebsiteDispatcher = ( websiteId, event, options ) => { - publishToWebsite(websiteId, event, options).catch((error) => { + void publishToWebsite(websiteId, event, options).catch((error) => { console.error("[RealtimePubSub] Failed to publish website event", { websiteId, + eventType: event.type, error, }); }); };
205-205: Prefer crypto.randomUUID() for connection IDsReduces collision risk vs Math.random and shortens code.
-export function generateConnectionId(): string { - return `conn_${Date.now()}_${Math.random().toString(36).substring(2, 9)}`; -} +export function generateConnectionId(): string { + return `conn_${crypto.randomUUID()}`; +}
795-797: Use a generic error response for handler failures (not “format” errors)Routing/handler exceptions aren’t message‑format issues. Send a general error to the client.
- console.error("[WebSocket] Error processing message:", error); - sendInvalidFormatResponse(ws, error); + console.error("[WebSocket] Error processing message:", error); + sendError(ws, { + error: "Message handling failed", + message: error instanceof Error ? error.message : "Unknown error", + });
754-758: Don’t fail the open handshake on “last seen” writesPresence/timestamp updates shouldn’t break the connection setup path.
- // Update last seen timestamps - await updateLastSeenTimestamps({ db, authResult }); + // Update last seen timestamps (best‑effort) + try { + await updateLastSeenTimestamps({ db, authResult }); + } catch { + // TODO: replace with logger.debug once available + }
333-343: Replace console. with the project logger*Guidelines prohibit
console. Use the central logger with levels and structured payloads.Also applies to: 459-466, 474-485, 217-220, 233-236, 278-281, 720-722, 746-753, 795-797, 820-821
410-413: Detect protocol behind proxies
c.req.url.startsWith("wss://")can be incorrect behind a TLS terminator. PreferX-Forwarded-Proto(and fall back to request URL).- const hostname = hostHeader.split(":")[0]; - const isSecure = c.req.url.startsWith("wss://"); - const protocol = isSecure ? "wss:" : "ws:"; + const hostname = hostHeader.split(":")[0]; + const xfProto = c.req.header("X-Forwarded-Proto"); + const isSecure = xfProto ? xfProto.toLowerCase() === "https" : c.req.url.startsWith("https://") || c.req.url.startsWith("wss://"); + const protocol = isSecure ? "wss:" : "ws:";
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
apps/api/src/env.ts(1 hunks)apps/api/src/ws/connection-registry.ts(1 hunks)apps/api/src/ws/realtime-pubsub.ts(1 hunks)apps/api/src/ws/router.test.ts(1 hunks)apps/api/src/ws/router.ts(1 hunks)apps/api/src/ws/socket.test.ts(2 hunks)apps/api/src/ws/socket.ts(2 hunks)
🧰 Additional context used
📓 Path-based instructions (9)
**/*.{ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/global-rules.mdc)
**/*.{ts,tsx}: Manage shared component state via React Context
Implement full TypeScript support for all public APIs
Provide proper type definitions for all props and refs
Use appropriate generic constraints where necessary
Export type definitions for the public API
Avoid TypeScript enums; prefer const objects with 'as const' or Object.freeze
Provide clear, descriptive runtime errors for missing context/props and misuse
Expose dedicated hooks for context access that throw helpful errors when context is missing
Add explicit error messages for: outside-context usage, missing required props, invalid prop combinations, and validation failures
Document each component with JSDoc: purpose, props (required/optional), context requirements, example usage, pitfalls
State management: use React Context for global state, perform correct state updates, and handle side effects appropriately
Validation: implement runtime checks and validate prop combinations with clear errors
Create context with createContext<ContextType|null>(null) and a useContext hook that throws a clear error when missing
**/*.{ts,tsx}: Don't use primitive type aliases or misleading types
Don't use empty type parameters in type aliases and interfaces
Don't use any or unknown as type constraints
Don't have unused function parameters
Don't have unused private class members
Don't return a value from a function that has a 'void' return type
Don't use the TypeScript directive @ts-ignore
Ensure switch-case statements are exhaustive (especially with unions)
Don't use TypeScript enums
Don't add type annotations to variables, parameters, and class properties initialized with literal expressions
Don't use TypeScript namespaces
Don't use non-null assertions with the ! postfix operator
Don't use parameter properties in class constructors
Don't use user-defined types (where discouraged by policy)
Use as const instead of literal types and type annotations
Use either T[] or Array consistently
Use consistent accessibili...
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
**/*.{test.ts,test.tsx,spec.ts,spec.tsx,test.js,test.jsx,spec.js,spec.jsx}
📄 CodeRabbit inference engine (.cursor/rules/global-rules.mdc)
Include tests: unit tests for components, integration tests for composed components, accessibility tests, and error boundary tests
Files:
apps/api/src/ws/socket.test.tsapps/api/src/ws/router.test.ts
**/*.{js,jsx,ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
**/*.{js,jsx,ts,tsx}: Don't use theaccessKeyattribute on any HTML element
Use correct ISO language/country codes for the lang attribute
Don't hardcode sensitive data like API keys and tokens
Use === and !==
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
**/*.{js,ts}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
**/*.{js,ts}: Don't use consecutive spaces in regular expression literals
Don't use the arguments object
Don't use the comma operator
Don't use unnecessary boolean casts
Don't use unnecessary callbacks with flatMap
Prefer for...of statements instead of Array.forEach
Don't use this and super in static contexts
Don't use unnecessary catch clauses
Don't use unnecessary constructors
Don't use unnecessary continue statements
Don't export empty modules that don't change anything
Don't use unnecessary escape sequences in regular expression literals
Don't use unnecessary labels
Don't use unnecessary nested block statements
Don't rename imports, exports, and destructured assignments to the same name
Don't use unnecessary string or template literal concatenation
Don't use String.raw in template literals when there are no escape sequences
Don't use useless case statements in switch statements
Avoid ternary operators when simpler alternatives exist
Don't use useless this aliasing
Don't initialize variables to undefined
Don't use void operators
Use Date.now() to get milliseconds since the Unix Epoch
Prefer .flatMap() over map().flat() when possible
Use literal property access instead of computed property access
Don't use parseInt() or Number.parseInt() when numeric literals suffice (binary/octal/hex)
Use concise optional chaining instead of chained logical expressions
Prefer regular expression literals over the RegExp constructor when possible
Don't use non–base-10 or underscore-separated number literal object member names
Remove redundant terms from logical expressions
Prefer while loops over for loops when no initializer/update is needed
Don't reassign const variables
Don't use constant expressions in conditions
Don't use Math.min/Math.max to clamp values when result is constant
Don't use empty character classes in regular expression literals
Don't use empty destructuring patterns
Don't call global object properties as functions
Don't declare functions and vars that are acce...
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
**/*.{test,spec}.{js,jsx,ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
**/*.{test,spec}.{js,jsx,ts,tsx}: Don't nest describe() blocks too deeply in test files
Don't use callbacks in asynchronous tests and hooks
Don't have duplicate hooks in describe blocks
Don't use export or module.exports in test files
Don't use focused tests
Ensure the assertion function (e.g., expect) is placed inside an it() function call
Don't use disabled tests
Files:
apps/api/src/ws/socket.test.tsapps/api/src/ws/router.test.ts
**/*.{js,ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
**/*.{js,ts,tsx}: Don't create classes that only have static members
Don't return a value from a constructor
Don't use super() incorrectly inside classes; ensure super() is called in derived constructors
Ensure super() is called exactly once on every code path in a class constructor before this is accessed if the class has a superclass
Place getters and setters for the same property next to each other in class and object definitions
Ensure JSDoc comment lines start with a single asterisk, except for the first one
Don't use duplicate class members
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
**/*.{js,tsx,ts,jsx}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
Use arrow functions instead of function expressions
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
**/*.{js,ts,jsx,tsx}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
**/*.{js,ts,jsx,tsx}: Prevent duplicate polyfills from Polyfill.io
Don't use irregular whitespace characters
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
**/*.{js,ts,ts,tsx}
📄 CodeRabbit inference engine (.cursor/rules/ultracite.mdc)
Don't redeclare variables, functions, classes, and types in the same scope
Files:
apps/api/src/ws/socket.test.tsapps/api/src/env.tsapps/api/src/ws/realtime-pubsub.tsapps/api/src/ws/router.tsapps/api/src/ws/router.test.tsapps/api/src/ws/connection-registry.tsapps/api/src/ws/socket.ts
🧠 Learnings (1)
📚 Learning: 2025-09-17T09:01:10.551Z
Learnt from: CR
PR: cossistantcom/cossistant#0
File: .cursor/rules/global-rules.mdc:0-0
Timestamp: 2025-09-17T09:01:10.551Z
Learning: Applies to **/*.{ts,tsx} : Export type definitions for the public API
Applied to files:
apps/api/src/ws/router.ts
🧬 Code graph analysis (4)
apps/api/src/ws/socket.test.ts (2)
packages/types/src/realtime-events.ts (1)
RealtimeEvent(60-64)apps/api/src/ws/router.ts (1)
EventContext(183-183)
apps/api/src/ws/realtime-pubsub.ts (3)
packages/types/src/realtime-events.ts (3)
RealtimeEvent(60-64)isValidEventType(84-86)validateRealtimeEvent(73-79)apps/api/src/ws/router.ts (1)
DispatchOptions(182-182)apps/api/src/env.ts (1)
env(27-58)
apps/api/src/ws/connection-registry.ts (1)
packages/types/src/realtime-events.ts (1)
RealtimeEvent(60-64)
apps/api/src/ws/socket.ts (7)
apps/api/src/db/queries/session.ts (3)
ApiKeyWithWebsiteAndOrganization(12-15)normalizeSessionToken(19-36)resolveSession(38-87)apps/api/src/ws/realtime-pubsub.ts (4)
initializeRealtimePubSub(342-346)publishToConnection(348-359)publishToVisitor(361-378)publishToWebsite(380-397)apps/api/src/ws/connection-registry.ts (6)
dispatchEventToLocalConnection(42-53)dispatchEventToLocalVisitor(55-79)dispatchEventToLocalWebsite(81-110)RawSocket(5-5)LocalConnectionRecord(7-13)localConnections(15-15)apps/api/src/ws/router.ts (2)
EventContext(183-183)routeEvent(160-178)apps/api/src/utils/websocket-connection.ts (6)
handleAuthenticationFailure(27-37)handleIdentificationFailure(39-49)storeConnectionId(51-56)sendConnectionEstablishedMessage(58-76)createConnectionEvent(78-113)updatePresenceIfNeeded(115-132)packages/types/src/realtime-events.ts (3)
RealtimeEvent(60-64)isValidEventType(84-86)validateRealtimeEvent(73-79)apps/api/src/lib/auth-validation.ts (3)
AuthValidationOptions(23-27)AuthValidationError(34-42)performAuthentication(382-431)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Cursor Bugbot
🔇 Additional comments (3)
apps/api/src/ws/router.test.ts (1)
4-4: LGTM: import reorder is fine.No behavior change; keeps types-first import style tidy.
apps/api/src/ws/router.ts (1)
180-187: Good: explicit type exports for the public router API.This aligns with “export type definitions for the public API” guidance.
If consumers import these types from barrel files, ensure re-exports are updated to avoid deep import paths.
apps/api/src/ws/socket.test.ts (1)
6-6: LGTM: tuple array typing is clearer.The
[RealtimeEvent, EventContext][]form is concise and accurate.
| UPSTASH_REDIS_REST_TOKEN: getEnvVariable( | ||
| "UPSTASH_REDIS_REST_TOKEN", | ||
| "example_token" | ||
| ), |
There was a problem hiding this comment.
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.
| "UPSTASH_REDIS_REST_TOKEN", | ||
| "example_token" | ||
| ), | ||
| REDIS_URL: getEnvVariable("REDIS_URL", "redis://localhost:6379"), |
There was a problem hiding this comment.
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.
| console.log("[WebSocket] Dispatching visitor event", { | ||
| visitorId, | ||
| connectionId, | ||
| eventType: event.type, | ||
| }); | ||
| sendEventToSocket(connection, serializedEvent); |
There was a problem hiding this comment.
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.
| const instanceId = `api-${process.pid ?? "pid"}-${Math.random() | ||
| .toString(36) | ||
| .slice(2, 10)}`; | ||
|
|
There was a problem hiding this comment.
Cursor persistence won’t survive process restarts; use a stable key or groups.
Cursor key includes a random instanceId, so a new process won’t load the previous cursor and will start at "$", potentially skipping backlog events.
Option A (simple): use a stable consumer ID from env; fall back to a deterministic value.
-const instanceId = `api-${process.pid ?? "pid"}-${Math.random()
- .toString(36)
- .slice(2, 10)}`;
+const instanceIdBase =
+ process.env.REALTIME_CONSUMER_ID ||
+ process.env.VERCEL_DEPLOYMENT_ID ||
+ process.env.HOSTNAME ||
+ "api";
+const instanceId = `${instanceIdBase}-${process.pid ?? "pid"}`;-const instanceCursorKey = `${CURSOR_KEY_PREFIX}${instanceId}`;
+// Persist a stable cursor across restarts of the same deployment
+const instanceCursorKey = `${CURSOR_KEY_PREFIX}${instanceIdBase}`;Option B (robust): switch to consumer groups (XGROUP/XREADGROUP) with a shared group and per‑instance consumer; ack each entry and use XCLAIM on failover.
Also applies to: 75-83, 141-155
🤖 Prompt for AI Agents
In apps/api/src/ws/realtime-pubsub.ts around lines 20-23 (also review similar
logic at 75-83 and 141-155): the cursor/key currently embeds a random instanceId
so restarts create new keys and lose cursor position; replace the random
instanceId with a stable consumer identifier (read from env like
PROCESS_CONSUMER_ID or generate a deterministic fallback based on hostname/pid)
to persist and reload the cursor across restarts, or preferably migrate to Redis
Streams consumer groups (create a shared XGROUP, use XREADGROUP per-instance
consumers, ACK entries and use XCLAIM on failover) so each instance is a
consumer within a durable group and cursor/state is maintained server-side.
| const redisOptions: RedisOptions = { | ||
| lazyConnect: true, | ||
| maxRetriesPerRequest: null, | ||
| enableAutoPipelining: true, | ||
| reconnectOnError: (error) => { | ||
| console.error("[RealtimeRedis] Connection error", error); | ||
| return true; | ||
| }, | ||
| retryStrategy: (attempt) => Math.min(1000 * attempt, 5000), | ||
| }; |
There was a problem hiding this comment.
Replace console with centralized logger.*
Multiple console.* calls violate guidelines. Switch to structured logging with consistent fields (component, streamKey, id, target, event.type).
No inline diff provided to avoid guessing your logger; if none exists, I can generate a minimal logger wrapper and apply it project‑wide.
Also applies to: 107-115, 117-139, 157-227, 259-304, 321-339
🤖 Prompt for AI Agents
In apps/api/src/ws/realtime-pubsub.ts around lines 24-33 (and also apply the
same change to ranges 107-115, 117-139, 157-227, 259-304, 321-339), replace all
console.* calls with the project’s centralized logger and emit structured logs
containing at minimum the fields: component: "RealtimeRedis" (or
"realtime-pubsub"), streamKey, id, target, and event.type where available; for
the reconnectOnError handler capture and log the error object via logger.error
including these fields, and for retry/other informational messages use
logger.info or logger.warn with the same structured context. Ensure you import
the existing logger (or use the project’s logger wrapper), remove console usage,
and preserve original message text while passing the error object separately to
keep stack/details intact.
| initializeRealtimePubSub({ | ||
| connection: dispatchEventToLocalConnection, | ||
| visitor: dispatchEventToLocalVisitor, | ||
| website: dispatchEventToLocalWebsite, | ||
| }); |
There was a problem hiding this comment.
Make Redis consumer initialization idempotent
Top‑level initializeRealtimePubSub(...) will start a consumer loop every time this module is evaluated (hot reloads, multiple workers). Add a guard to avoid duplicate consumers.
+let __pubsubInitialized = false;
+
- initializeRealtimePubSub({
- connection: dispatchEventToLocalConnection,
- visitor: dispatchEventToLocalVisitor,
- website: dispatchEventToLocalWebsite,
- });
+if (!__pubsubInitialized) {
+ initializeRealtimePubSub({
+ connection: dispatchEventToLocalConnection,
+ visitor: dispatchEventToLocalVisitor,
+ website: dispatchEventToLocalWebsite,
+ });
+ __pubsubInitialized = true;
+}Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In apps/api/src/ws/socket.ts around lines 64 to 68, calling
initializeRealtimePubSub at module top-level can spawn duplicate Redis consumers
on hot reloads or multiple workers; make initialization idempotent by adding a
guard flag on a global (e.g., globalThis.__realtimePubSubInitialized) or
module-scoped singleton check, only call initializeRealtimePubSub when the flag
is not set, and set the flag immediately after successful initialization so
subsequent evaluations skip creating another consumer.
| type SocketContext = { | ||
| raw?: RawSocket; | ||
| send: (data: string) => void; | ||
| }; |
There was a problem hiding this comment.
🧩 Analysis chain
Verify WSContext compatibility of SocketContext in helpers
getConnectionIdFromSocket/sendError are imported from utils typed against WSContext. Here we pass a minimal SocketContext. Please confirm structural type compatibility to avoid subtle type erasure or runtime assumptions in helpers.
Run:
Also applies to: 118-134
🏁 Script executed:
#!/bin/bash
# Check the expected parameter types in websocket-connection helpers
rg -nP -C2 'export\s+function\s+(getConnectionIdFromSocket|sendError)\s*\(' apps/api/src/utils/websocket-connection.tsLength of output: 323
🏁 Script executed:
#!/usr/bin/env bash
# Locate WSContext definition in websocket-connection utilities
rg -nP "export (?:type|interface) WSContext" -n apps/api/src/utils/websocket-connection.ts
rg -nP "type WSContext" -n apps/api/src/utils/websocket-connection.tsLength of output: 21
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "---- apps/api/src/utils/websocket-connection.ts (lines 1-220) ----"
nl -ba apps/api/src/utils/websocket-connection.ts | sed -n '1,220p'
echo
echo "---- apps/api/src/ws/socket.ts (lines 1-220) ----"
nl -ba apps/api/src/ws/socket.ts | sed -n '1,220p'Length of output: 107
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "---- apps/api/src/utils/websocket-connection.ts (lines 1-220) ----"
sed -n '1,220p' apps/api/src/utils/websocket-connection.ts || true
echo
echo "---- apps/api/src/ws/socket.ts (lines 1-240) ----"
sed -n '1,240p' apps/api/src/ws/socket.ts || true
echo
echo "---- Search for RawSocket / WSContext / ServerWebSocket occurrences in repo ----"
rg -nP --hidden --no-ignore 'RawSocket|WSContext|ServerWebSocket' || trueLength of output: 13260
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "---- socket.ts lines 640-920 ----"
sed -n '640,920p' apps/api/src/ws/socket.ts || true
echo
echo "---- occurrences of handleAuthenticationFailure & handleIdentificationFailure in repo ----"
rg -n --hidden --no-ignore --line-number 'handleAuthenticationFailure|handleIdentificationFailure' || true
echo
echo "---- occurrences of websocket-connection helpers in socket.ts ----"
rg -n --hidden --no-ignore --line-number 'getConnectionIdFromSocket|sendError|storeConnectionId|sendConnectionEstablishedMessage|updatePresenceIfNeeded|createConnectionEvent' apps/api/src/ws/socket.ts || trueLength of output: 6758
Make SocketContext structurally compatible with WSContext
WSContext in apps/api/src/utils/websocket-connection.ts (type at line ~15) requires:
- send: (data: string | ArrayBuffer) => void
- close: (code?: number, reason?: string) => void
- raw?: RawSocket
SocketContext in apps/api/src/ws/socket.ts (lines ~108–111) is:
- raw?: RawSocket
- send: (data: string) => void
This is not structurally compatible and will cause type errors when calling getConnectionIdFromSocket/sendError.
- Fix: export/reuse WSContext or update SocketContext to match (add close and use send: (data: string | ArrayBuffer) => void).
- Temporary: cast at call sites (ws as unknown as WSContext) only while aligning types.
🤖 Prompt for AI Agents
In apps/api/src/ws/socket.ts around lines 108 to 111, SocketContext is missing
the close method and has a narrower send signature than WSContext; update
SocketContext to match WSContext by changing send to (data: string |
ArrayBuffer) => void and adding close: (code?: number, reason?: string) => void,
or better yet export/reuse the WSContext type from
apps/api/src/utils/websocket-connection.ts to ensure structural compatibility;
if you need a quick workaround while types are aligned, add a cast at call sites
(ws as unknown as WSContext).
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
This branch was successfully deployed
1 participant
Summary
Testing
https://chatgpt.com/codex/tasks/task_e_68cdbe673b0c832bae5a672b89926ecd
Summary by CodeRabbit