feat: add redis stream support for realtime messaging

apps/api/src/index.ts

@@ -20,6 +20,7 @@ import { db } from "./db";
 import { getTRPCSession } from "./db/queries/session";
 import { polarRouters } from "./polar";
 import { realtimeEmitter } from "./realtime/emitter";
+import { registerRealtimeRoutes } from "./realtime/routes";
 import { workflowsRouters } from "./workflows";
 import { upgradedWebsocket, websocket } from "./ws/socket";
 
@@ -139,6 +140,8 @@ app.route("/polar", polarRouters);
 
 app.route("/workflow", workflowsRouters);
 
+registerRealtimeRoutes(app);
+
 // WebSocket endpoint for real-time communication with rate limiting
 app.use("/ws", websocketRateLimiter);
 app.get("/ws", upgradedWebsocket);

apps/api/src/realtime/emitter.ts

@@ -1,9 +1,10 @@
 import { type EventContext, routeEvent } from "@api/ws/router";
 import {
-	sendEventToConnection,
-	sendEventToVisitor,
-	sendEventToWebsite,
+        sendEventToConnection,
+        sendEventToVisitor,
+        sendEventToWebsite,
 } from "@api/ws/socket";
+import { forwardMessageEventToRealtimeStreams } from "./stream-dispatcher";
 import {
 	type RealtimeEvent,
 	type RealtimeEventData,
@@ -126,11 +127,11 @@ export class RealtimeEmitter {
 			);
 		}
 
-		const event: RealtimeEvent<TType> = {
-			type,
-			payload: data,
-			timestamp: options.timestamp ?? Date.now(),
-			websiteId,
+                const event: RealtimeEvent<TType> = {
+                        type,
+                        payload: data,
+                        timestamp: options.timestamp ?? Date.now(),
+                        websiteId,
 			organizationId,
 			visitorId: options.visitorId ?? extractVisitorId(data) ?? null,
 		};
@@ -142,12 +143,18 @@ export class RealtimeEmitter {
 			userId: options.userId ?? extractUserId(data) ?? undefined,
 			organizationId,
 			sendToConnection: sendEventToConnection,
-			sendToVisitor: sendEventToVisitor,
-			sendToWebsite: sendEventToWebsite,
-		};
-
-		await routeEvent(event, context);
-	}
+                        sendToVisitor: sendEventToVisitor,
+                        sendToWebsite: sendEventToWebsite,
+                };
+
+                try {
+                        await forwardMessageEventToRealtimeStreams(event);
+                } catch (error) {
+                        console.error("[Realtime] Failed to forward stream event", error);
+                }
+
+                await routeEvent(event, context);
+        }
 }
 
 const realtimeEmitter = new RealtimeEmitter();

apps/api/src/realtime/routes.ts

@@ -0,0 +1,152 @@
+import { db } from "@api/db";
+import { getTRPCSession } from "@api/db/queries/session";
+import { findVisitorForWebsite } from "@api/db/queries/visitor";
+import { getWebsiteByIdWithAccess } from "@api/db/queries/website";
+import {
+        AuthValidationError,
+        performAuthentication,
+} from "@api/lib/auth-validation";
+import { realtimeStreams } from "@api/realtime/streams";
+import { handle } from "@cossistant/realtime/server";
+import type { RealtimeEvent } from "@cossistant/types/realtime-events";
+import type { OpenAPIHono } from "@hono/zod-openapi";
+
+function parseOriginHeaders(request: Request): {
+        origin?: string;
+        protocol?: string;
+        hostname?: string;
+} {
+        const origin = request.headers.get("origin") ?? undefined;
+        if (!origin) {
+                return {};
+        }
+
+        try {
+                const url = new URL(origin);
+                return { origin, protocol: url.protocol, hostname: url.hostname };
+        } catch {
+                return { origin };
+        }
+}
+
+const visitorStreamHandler = handle({
+        realtime: realtimeStreams,
+        middleware: async ({ request, channel }) => {
+                const url = new URL(request.url);
+                const publicKey = url.searchParams.get("publicKey");
+                const visitorId = url.searchParams.get("visitorId");
+
+                if (!publicKey || !visitorId) {
+                        return new Response("Missing credentials", { status: 401 });
+                }
+
+                try {
+                        const authResult = await performAuthentication(
+                                undefined,
+                                publicKey,
+                                db,
+                                parseOriginHeaders(request)
+                        );
+
+                        const websiteId = authResult.apiKey.website?.id;
+
+                        if (!websiteId) {
+                                return new Response("Website not found", { status: 403 });
+                        }
+
+                        if (channel !== websiteId) {
+                                return new Response("Channel mismatch", { status: 403 });
+                        }
+
+                        const visitor = await findVisitorForWebsite(db, {
+                                visitorId,
+                                websiteId,
+                        });
+
+                        if (!visitor || visitor.blockedAt) {
+                                return new Response("Visitor not allowed", { status: 403 });
+                        }
+                } catch (error) {
+                        if (error instanceof AuthValidationError) {
+                                return new Response(error.message, { status: error.statusCode });
+                        }
+                        console.error("[Realtime] Visitor auth failed", error);
+                        return new Response("Authentication failed", { status: 401 });
+                }
+        },
+        filter: async ({ event, request }) => {
+                if (!Array.isArray(event.__event_path)) {
+                        return false;
+                }
+
+                if (event.__event_path[0] !== "visitor") {
+                        return false;
+                }
+
+                const data = event.data as RealtimeEvent;
+                if (!data || data.type !== "MESSAGE_CREATED") {
+                        return false;
+                }
+
+                const visitorId = new URL(request.url).searchParams.get("visitorId");
+                if (!visitorId) {
+                        return false;
+                }
+
+                const payloadVisitorId =
+                        data.payload.message.visitorId ?? data.visitorId ?? null;
+
+                return !payloadVisitorId || payloadVisitorId === visitorId;
+        },
+});
+
+const dashboardStreamHandler = handle({
+        realtime: realtimeStreams,
+        middleware: async ({ request, channel }) => {
+                if (!channel) {
+                        return new Response("Channel is required", { status: 400 });
+                }
+
+                const session = await getTRPCSession(db, { headers: request.headers });
+
+                if (!session?.user?.id) {
+                        return new Response("Unauthorized", { status: 401 });
+                }
+
+                const website = await getWebsiteByIdWithAccess(db, {
+                        userId: session.user.id,
+                        websiteId: channel,
+                });
+
+                if (!website) {
+                        return new Response("Forbidden", { status: 403 });
+                }
+        },
+        filter: ({ event }) => {
+                return Array.isArray(event.__event_path) && event.__event_path[0] === "dashboard";
+        },
+        responseHeaders: ({ request }) => {
+                const origin = request.headers.get("origin");
+                if (!origin) {
+                        return undefined;
+                }
+
+                return {
+                        "Access-Control-Allow-Origin": origin,
+                        "Access-Control-Allow-Credentials": "true",
+                        Vary: "Origin",
+                } satisfies HeadersInit;
+        },
+});
+
+export function registerRealtimeRoutes(app: OpenAPIHono): void {
+        app.get("/v1/realtime/visitor", async (c) => {
+                const response = await visitorStreamHandler(c.req.raw);
+                return response ?? new Response(null, { status: 204 });
+        });
+
+        app.get("/v1/realtime/dashboard", async (c) => {
+                const response = await dashboardStreamHandler(c.req.raw);
+                return response ?? new Response(null, { status: 204 });
+        });
+}

