From ac804c501ee29002b949338109995251362df229 Mon Sep 17 00:00:00 2001 From: Anthony Riera Date: Thu, 9 Oct 2025 15:18:55 +0300 Subject: [PATCH 1/2] feat(realtime): add redis stream support --- apps/api/src/index.ts | 3 + apps/api/src/realtime/emitter.ts | 35 +- apps/api/src/realtime/routes.ts | 152 +++++++ apps/api/src/realtime/stream-dispatcher.ts | 39 ++ apps/api/src/realtime/streams.ts | 30 ++ apps/web/package.json | 2 +- .../[websiteSlug]/providers/realtime.tsx | 16 +- bun.lock | 46 ++- .../react/src/realtime/support-provider.tsx | 69 ++-- packages/realtime/package.json | 2 +- packages/realtime/src/client/index.ts | 2 + .../src/client/use-dashboard-realtime.ts | 69 ++++ packages/realtime/src/client/use-realtime.ts | 53 ++- .../src/client/use-visitor-realtime.ts | 74 ++++ packages/realtime/src/server/handler.ts | 388 +++++++++++------- packages/realtime/src/server/realtime.ts | 28 +- 16 files changed, 784 insertions(+), 224 deletions(-) create mode 100644 apps/api/src/realtime/routes.ts create mode 100644 apps/api/src/realtime/stream-dispatcher.ts create mode 100644 apps/api/src/realtime/streams.ts create mode 100644 packages/realtime/src/client/use-dashboard-realtime.ts create mode 100644 packages/realtime/src/client/use-visitor-realtime.ts diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 5cc8795ac..db3e1ea4a 100644 --- a/apps/api/src/index.ts +++ b/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); diff --git a/apps/api/src/realtime/emitter.ts b/apps/api/src/realtime/emitter.ts index 0adc08d8f..830d08e7c 100644 --- a/apps/api/src/realtime/emitter.ts +++ b/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 = { - type, - payload: data, - timestamp: options.timestamp ?? Date.now(), - websiteId, + const event: RealtimeEvent = { + 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(); diff --git a/apps/api/src/realtime/routes.ts b/apps/api/src/realtime/routes.ts new file mode 100644 index 000000000..4502142d5 --- /dev/null +++ b/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 }); + }); +} diff --git a/apps/api/src/realtime/stream-dispatcher.ts b/apps/api/src/realtime/stream-dispatcher.ts new file mode 100644 index 000000000..d50e83242 --- /dev/null +++ b/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 { + if (!event.websiteId) { + return; + } + + await realtimeStreams + .channel(event.websiteId) + .visitor.MESSAGE_CREATED.emit(event); +} + +async function emitDashboardMessage( + event: RealtimeEvent<"MESSAGE_CREATED"> +): Promise { + if (!event.websiteId) { + return; + } + + await realtimeStreams + .channel(event.websiteId) + .dashboard.MESSAGE_CREATED.emit(event); +} + +export async function forwardMessageEventToRealtimeStreams( + event: RealtimeEvent +): Promise { + if (event.type !== "MESSAGE_CREATED") { + return; + } + + await Promise.allSettled([ + emitVisitorMessage(event), + emitDashboardMessage(event), + ]); +} diff --git a/apps/api/src/realtime/streams.ts b/apps/api/src/realtime/streams.ts new file mode 100644 index 000000000..8984a2c2d --- /dev/null +++ b/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; diff --git a/apps/web/package.json b/apps/web/package.json index 6ea3f13ef..095c7eb88 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -106,4 +106,4 @@ "typescript": "^5", "@types/bun": "^1.2.22" } -} \ No newline at end of file +} diff --git a/apps/web/src/app/(dashboard)/[websiteSlug]/providers/realtime.tsx b/apps/web/src/app/(dashboard)/[websiteSlug]/providers/realtime.tsx index 41d1dce27..da2fa1d8d 100644 --- a/apps/web/src/app/(dashboard)/[websiteSlug]/providers/realtime.tsx +++ b/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; } diff --git a/bun.lock b/bun.lock index 2d4819db6..958e0490a 100644 --- a/bun.lock +++ b/bun.lock @@ -22,6 +22,7 @@ "apps/api": { "name": "@cossistant/api", "dependencies": { + "@cossistant/realtime": "workspace:*", "@cossistant/transactional": "workspace:*", "@cossistant/types": "workspace:*", "@hono-rate-limiter/redis": "^0.1.4", @@ -65,6 +66,7 @@ "@cossistant/core": "workspace:*", "@cossistant/location": "workspace:*", "@cossistant/next": "workspace:*", + "@cossistant/realtime": "workspace:*", "@cossistant/transactional": "workspace:*", "@cossistant/types": "workspace:*", "@hookform/resolvers": "^5.2.2", @@ -195,6 +197,7 @@ "version": "0.0.0", "dependencies": { "@cossistant/core": "workspace:*", + "@cossistant/realtime": "workspace:*", "@cossistant/types": "workspace:*", "nanoid": "^5.1.5", "react-markdown": "^10.1.0", @@ -220,6 +223,23 @@ "tailwindcss", ], }, + "packages/realtime": { + "name": "@cossistant/realtime", + "version": "0.0.0", + "devDependencies": { + "@types/node": "^24.6.0", + "@types/react": "^19.1.16", + "@types/react-dom": "^19.1.9", + "ioredis": "^5.4.2", + "typescript": "^5.9.2", + "zod": "^4.1.11", + }, + "peerDependencies": { + "react": ">=16.8.0", + "react-dom": ">=16.8.0", + "zod": "^3.25.0 || ^4.0.0", + }, + }, "packages/transactional": { "name": "@cossistant/transactional", "version": "1.0.0", @@ -408,6 +428,8 @@ "@cossistant/react": ["@cossistant/react@workspace:packages/react"], + "@cossistant/realtime": ["@cossistant/realtime@workspace:packages/realtime"], + "@cossistant/transactional": ["@cossistant/transactional@workspace:packages/transactional"], "@cossistant/types": ["@cossistant/types@workspace:packages/types"], @@ -1044,7 +1066,7 @@ "@types/ms": ["@types/ms@2.1.0", "", {}, "sha512-GsCCIZDE/p3i96vtEqx+7dBUGXrc7zeSK3wwPHIaRThS+9OhWIXRqzs4d6k1SVU8g91DrNRWxWUGhp5KXQb2VA=="], - "@types/node": ["@types/node@24.5.0", "", { "dependencies": { "undici-types": "~7.12.0" } }, "sha512-y1dMvuvJspJiPSDZUQ+WMBvF7dpnEqN4x9DDC9ie5Fs/HUZJA3wFp7EhHoVaKX/iI0cRoECV8X2jL8zi0xrHCg=="], + "@types/node": ["@types/node@24.7.0", "", { "dependencies": { "undici-types": "~7.14.0" } }, "sha512-IbKooQVqUBrlzWTi79E8Fw78l8k1RNtlDDNWsFZs7XonuQSJ8oNYfEeclhprUldXISRMLzBpILuKgPlIxm+/Yw=="], "@types/normalize-path": ["@types/normalize-path@3.0.2", "", {}, "sha512-DO++toKYPaFn0Z8hQ7Tx+3iT9t77IJo/nDiqTXilgEP+kPNIYdpS9kh3fXuc53ugqwp9pxC1PVjCpV1tQDyqMA=="], @@ -1056,7 +1078,7 @@ "@types/prismjs": ["@types/prismjs@1.26.5", "", {}, "sha512-AUZTa7hQ2KY5L7AmtSiqxlhWxb4ina0yd8hNbl4TWuqnv/pFP0nDMb3YrfSBf4hJVGLh2YEIBfKaBW/9UEl6IQ=="], - "@types/react": ["@types/react@19.1.12", "", { "dependencies": { "csstype": "^3.0.2" } }, "sha512-cMoR+FoAf/Jyq6+Df2/Z41jISvGZZ2eTlnsaJRptmZ76Caldwy1odD4xTr/gNV9VLj0AWgg/nmkevIyUfIIq5w=="], + "@types/react": ["@types/react@19.2.2", "", { "dependencies": { "csstype": "^3.0.2" } }, "sha512-6mDvHUFSjyT2B2yeNx2nUgMxh9LtOWvkhIU3uePn2I2oyNymUAX1NIsdgviM4CH+JSrp2D2hsMvJOkxY+0wNRA=="], "@types/react-dom": ["@types/react-dom@19.1.9", "", { "peerDependencies": { "@types/react": "^19.0.0" } }, "sha512-qXRuZaOsAdXKFyOhRBg6Lqqc0yay13vN7KrIg4L7N4aaHN68ma9OK3NE1BoDFgFOTfM7zg+3/8+2n8rLUH3OKQ=="], @@ -2592,7 +2614,7 @@ "uncrypto": ["uncrypto@0.1.3", "", {}, "sha512-Ql87qFHB3s/De2ClA9e0gsnS6zXG27SkTiSJwjCc9MebbfapQfuPzumMIUMi38ezPZVNFcHI9sUIepeQfw8J8Q=="], - "undici-types": ["undici-types@7.12.0", "", {}, "sha512-goOacqME2GYyOZZfb5Lgtu+1IDmAlAEu5xnD3+xTzS10hT0vzpf0SPjkXwAw9Jm+4n/mQGDP3LO8CPbYROeBfQ=="], + "undici-types": ["undici-types@7.14.0", "", {}, "sha512-QQiYxHuyZ9gQUIrmPo3IA+hUl4KYk8uSA7cHrcKd/l3p1OTpZcM0Tbp9x7FAtXdAYhlasd60ncPpgu6ihG6TOA=="], "unicorn-magic": ["unicorn-magic@0.1.0", "", {}, "sha512-lRfVq8fE8gz6QMBuDM6a+LO3IAzTi05H6gCVaUpir2E1Rwpo4ZUog45KpNXKC/Mn3Yb9UDuHumeFTo9iV/D9FQ=="], @@ -2690,7 +2712,7 @@ "yoctocolors": ["yoctocolors@2.1.2", "", {}, "sha512-CzhO+pFNo8ajLM2d2IW/R93ipy99LWjtwblvC1RsoSUMZgyLbYFr221TnSNT7GjGdYui6P459mw9JH/g/zW2ug=="], - "zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + "zod": ["zod@4.1.12", "", {}, "sha512-JInaHOamG8pt5+Ey8kGmdcAcg3OL9reK8ltczgHTAwNhMys/6ThXHityHxVV2p3fkw/c+MAvBHFVYHFZDmjMCQ=="], "zod-openapi": ["zod-openapi@4.2.4", "", { "peerDependencies": { "zod": "^3.21.4" } }, "sha512-tsrQpbpqFCXqVXUzi3TPwFhuMtLN3oNZobOtYnK6/5VkXsNdnIgyNr4r8no4wmYluaxzN3F7iS+8xCW8BmMQ8g=="], @@ -2708,6 +2730,8 @@ "@ai-sdk/react/@ai-sdk/provider-utils": ["@ai-sdk/provider-utils@2.2.8", "", { "dependencies": { "@ai-sdk/provider": "1.1.3", "nanoid": "^3.3.8", "secure-json-parse": "^2.7.0" }, "peerDependencies": { "zod": "^3.23.8" } }, "sha512-fqhG+4sCVv8x7nFzYnFo19ryhAa3w096Kmc3hWxMQfW/TubPOmt3A6tYZhl4mUfQWWQMsuSkLrtjlWuXBVSGQA=="], + "@ai-sdk/react/zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + "@ai-sdk/ui-utils/@ai-sdk/provider": ["@ai-sdk/provider@1.1.3", "", { "dependencies": { "json-schema": "^0.4.0" } }, "sha512-qZMxYJ0qqX/RfnuIaab+zp8UAeJn/ygXXAffR5I4N0n1IrvA6qBsjc8hXLmBiMV2zoXlifkacF7sEFnYnjBcqg=="], "@ai-sdk/ui-utils/@ai-sdk/provider-utils": ["@ai-sdk/provider-utils@2.2.8", "", { "dependencies": { "@ai-sdk/provider": "1.1.3", "nanoid": "^3.3.8", "secure-json-parse": "^2.7.0" }, "peerDependencies": { "zod": "^3.23.8" } }, "sha512-fqhG+4sCVv8x7nFzYnFo19ryhAa3w096Kmc3hWxMQfW/TubPOmt3A6tYZhl4mUfQWWQMsuSkLrtjlWuXBVSGQA=="], @@ -2876,12 +2900,16 @@ "anymatch/picomatch": ["picomatch@2.3.1", "", {}, "sha512-JU3teHTNjmE2VCGFzuY8EXzCDVwEqB2a8fsIvwaStHhAWJEeVd1o1QD80CU6+ZdEXXSLbSsuLwJjkCBWqRQUVA=="], + "better-auth/zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + "bl/buffer": ["buffer@5.7.1", "", { "dependencies": { "base64-js": "^1.3.1", "ieee754": "^1.1.13" } }, "sha512-EHcyIPBQ4BSGlvjB16k5KgAJ27CIsHY/2JBmCRReo48y9rQ3MaUzWX3KVlBa4U7MyX02HdVj0K7C3WaB3ju7FQ=="], "body-parser/raw-body": ["raw-body@3.0.1", "", { "dependencies": { "bytes": "3.1.2", "http-errors": "2.0.0", "iconv-lite": "0.7.0", "unpipe": "1.0.0" } }, "sha512-9G8cA+tuMS75+6G/TzW8OtLzmBDMo8p1JRxN5AZ+LAp8uxGA8V8GZm4GQ4/N5QNQEnLmg6SS7wyuSmbKepiKqA=="], "body-parser/type-is": ["type-is@2.0.1", "", { "dependencies": { "content-type": "^1.0.5", "media-typer": "^1.1.0", "mime-types": "^3.0.0" } }, "sha512-OZs6gsjF4vMp32qrCbiVSkrFmXtG/AZhY3t0iAMrMBiAZyV9oALtXO8hsrHbMXF9x6L3grlFuwW2oAz7cav+Gw=="], + "bun-types/@types/node": ["@types/node@24.5.0", "", { "dependencies": { "undici-types": "~7.12.0" } }, "sha512-y1dMvuvJspJiPSDZUQ+WMBvF7dpnEqN4x9DDC9ie5Fs/HUZJA3wFp7EhHoVaKX/iI0cRoECV8X2jL8zi0xrHCg=="], + "chalk/supports-color": ["supports-color@7.2.0", "", { "dependencies": { "has-flag": "^4.0.0" } }, "sha512-qpCAvRl9stuOHveKsn7HncJRvv501qIacKzQlO/+Lwxc9+0q2wLyv4Dfvt80/DPn2pqOBsJdDiogXGR9+OvwRw=="], "cosmiconfig-typescript-loader/jiti": ["jiti@2.5.1", "", { "bin": { "jiti": "lib/jiti-cli.mjs" } }, "sha512-twQoecYPiVA5K/h6SxtORw/Bs3ar+mLUtoPSc7iMXzQzK8d7eJ/R09wmTwAjiamETn1cXYPGfNnu7DMoHgu12w=="], @@ -2902,6 +2930,8 @@ "fast-glob/glob-parent": ["glob-parent@5.1.2", "", { "dependencies": { "is-glob": "^4.0.1" } }, "sha512-AOIgSQCepiJYwP3ARnGx+5VnTu2HBYdzbGP45eLw1vr3zB3vZLeyed1sC9hnbcOc9/SrMyM5RPQrkGz4aS9Zow=="], + "fumadocs-mdx/zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + "fumadocs-ui/fumadocs-core": ["fumadocs-core@15.7.12", "", { "dependencies": { "@formatjs/intl-localematcher": "^0.6.1", "@orama/orama": "^3.1.13", "@shikijs/rehype": "^3.12.2", "@shikijs/transformers": "^3.12.2", "github-slugger": "^2.0.0", "hast-util-to-estree": "^3.1.3", "hast-util-to-jsx-runtime": "^2.3.6", "image-size": "^2.0.2", "negotiator": "^1.0.0", "npm-to-yarn": "^3.0.1", "react-remove-scroll": "^2.7.1", "remark": "^15.0.0", "remark-gfm": "^4.0.1", "remark-rehype": "^11.1.2", "scroll-into-view-if-needed": "^3.1.0", "shiki": "^3.12.2", "unist-util-visit": "^5.0.0" }, "peerDependencies": { "@mixedbread/sdk": "^0.19.0", "@oramacloud/client": "1.x.x || 2.x.x", "@tanstack/react-router": "1.x.x", "@types/react": "*", "algoliasearch": "5.x.x", "next": "14.x.x || 15.x.x", "react": "18.x.x || 19.x.x", "react-dom": "18.x.x || 19.x.x", "react-router": "7.x.x", "waku": "^0.26.0" }, "optionalPeers": ["@mixedbread/sdk", "@oramacloud/client", "@tanstack/react-router", "@types/react", "algoliasearch", "next", "react", "react-dom", "react-router", "waku"] }, "sha512-b30+TYKaqdNoOQ75dQqWw338KZsgLPCdZrQNP1gyLiDhE/A0kzJBe+jjLyIFL/6rLtZFa3HYVaV9GnRtazPW6Q=="], "glob/minimatch": ["minimatch@10.0.3", "", { "dependencies": { "@isaacs/brace-expansion": "^5.0.0" } }, "sha512-IPZ167aShDZZUMdRk66cyQAW3qr0WzbHkPdMYa8bzZhlHhO3jALbKdxcaak7W9FfT2rZNpQuUu4Od7ILEpXSaw=="], @@ -3004,6 +3034,8 @@ "type-is/mime-types": ["mime-types@2.1.35", "", { "dependencies": { "mime-db": "1.52.0" } }, "sha512-ZDY+bPm5zTTF+YpCrAU9nK0UgICYPT0QtT1NZWFv4s++TNkcgVaT0g6+4R2uI4MjQjzysHB1zxuWL50hzaeXiw=="], + "ultracite/zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + "unconfig/jiti": ["jiti@2.5.1", "", { "bin": { "jiti": "lib/jiti-cli.mjs" } }, "sha512-twQoecYPiVA5K/h6SxtORw/Bs3ar+mLUtoPSc7iMXzQzK8d7eJ/R09wmTwAjiamETn1cXYPGfNnu7DMoHgu12w=="], "vite-node/debug": ["debug@4.4.1", "", { "dependencies": { "ms": "^2.1.3" } }, "sha512-KcKCqiftBJcZr++7ykoDIEwSa3XWowTfNPo92BYxjXiyYEVrUQh2aLyhxBCwww+heortUFxEJYcRzosstTEBYQ=="], @@ -3040,6 +3072,10 @@ "@commitlint/top-level/find-up/path-exists": ["path-exists@5.0.0", "", {}, "sha512-RjhtfwJOxzcFmNOi6ltcbcu4Iu+FL3zEj83dk4kAS+fVpTxXLO1b38RvJgT/0QwvV/L3aY9TAnyv0EOqW4GoMQ=="], + "@cossistant/api/better-auth/zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + + "@cossistant/web/better-auth/zod": ["zod@4.1.8", "", {}, "sha512-5R1P+WwQqmmMIEACyzSvo4JXHY5WiAFHRMg+zBZKgKS+Q1viRa0C1hmUKtHltoIFKtIdki3pRxkmpP74jnNYHQ=="], + "@esbuild-kit/core-utils/esbuild/@esbuild/android-arm": ["@esbuild/android-arm@0.18.20", "", { "os": "android", "cpu": "arm" }, "sha512-fyi7TDI/ijKKNZTUJAQqiG5T7YjJXgnzkURqmGj13C6dCqckZBLdl4h7bkhHt/t0WP+zO9/zwroDvANaOqO5Sw=="], "@esbuild-kit/core-utils/esbuild/@esbuild/android-arm64": ["@esbuild/android-arm64@0.18.20", "", { "os": "android", "cpu": "arm64" }, "sha512-Nz4rJcchGDtENV0eMKUNa6L12zz2zBDXuhj/Vjh18zGqB44Bi7MBMSXjgunJgjRhCmKOjnPuZp4Mb6OKqtMHLQ=="], @@ -3312,6 +3348,8 @@ "body-parser/type-is/media-typer": ["media-typer@1.1.0", "", {}, "sha512-aisnrDP4GNe06UcKFnV5bfMNPBUw4jsLGaWwWfnH3v02GnBuXX2MCVn5RbrWo0j3pczUilYblq7fQ7Nw2t5XKw=="], + "bun-types/@types/node/undici-types": ["undici-types@7.12.0", "", {}, "sha512-goOacqME2GYyOZZfb5Lgtu+1IDmAlAEu5xnD3+xTzS10hT0vzpf0SPjkXwAw9Jm+4n/mQGDP3LO8CPbYROeBfQ=="], + "engine.io/@types/node/undici-types": ["undici-types@7.10.0", "", {}, "sha512-t5Fy/nfn+14LuOc2KNYg75vZqClpAiqscVvMygNnlsHBFpSXdJaYtXMcdNLpl/Qvc3P2cB3s6lOV51nqsFq4ag=="], "express/type-is/media-typer": ["media-typer@1.1.0", "", {}, "sha512-aisnrDP4GNe06UcKFnV5bfMNPBUw4jsLGaWwWfnH3v02GnBuXX2MCVn5RbrWo0j3pczUilYblq7fQ7Nw2t5XKw=="], diff --git a/packages/react/src/realtime/support-provider.tsx b/packages/react/src/realtime/support-provider.tsx index eb6f69265..5d276b768 100644 --- a/packages/react/src/realtime/support-provider.tsx +++ b/packages/react/src/realtime/support-provider.tsx @@ -1,7 +1,8 @@ import type { CossistantClient } from "@cossistant/core"; import type { RealtimeEvent } from "@cossistant/types/realtime-events"; import type React from "react"; -import { useMemo } from "react"; +import { useCallback, useMemo } from "react"; +import { useVisitorRealtime } from "@cossistant/realtime/client"; import { useSupport } from "../provider"; import { applyConversationSeenEvent } from "./seen-store"; import { @@ -27,7 +28,19 @@ type SupportRealtimeProviderProps = { export function SupportRealtimeProvider({ children, }: SupportRealtimeProviderProps) { - const { website, client, visitor } = useSupport(); + const { website, client, visitor } = useSupport(); + + const handleMessageCreated = useCallback( + (event: RealtimeEvent<"MESSAGE_CREATED">) => { + if (website?.id && event.websiteId !== website.id) { + return; + } + + clearTypingFromMessage(event); + client.handleRealtimeEvent(event); + }, + [client, website?.id] + ); const realtimeContext = useMemo( () => ({ @@ -38,30 +51,11 @@ export function SupportRealtimeProvider({ [website?.id, visitor?.id, client] ); - const events = useMemo( - () => ({ - MESSAGE_CREATED: ( - _data: RealtimeEvent["payload"], - { - event, - context, - }: { - event: RealtimeEvent<"MESSAGE_CREATED">; - context: SupportRealtimeContext; - } - ) => { - if (context.websiteId && event.websiteId !== context.websiteId) { - return; - } - - // Clear typing state when a message is sent - clearTypingFromMessage(event); - - context.client.handleRealtimeEvent(event); - }, - CONVERSATION_SEEN: ( - _data: RealtimeEvent["payload"], - { + const events = useMemo( + () => ({ + CONVERSATION_SEEN: ( + _data: RealtimeEvent["payload"], + { event, context, }: { @@ -102,12 +96,21 @@ export function SupportRealtimeProvider({ [] ); - useRealtime({ - context: realtimeContext, - events, - websiteId: realtimeContext.websiteId, - visitorId: website?.visitor?.id ?? null, - }); + useRealtime({ + context: realtimeContext, + events, + websiteId: realtimeContext.websiteId, + visitorId: website?.visitor?.id ?? null, + }); + + const { publicKey } = client.getConfiguration(); + + useVisitorRealtime({ + websiteId: realtimeContext.websiteId, + visitorId: realtimeContext.visitorId, + publicKey: publicKey ?? null, + onMessageCreated: handleMessageCreated, + }); - return <>{children}; + return <>{children}; } diff --git a/packages/realtime/package.json b/packages/realtime/package.json index d8451ac01..9c7a5d8f8 100644 --- a/packages/realtime/package.json +++ b/packages/realtime/package.json @@ -53,7 +53,7 @@ "@types/node": "^24.6.0", "@types/react": "^19.1.16", "@types/react-dom": "^19.1.9", - "@upstash/redis": "^1.35.4", + "ioredis": "^5.4.2", "typescript": "^5.9.2" }, "dependencies": {}, diff --git a/packages/realtime/src/client/index.ts b/packages/realtime/src/client/index.ts index b15510192..f73927307 100644 --- a/packages/realtime/src/client/index.ts +++ b/packages/realtime/src/client/index.ts @@ -1 +1,3 @@ export * from "./use-realtime"; +export * from "./use-visitor-realtime"; +export * from "./use-dashboard-realtime"; diff --git a/packages/realtime/src/client/use-dashboard-realtime.ts b/packages/realtime/src/client/use-dashboard-realtime.ts new file mode 100644 index 000000000..faf70085b --- /dev/null +++ b/packages/realtime/src/client/use-dashboard-realtime.ts @@ -0,0 +1,69 @@ +import { useEffect, useMemo, useRef } from "react"; +import type { RealtimeEvent } from "@cossistant/types/realtime-events"; +import { useRealtime } from "./use-realtime"; + +export type DashboardRealtimeEvent = RealtimeEvent<"MESSAGE_CREATED">; + +type DashboardRealtimeSchema = { + dashboard: { + MESSAGE_CREATED: DashboardRealtimeEvent; + }; +}; + +export type UseDashboardRealtimeOptions = { + websiteId: string | null | undefined; + endpoint?: string; + enabled?: boolean; + onMessageCreated?: (event: DashboardRealtimeEvent) => void; +}; + +export function useDashboardRealtime({ + websiteId, + endpoint = "/v1/realtime/dashboard", + enabled = true, + onMessageCreated, +}: UseDashboardRealtimeOptions) { + const handlerRef = useRef< + ((event: DashboardRealtimeEvent) => void) | undefined + >(undefined); + + useEffect(() => { + handlerRef.current = onMessageCreated; + }, [onMessageCreated]); + + const shouldEnable = + enabled && Boolean(websiteId && onMessageCreated); + + const events = useMemo(() => { + if (!onMessageCreated) { + return undefined; + } + + return { + dashboard: { + MESSAGE_CREATED: (event: DashboardRealtimeEvent) => { + const handler = handlerRef.current; + if (handler) { + handler(event); + } + }, + }, + } satisfies DashboardEventHandlers; + }, [onMessageCreated]); + + return useRealtime({ + channel: websiteId ?? undefined, + endpoint, + enabled: shouldEnable, + events, + params: { + websiteId: websiteId ?? undefined, + }, + }); +} + +type DashboardEventHandlers = { + dashboard: { + MESSAGE_CREATED: (event: DashboardRealtimeEvent) => void; + }; +}; diff --git a/packages/realtime/src/client/use-realtime.ts b/packages/realtime/src/client/use-realtime.ts index 1595032c7..4190a064d 100644 --- a/packages/realtime/src/client/use-realtime.ts +++ b/packages/realtime/src/client/use-realtime.ts @@ -11,6 +11,8 @@ import type { type Opts = { channel?: string; enabled?: boolean; + endpoint?: string; + params?: Record; events?: Partial<{ [N in keyof T]: Partial<{ [K in keyof T[N]]: (data: T[N][K]) => void; @@ -24,6 +26,8 @@ const PING_TIMEOUT_MS = 20_000; export const useRealtime = >>({ channel = "default", enabled = true, + endpoint = "/api/realtime", + params, events, maxReconnectAttempts = 3, }: Opts = {}) => { @@ -84,18 +88,43 @@ export const useRealtime = >>({ setStatus("connecting"); try { - const reconnectParam = reconnect ? "&reconnect=true" : ""; - - const lastAckParam = - reconnect && lastAckRef.current - ? `&last_ack=${encodeURIComponent(lastAckRef.current)}` - : ""; - - const eventSource = new EventSource( - `/api/realtime?channel=${encodeURIComponent( - channel - )}${reconnectParam}${lastAckParam}` - ); + if (typeof window === "undefined") { + return; + } + + const baseUrl = (() => { + try { + return new URL(endpoint, window.location.origin); + } catch (error) { + console.warn("Invalid realtime endpoint", { endpoint, error }); + return null; + } + })(); + + if (!baseUrl) { + setStatus("error"); + return; + } + + baseUrl.searchParams.set("channel", channel); + + if (params) { + for (const [key, value] of Object.entries(params)) { + if (value !== undefined && value !== null) { + baseUrl.searchParams.set(key, value); + } + } + } + + if (reconnect) { + baseUrl.searchParams.set("reconnect", "true"); + } + + if (reconnect && lastAckRef.current) { + baseUrl.searchParams.set("last_ack", lastAckRef.current); + } + + const eventSource = new EventSource(baseUrl.toString()); eventSourceRef.current = eventSource; eventSource.onopen = () => { diff --git a/packages/realtime/src/client/use-visitor-realtime.ts b/packages/realtime/src/client/use-visitor-realtime.ts new file mode 100644 index 000000000..ed1a92776 --- /dev/null +++ b/packages/realtime/src/client/use-visitor-realtime.ts @@ -0,0 +1,74 @@ +import { useEffect, useMemo, useRef } from "react"; +import type { RealtimeEvent } from "@cossistant/types/realtime-events"; +import { useRealtime } from "./use-realtime"; + +export type VisitorRealtimeEvent = RealtimeEvent<"MESSAGE_CREATED">; + +type VisitorRealtimeSchema = { + visitor: { + MESSAGE_CREATED: VisitorRealtimeEvent; + }; +}; + +export type UseVisitorRealtimeOptions = { + websiteId: string | null | undefined; + visitorId: string | null | undefined; + publicKey: string | null | undefined; + endpoint?: string; + enabled?: boolean; + onMessageCreated?: (event: VisitorRealtimeEvent) => void; +}; + +export function useVisitorRealtime({ + websiteId, + visitorId, + publicKey, + endpoint = "/v1/realtime/visitor", + enabled = true, + onMessageCreated, +}: UseVisitorRealtimeOptions) { + const handlerRef = useRef< + ((event: VisitorRealtimeEvent) => void) | undefined + >(undefined); + + useEffect(() => { + handlerRef.current = onMessageCreated; + }, [onMessageCreated]); + + const shouldEnable = + enabled && Boolean(websiteId && visitorId && publicKey && onMessageCreated); + + const events = useMemo(() => { + if (!onMessageCreated) { + return undefined; + } + + return { + visitor: { + MESSAGE_CREATED: (event: VisitorRealtimeEvent) => { + const handler = handlerRef.current; + if (handler) { + handler(event); + } + }, + }, + } satisfies VisitorEventHandlers; + }, [onMessageCreated]); + + return useRealtime({ + channel: websiteId ?? undefined, + endpoint, + enabled: shouldEnable, + params: { + visitorId: visitorId ?? undefined, + publicKey: publicKey ?? undefined, + }, + events, + }); +} + +type VisitorEventHandlers = { + visitor: { + MESSAGE_CREATED: (event: VisitorRealtimeEvent) => void; + }; +}; diff --git a/packages/realtime/src/server/handler.ts b/packages/realtime/src/server/handler.ts index d31ab69fd..3079f8dfd 100644 --- a/packages/realtime/src/server/handler.ts +++ b/packages/realtime/src/server/handler.ts @@ -5,15 +5,37 @@ import type { SystemEvent, UserEvent } from "../types.js"; import type { Opts, Realtime } from "./realtime.js"; +type MiddlewareFn = ({ + request, + channel, +}: { + request: Request; + channel: string; +}) => Response | void | Promise; + +type MessageFilter = ({ + event, + request, + channel, +}: { + event: UserEvent; + request: Request; + channel: string; +}) => boolean | Promise; + +type ResponseHeadersFn = ({ + request, + channel, +}: { + request: Request; + channel: string; +}) => HeadersInit | undefined; + export function handle(config: { realtime: Realtime; - middleware?: ({ - request, - channel, - }: { - request: Request; - channel: string; - }) => Response | void | Promise; + middleware?: MiddlewareFn; + filter?: MessageFilter; + responseHeaders?: ResponseHeadersFn; }): (request: Request) => Promise { return async (request: Request) => { const requestStartTime = Date.now(); @@ -41,14 +63,13 @@ export function handle(config: { } let cleanup: (() => Promise) | undefined; - let subscriber: ReturnType; + let subscriber: ReturnType | undefined; let reconnectTimeout: NodeJS.Timeout | undefined; let keepaliveInterval: NodeJS.Timeout | undefined; let isClosed = false; let handleAbort: (() => Promise) | undefined; - let onSubscribe: (() => Promise) | undefined; - let onError: ((err: Error) => void) | undefined; - let onUnsubscribe: (() => void) | undefined; + + const streamKey = `channel:${channel}`; const stream = new ReadableStream({ async start(controller) { @@ -57,7 +78,33 @@ export function handle(config: { return; } - cleanup = async () => { + const safeEnqueue = (data: Uint8Array) => { + if (!isClosed) { + controller.enqueue(data); + } + }; + + const sendUserEvent = async (event: UserEvent) => { + if (config.filter) { + try { + const shouldSend = await config.filter({ + event, + request, + channel, + }); + if (!shouldSend) { + return; + } + } catch (error) { + logger.error("⚠️ Filter function failed", error); + return; + } + } + + safeEnqueue(json(event)); + }; + + const closeStream = async () => { if (isClosed) { return; } @@ -70,26 +117,30 @@ export function handle(config: { request.signal.removeEventListener("abort", handleAbort); } - await subscriber?.unsubscribe().catch((err) => { - logger.error("⚠️ Error during unsubscribe:", err); - }); + if (subscriber) { + try { + await subscriber.unsubscribe(streamKey); + } catch (err) { + logger.error("⚠️ Error during unsubscribe:", err); + } + subscriber.disconnect(); + subscriber = undefined; + } controller.close(); }; + cleanup = closeStream; + handleAbort = async () => { - await cleanup?.(); + await closeStream(); }; request.signal.addEventListener("abort", handleAbort); - subscriber = redis.subscribe(`channel:${channel}`); - - const safeEnqueue = (data: Uint8Array) => { - if (!isClosed) { - controller.enqueue(data); - } - }; + subscriber = redis.duplicate(); + await subscriber.connect(); + await subscriber.subscribe(streamKey); const elapsedMs = Date.now() - requestStartTime; const remainingMs = config.realtime._maxDurationSecs * 1000 - elapsedMs; @@ -97,76 +148,56 @@ export function handle(config: { reconnectTimeout = setTimeout(async () => { safeEnqueue(json({ type: "reconnect" })); - await cleanup?.(); + await closeStream(); }, streamDurationMs); - onSubscribe = async () => { - logger.log("✅ Subscription established:", { channel }); - try { - if (reconnect === "true" && last_ack) { - const startId = `(${last_ack}`; - const missingMessages = await redis.xrange( - `channel:${channel}`, - startId, - "+" - ); - - const connectedEvent: SystemEvent = { - type: "connected", - channel, - }; - - safeEnqueue(json(connectedEvent)); - - if (Array.from(Object.keys(missingMessages)).length > 0) { - for (const [__stream_id, value] of Object.entries( - missingMessages - )) { - if (typeof value === "object" && value !== null) { - const { __event_path, data } = value as Record< - string, - unknown - >; - const userEvent: UserEvent = { - data, - __event_path: __event_path as string[], - __stream_id, - }; - safeEnqueue(json(userEvent)); - } - } - } - } else { - const lastMessage = await redis.xrevrange( - `channel:${channel}`, - "+", - "-", - 1 - ); - - const messageEntries = Object.entries(lastMessage); - const currentCursor = messageEntries[0]?.[0] ?? "0-0"; - - const connectedEvent: SystemEvent = { - type: "connected", - channel, - cursor: currentCursor, - }; + const sendConnectedEvent = (cursor?: string) => { + const connectedEvent: SystemEvent = { + type: "connected", + channel, + ...(cursor ? { cursor } : {}), + }; + safeEnqueue(json(connectedEvent)); + }; - safeEnqueue(json(connectedEvent)); - } - } catch (err) { - logger.error("Error in subscribe handler:", err); - safeEnqueue( - json({ - type: "error", - error: err instanceof Error ? err.message : "Unknown error", - }) + const replayMissedMessages = async () => { + if (!(reconnect === "true" && last_ack)) { + const lastMessage = await redis.xrevrange( + streamKey, + "+", + "-", + "COUNT", + 1 ); + const cursor = lastMessage[0]?.[0] ?? "0-0"; + sendConnectedEvent(cursor); + return; + } + + const startId = `(${last_ack}`; + const missingMessages = await redis.xrange(streamKey, startId, "+"); + + sendConnectedEvent(); + + for (const entry of missingMessages) { + const parsed = parseStreamEntry(entry); + if (!parsed) { + continue; + } + + const userEvent: UserEvent = { + data: parsed.data, + __event_path: parsed.__event_path, + __stream_id: parsed.id, + }; + + await sendUserEvent(userEvent); } }; - onError = (err: Error) => { + await replayMissedMessages(); + + subscriber.on("error", (err: Error) => { logger.error("⚠️ Redis subscriber error:", err); const errorEvent: SystemEvent = { @@ -175,9 +206,9 @@ export function handle(config: { }; safeEnqueue(json(errorEvent)); - }; + }); - onUnsubscribe = async () => { + subscriber.on("unsubscribe", () => { logger.log("⬅️ Client unsubscribed from channel:", channel); const unsubscribedEvent: SystemEvent = { @@ -186,82 +217,147 @@ export function handle(config: { }; safeEnqueue(json(unsubscribedEvent)); + }); - await cleanup?.(); - }; - - subscriber.on("subscribe", onSubscribe); - subscriber.on("error", onError); - subscriber.on("unsubscribe", onUnsubscribe); - subscriber.on( - "message", - async ({ - message, - channel: messageChannel, - }: { - message: unknown; - channel: string; - }) => { - let payload: Record; - if (typeof message === "string") { - try { - payload = JSON.parse(message); - } catch { - payload = { data: message }; - } - } else if (typeof message === "object" && message !== null) { - payload = message as Record; - } else { - payload = { data: message }; - } + subscriber.on("message", async (_channel, message) => { + let parsed: PublishedPayload | null = null; + try { + parsed = parsePublishedPayload(message); + } catch (error) { + logger.error("⚠️ Failed to parse realtime payload", error); + } - if (payload.type === "ping") { - const pingEvent: SystemEvent = { - type: "ping", - timestamp: payload.timestamp as number, - }; - safeEnqueue(json(pingEvent)); - return; - } + if (!parsed) { + return; + } - const { __stream_id, __event_path, data } = payload; + if (parsed.type === "ping") { + const pingEvent: SystemEvent = { + type: "ping", + timestamp: parsed.timestamp, + }; + safeEnqueue(json(pingEvent)); + return; + } - logger.log("⬇️ Received event:", { - channel: messageChannel, - __event_path, - data, - }); + const userEvent: UserEvent = { + data: parsed.data, + __event_path: parsed.__event_path, + __stream_id: parsed.__stream_id, + }; - const userEvent: UserEvent = { - data, - __event_path: __event_path as string[], - __stream_id: __stream_id as string, - }; + logger.log("⬇️ Received event:", { + channel: _channel, + __event_path: parsed.__event_path, + data: parsed.data, + }); - safeEnqueue(json(userEvent)); - } - ); + await sendUserEvent(userEvent); + }); keepaliveInterval = setInterval(async () => { - await redis.publish(`channel:${channel}`, { - type: "ping", - timestamp: Date.now(), - }); + await redis.publish( + streamKey, + JSON.stringify({ type: "ping", timestamp: Date.now() }) + ); }, 10_000); }, async cancel() { - if (isClosed) { - return; - } await cleanup?.(); }, }); - return new StreamingResponse(stream); + const headers = config.responseHeaders + ? config.responseHeaders({ request, channel }) + : undefined; + + return new StreamingResponse(stream, { + headers, + }); }; } +type PublishedPayload = + | ({ type: "ping"; timestamp: number } & Record) + | ({ + __event_path: string[]; + __stream_id: string; + data: unknown; + } & Record); + +function parsePublishedPayload(message: unknown): PublishedPayload | null { + if (typeof message !== "string") { + return null; + } + + const parsed = JSON.parse(message) as PublishedPayload | null; + + if (!parsed) { + return null; + } + + if ((parsed as { type?: string }).type === "ping") { + return parsed; + } + + if ( + "__event_path" in parsed && + Array.isArray(parsed.__event_path) && + "data" in parsed && + "__stream_id" in parsed + ) { + return parsed; + } + + return null; +} + +type StreamEntry = [string, string[]]; + +type ParsedStreamEntry = { + id: string; + data: unknown; + __event_path: string[]; +}; + +function parseStreamEntry(entry: StreamEntry): ParsedStreamEntry | null { + const [id, fields] = entry; + const record: Record = {}; + + for (let i = 0; i < fields.length; i += 2) { + const key = fields[i]; + const value = fields[i + 1]; + + if (typeof key !== "string") { + continue; + } + + record[key] = typeof value === "string" ? value : String(value ?? ""); + } + + const raw = record.payload ?? null; + if (!raw) { + return null; + } + + try { + const parsed = JSON.parse(raw) as { + data: unknown; + __event_path: string[]; + }; + + return { + id, + data: parsed.data, + __event_path: parsed.__event_path, + }; + } catch (error) { + console.warn("Failed to parse stream entry", { error }); + return null; + } +} + function json(data: SystemEvent | UserEvent) { return new TextEncoder().encode(`data: ${JSON.stringify(data)}\n\n`); } @@ -277,7 +373,7 @@ class StreamingResponse extends Response { Connection: "keep-alive", "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Headers": "Cache-Control", - ...init?.headers, + ...(init?.headers ?? {}), }, }); } diff --git a/packages/realtime/src/server/realtime.ts b/packages/realtime/src/server/realtime.ts index 6686187e4..da74029c6 100644 --- a/packages/realtime/src/server/realtime.ts +++ b/packages/realtime/src/server/realtime.ts @@ -1,7 +1,7 @@ /** biome-ignore-all lint/style/useBlockStatements: ok */ /** biome-ignore-all lint/style/useConsistentMemberAccessibility: ok */ /** biome-ignore-all lint/suspicious/noExplicitAny: ok */ -import type { Redis } from "@upstash/redis"; +import type { Redis } from "ioredis"; import * as z from "zod/v4/core"; const DEFAULT_VERCEL_FLUID_TIMEOUT = 300; @@ -66,10 +66,10 @@ class RealtimeBase { return; } - const payload = { + const streamPayload = JSON.stringify({ data, __event_path: [outerKey, innerKey], - }; + }); this._logger.log("⬆️ Emitting event:", { channel, @@ -79,18 +79,22 @@ class RealtimeBase { const id = await this._redis.xadd( `channel:${channel}`, + "MAXLEN", + "~", + "100", "*", - payload, - { - trim: { type: "MAXLEN", threshold: 100, comparison: "~" }, - } + "payload", + streamPayload ); - await this._redis.publish(`channel:${channel}`, { - data, - __event_path: [outerKey, innerKey], - __stream_id: id, - }); + await this._redis.publish( + `channel:${channel}`, + JSON.stringify({ + data, + __event_path: [outerKey, innerKey], + __stream_id: id, + }) + ); }, }; } From 0ca3b5bd95be89a64d635970f625239a6020be50 Mon Sep 17 00:00:00 2001 From: Rieranthony Date: Thu, 9 Oct 2025 15:26:37 +0300 Subject: [PATCH 2/2] feat: fix tsup issue --- packages/realtime/tsdown.config.ts | 43 +++++++++++++++++++++++------- 1 file changed, 33 insertions(+), 10 deletions(-) diff --git a/packages/realtime/tsdown.config.ts b/packages/realtime/tsdown.config.ts index fccd336b4..bbf396174 100644 --- a/packages/realtime/tsdown.config.ts +++ b/packages/realtime/tsdown.config.ts @@ -1,13 +1,36 @@ -import { defineConfig } from "tsup" +import { defineConfig } from "tsdown"; export default defineConfig({ - entry: ["src/client", "src/server"], - format: ["cjs", "esm"], - splitting: false, - clean: true, - bundle: false, - dts: true, + entry: ["src/**/*.{ts,tsx}"], + platform: "browser", + sourcemap: true, + minify: false, treeshake: true, - minify: true, - sourcemap: false, -}) + dts: true, + // external: [ + // // React ecosystem (peer dependencies) + // "react", + // "react-dom", + // "react/jsx-runtime", + + // // Peer dependencies + // "@tanstack/react-query", + // "motion", + // "motion/*", + // "tailwindcss", + + // // Internal workspace packages that should be bundled + // // Remove "@cossistant/*" to bundle core and types + // "@cossistant/core", + // "@cossistant/types", + + // // Regular dependencies that should stay external + // "react-use-websocket", + // "react-markdown", + // "zustand", + // "zustand/middleware", + // "nanoid", + // "ulid", + // "tailwind-merge", + // ], +});