apps/api/src/realtime/stream-dispatcher.ts

@@ -0,0 +1,39 @@
+import type { RealtimeEvent } from "@cossistant/types/realtime-events";
+import { realtimeStreams } from "./streams";
+
+async function emitVisitorMessage(
+        event: RealtimeEvent<"MESSAGE_CREATED">
+): Promise<void> {
+        if (!event.websiteId) {
+                return;
+        }
+
+        await realtimeStreams
+                .channel(event.websiteId)
+                .visitor.MESSAGE_CREATED.emit(event);
+}
+
+async function emitDashboardMessage(
+        event: RealtimeEvent<"MESSAGE_CREATED">
+): Promise<void> {
+        if (!event.websiteId) {
+                return;
+        }
+
+        await realtimeStreams
+                .channel(event.websiteId)
+                .dashboard.MESSAGE_CREATED.emit(event);
+}
+
+export async function forwardMessageEventToRealtimeStreams(
+        event: RealtimeEvent
+): Promise<void> {
+        if (event.type !== "MESSAGE_CREATED") {
+                return;
+        }
+
+        await Promise.allSettled([
+                emitVisitorMessage(event),
+                emitDashboardMessage(event),
+        ]);
+}

apps/api/src/realtime/streams.ts

@@ -0,0 +1,30 @@
+import { getRedis } from "@api/redis";
+import { Realtime } from "@cossistant/realtime/server";
+import { RealtimeEvents } from "@cossistant/types/realtime-events";
+import { z } from "zod";
+
+const messageCreatedEventSchema = z.object({
+        type: z.literal("MESSAGE_CREATED"),
+        payload: RealtimeEvents.MESSAGE_CREATED,
+        timestamp: z.number(),
+        organizationId: z.string(),
+        websiteId: z.string(),
+        visitorId: z.string().nullable(),
+});
+
+const redis = getRedis();
+
+export const realtimeStreams = new Realtime({
+        redis,
+        verbose: process.env.NODE_ENV !== "production",
+        schema: {
+                visitor: z.object({
+                        MESSAGE_CREATED: messageCreatedEventSchema,
+                }),
+                dashboard: z.object({
+                        MESSAGE_CREATED: messageCreatedEventSchema,
+                }),
+        },
+});
+
+export type RealtimeMessageEvent = z.infer<typeof messageCreatedEventSchema>;

apps/web/package.json

@@ -106,4 +106,4 @@
     "typescript": "^5",
     "@types/bun": "^1.2.22"
   }
-}
+}

apps/web/src/app/(dashboard)/[websiteSlug]/providers/realtime.tsx

@@ -4,6 +4,8 @@ import {
 	type RealtimeEventHandlersMap,
 	useRealtime,
 } from "@cossistant/next/realtime";
+import { useDashboardRealtime } from "@cossistant/realtime/client";
+import type { RealtimeEvent } from "@cossistant/types/realtime-events";
 import { useQueryNormalizer } from "@normy/react-query";
 import { useQueryClient } from "@tanstack/react-query";
 import { type ReactNode, useMemo } from "react";
@@ -47,7 +49,6 @@ export function DashboardRealtimeProvider({
 					});
 				},
 			],
-			MESSAGE_CREATED: handleMessageCreated,
 			CONVERSATION_SEEN: [
 				(_data, meta) => {
 					void handleConversationSeen({
@@ -80,5 +81,18 @@ export function DashboardRealtimeProvider({
 		},
 	});
 
+	useDashboardRealtime({
+		websiteId: website.id,
+		onMessageCreated: (event: RealtimeEvent<"MESSAGE_CREATED">) => {
+			const handlers = Array.isArray(handleMessageCreated)
+				? handleMessageCreated
+				: [handleMessageCreated];
+
+			for (const handler of handlers) {
+				handler(event.payload, { event, context: realtimeContext });
+			}
+		},
+	});
+
 	return children;
 }