diff --git a/dev-test.rar b/dev-test.rar new file mode 100644 index 00000000..ca169592 Binary files /dev/null and b/dev-test.rar differ diff --git a/package-lock.json b/package-lock.json index 46a5af0e..c0e0dbe7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -18,6 +18,7 @@ "@react-email/components": "^0.0.42", "@t3-oss/env-nextjs": "^0.12.0", "@tanstack/react-query": "^5.69.0", + "@tanstack/react-query-devtools": "^5.80.3", "@trpc/client": "^11.0.0", "@trpc/react-query": "^11.0.0", "@trpc/server": "^11.0.0", @@ -28,6 +29,7 @@ "clsx": "^2.1.1", "i18next": "^25.2.1", "i18next-resources-to-backend": "^1.2.1", + "ioredis": "^5.7.0", "lucide-react": "^0.514.0", "meilisearch": "^0.51.0", "next": "15.3.4", @@ -1650,6 +1652,12 @@ "url": "https://opencollective.com/libvips" } }, + "node_modules/@ioredis/commands": { + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/@ioredis/commands/-/commands-1.3.0.tgz", + "integrity": "sha512-M/T6Zewn7sDaBQEqIZ8Rb+i9y8qfGmq+5SDFSf9sA2lUZTmdDLVdOiQaeDp+Q4wElZ9HG1GAX5KhDaidp6LQsQ==", + "license": "MIT" + }, "node_modules/@isaacs/fs-minipass": { "version": "4.0.1", "resolved": "https://registry.npmjs.org/@isaacs/fs-minipass/-/fs-minipass-4.0.1.tgz", @@ -3128,9 +3136,19 @@ } }, "node_modules/@tanstack/query-core": { - "version": "5.81.2", - "resolved": "https://registry.npmjs.org/@tanstack/query-core/-/query-core-5.81.2.tgz", - "integrity": "sha512-QLYkPdrudoMATDFa3MiLEwRhNnAlzHWDf0LKaXUqJd0/+QxN8uTPi7bahRlxoAyH0UbLMBdeDbYzWALj7THOtw==", + "version": "5.83.1", + "resolved": "https://registry.npmjs.org/@tanstack/query-core/-/query-core-5.83.1.tgz", + "integrity": "sha512-OG69LQgT7jSp+5pPuCfzltq/+7l2xoweggjme9vlbCPa/d7D7zaqv5vN/S82SzSYZ4EDLTxNO1PWrv49RAS64Q==", + "license": "MIT", + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + } + }, + "node_modules/@tanstack/query-devtools": { + "version": "5.84.0", + "resolved": "https://registry.npmjs.org/@tanstack/query-devtools/-/query-devtools-5.84.0.tgz", + "integrity": "sha512-fbF3n+z1rqhvd9EoGp5knHkv3p5B2Zml1yNRjh7sNXklngYI5RVIWUrUjZ1RIcEoscarUb0+bOvIs5x9dwzOXQ==", "license": "MIT", "funding": { "type": "github", @@ -3138,12 +3156,12 @@ } }, "node_modules/@tanstack/react-query": { - "version": "5.81.2", - "resolved": "https://registry.npmjs.org/@tanstack/react-query/-/react-query-5.81.2.tgz", - "integrity": "sha512-pe8kFlTrL2zFLlcAj2kZk9UaYYHDk9/1hg9EBaoO3cxDhOZf1FRGJeziSXKrVZyxIfs7b3aoOj/bw7Lie0mDUg==", + "version": "5.84.2", + "resolved": "https://registry.npmjs.org/@tanstack/react-query/-/react-query-5.84.2.tgz", + "integrity": "sha512-cZadySzROlD2+o8zIfbD978p0IphuQzRWiiH3I2ugnTmz4jbjc0+TdibpwqxlzynEen8OulgAg+rzdNF37s7XQ==", "license": "MIT", "dependencies": { - "@tanstack/query-core": "5.81.2" + "@tanstack/query-core": "5.83.1" }, "funding": { "type": "github", @@ -3153,6 +3171,23 @@ "react": "^18 || ^19" } }, + "node_modules/@tanstack/react-query-devtools": { + "version": "5.84.2", + "resolved": "https://registry.npmjs.org/@tanstack/react-query-devtools/-/react-query-devtools-5.84.2.tgz", + "integrity": "sha512-ojJ66QoW9noqK35Lsmfqpfucj6wuOxLL2TYwEwpvU+iUQ5R/7TKpapWvpy9kZyNSl0mxv5mpS+ImfR8aL8/x3g==", + "license": "MIT", + "dependencies": { + "@tanstack/query-devtools": "5.84.0" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + }, + "peerDependencies": { + "@tanstack/react-query": "^5.84.2", + "react": "^18 || ^19" + } + }, "node_modules/@testing-library/dom": { "version": "10.4.0", "resolved": "https://registry.npmjs.org/@testing-library/dom/-/dom-10.4.0.tgz", @@ -4874,6 +4909,15 @@ "node": ">=6" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz", + "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/color": { "version": "4.2.3", "resolved": "https://registry.npmjs.org/color/-/color-4.2.3.tgz", @@ -5114,7 +5158,6 @@ "version": "4.4.1", "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.1.tgz", "integrity": "sha512-KcKCqiftBJcZr++7ykoDIEwSa3XWowTfNPo92BYxjXiyYEVrUQh2aLyhxBCwww+heortUFxEJYcRzosstTEBYQ==", - "dev": true, "license": "MIT", "dependencies": { "ms": "^2.1.3" @@ -5206,6 +5249,15 @@ "node": ">=0.4.0" } }, + "node_modules/denque": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz", + "integrity": "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10" + } + }, "node_modules/dequal": { "version": "2.0.3", "resolved": "https://registry.npmjs.org/dequal/-/dequal-2.0.3.tgz", @@ -6839,6 +6891,30 @@ "node": ">= 0.4" } }, + "node_modules/ioredis": { + "version": "5.7.0", + "resolved": "https://registry.npmjs.org/ioredis/-/ioredis-5.7.0.tgz", + "integrity": "sha512-NUcA93i1lukyXU+riqEyPtSEkyFq8tX90uL659J+qpCZ3rEdViB/APC58oAhIh3+bJln2hzdlZbBZsGNrlsR8g==", + "license": "MIT", + "dependencies": { + "@ioredis/commands": "^1.3.0", + "cluster-key-slot": "^1.1.0", + "debug": "^4.3.4", + "denque": "^2.1.0", + "lodash.defaults": "^4.2.0", + "lodash.isarguments": "^3.1.0", + "redis-errors": "^1.2.0", + "redis-parser": "^3.0.0", + "standard-as-callback": "^2.1.0" + }, + "engines": { + "node": ">=12.22.0" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/ioredis" + } + }, "node_modules/is-array-buffer": { "version": "3.0.5", "resolved": "https://registry.npmjs.org/is-array-buffer/-/is-array-buffer-3.0.5.tgz", @@ -7858,6 +7934,18 @@ "dev": true, "license": "MIT" }, + "node_modules/lodash.defaults": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/lodash.defaults/-/lodash.defaults-4.2.0.tgz", + "integrity": "sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==", + "license": "MIT" + }, + "node_modules/lodash.isarguments": { + "version": "3.1.0", + "resolved": "https://registry.npmjs.org/lodash.isarguments/-/lodash.isarguments-3.1.0.tgz", + "integrity": "sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==", + "license": "MIT" + }, "node_modules/lodash.merge": { "version": "4.6.2", "resolved": "https://registry.npmjs.org/lodash.merge/-/lodash.merge-4.6.2.tgz", @@ -9234,6 +9322,27 @@ "node": ">=8" } }, + "node_modules/redis-errors": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz", + "integrity": "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/redis-parser": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-3.0.0.tgz", + "integrity": "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==", + "license": "MIT", + "dependencies": { + "redis-errors": "^1.0.0" + }, + "engines": { + "node": ">=4" + } + }, "node_modules/reflect.getprototypeof": { "version": "1.0.10", "resolved": "https://registry.npmjs.org/reflect.getprototypeof/-/reflect.getprototypeof-1.0.10.tgz", @@ -9848,6 +9957,12 @@ "dev": true, "license": "MIT" }, + "node_modules/standard-as-callback": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/standard-as-callback/-/standard-as-callback-2.1.0.tgz", + "integrity": "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==", + "license": "MIT" + }, "node_modules/std-env": { "version": "3.9.0", "resolved": "https://registry.npmjs.org/std-env/-/std-env-3.9.0.tgz", diff --git a/package.json b/package.json index 0c65b9d8..36a2b610 100644 --- a/package.json +++ b/package.json @@ -49,6 +49,7 @@ "clsx": "^2.1.1", "i18next": "^25.2.1", "i18next-resources-to-backend": "^1.2.1", + "ioredis": "^5.7.0", "lucide-react": "^0.514.0", "meilisearch": "^0.51.0", "next": "15.3.4", diff --git a/src/app/(protected)/sse-demo/page.tsx b/src/app/(protected)/sse-demo/page.tsx new file mode 100644 index 00000000..9804aaec --- /dev/null +++ b/src/app/(protected)/sse-demo/page.tsx @@ -0,0 +1,476 @@ +"use client"; +import { useEffect, useRef, useState } from "react"; +import { api } from "@/trpc/react"; + +interface SseMessage { + event: string; + data: any; + timestamp: number; +} + +interface ConnectionInfo { + connId: string; + userId?: string; + sessionId?: string; + topics: string[]; + connectedAt: number; + lastActivity: number; + userAgent?: string; + ip?: string; +} + +export default function SseDemo() { + const [messages, setMessages] = useState([]); + const [connectionStatus, setConnectionStatus] = useState<'connecting' | 'connected' | 'disconnected' | 'error'>('disconnected'); + const [connectionInfo, setConnectionInfo] = useState(null); + const [stats, setStats] = useState(null); + const [testMessage, setTestMessage] = useState(''); + const [testTopic, setTestTopic] = useState('test-topic'); + const [selectedEvent, setSelectedEvent] = useState('test_message'); + const esRef = useRef(null); + + // SSE connection management + useEffect(() => { + const connectSSE = () => { + setConnectionStatus('connecting'); + const es = new EventSource("/api/sse?topic=test-topic,notifications,updates", { + withCredentials: true + }); + + es.addEventListener("open", () => { + setConnectionStatus('connected'); + console.log("SSE connection opened"); + }); + + es.addEventListener("message", (e) => { + const message: SseMessage = { + event: 'message', + data: e.data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); // Keep last 50 messages + }); + + es.addEventListener("welcome", (e) => { + const data = JSON.parse(e.data); + setConnectionInfo(data); + const message: SseMessage = { + event: 'welcome', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("test_message", (e) => { + const data = JSON.parse(e.data); + const message: SseMessage = { + event: 'test_message', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("topic_message", (e) => { + const data = JSON.parse(e.data); + const message: SseMessage = { + event: 'topic_message', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("broadcast_message", (e) => { + const data = JSON.parse(e.data); + const message: SseMessage = { + event: 'broadcast_message', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("new_message", (e) => { + const data = JSON.parse(e.data); + const message: SseMessage = { + event: 'new_message', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("system_update", (e) => { + const data = JSON.parse(e.data); + const message: SseMessage = { + event: 'system_update', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("live_update", (e) => { + const data = JSON.parse(e.data); + const message: SseMessage = { + event: 'live_update', + data, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("error", (e) => { + setConnectionStatus('error'); + console.error("SSE connection error:", e); + const message: SseMessage = { + event: 'error', + data: { message: 'Connection error occurred' }, + timestamp: Date.now() + }; + setMessages(prev => [message, ...prev.slice(0, 49)]); + }); + + es.addEventListener("close", () => { + setConnectionStatus('disconnected'); + console.log("SSE connection closed"); + }); + + esRef.current = es; + }; + + connectSSE(); + + return () => { + if (esRef.current) { + esRef.current.close(); + } + }; + }, []); + + // tRPC mutations + const sendTestToUser = api.sse.sendTestToUser.useMutation(); + const sendTestToTopic = api.sse.sendTestToTopic.useMutation(); + const sendTestBroadcast = api.sse.sendTestBroadcast.useMutation(); + const sendTestToSession = api.sse.sendTestToSession.useMutation(); + const sendNewMessage = api.sse.sendNewMessage.useMutation(); + const sendSystemUpdate = api.sse.sendSystemUpdate.useMutation(); + const sendLiveUpdate = api.sse.sendLiveUpdate.useMutation(); + const getStats = api.sse.getStats.useQuery(undefined, { + refetchInterval: 5000, // Refresh every 5 seconds + }); + const getMyConnections = api.sse.getMyConnections.useQuery(undefined, { + refetchInterval: 3000, // Refresh every 3 seconds + }); + + const handleSendTestToUser = () => { + sendTestToUser.mutate({ + message: testMessage || "Hello from user test!", + event: selectedEvent + }); + }; + + const handleSendTestToSession = () => { + const sessionId = + connectionInfo?.sessionId || + (getMyConnections.data && (getMyConnections.data as any[])[0]?.sessionId) || + undefined; + + if (!sessionId) { + alert("No sessionId available. Open the SSE connection first."); + return; + } + + sendTestToSession.mutate({ + sessionId, + message: testMessage || "Hello from session test!", + event: selectedEvent, + }); + }; + + const handleSendTestToTopic = () => { + sendTestToTopic.mutate({ + topic: testTopic, + message: testMessage || "Hello from topic test!", + event: selectedEvent + }); + }; + + const handleSendBroadcast = () => { + sendTestBroadcast.mutate({ + message: testMessage || "Hello everyone!", + event: selectedEvent + }); + }; + + const handleSendNewMessage = () => { + sendNewMessage.mutate({ + userId: connectionInfo?.userId || "test-user", + message: { + id: `msg-${Date.now()}`, + content: testMessage || "This is a new message!", + sender: "Demo User", + roomId: "demo-room" + } + }); + }; + + const handleSendSystemUpdate = () => { + sendSystemUpdate.mutate({ + event: "demo_update", + data: { + message: testMessage || "System update from demo", + version: "1.0.0", + timestamp: Date.now() + } + }); + }; + + const handleSendLiveUpdate = () => { + sendLiveUpdate.mutate({ + topic: testTopic, + data: { + message: testMessage || "Live update from demo", + value: Math.random() * 100, + timestamp: Date.now() + } + }); + }; + + const formatTimestamp = (timestamp: number) => { + return new Date(timestamp).toLocaleTimeString(); + }; + + return ( +
+
+

SSE Demo Dashboard

+ + {/* Connection Status */} +
+

Connection Status

+
+
+ {connectionStatus.toUpperCase()} +
+ {connectionInfo && ( +
+ Connection ID: {connectionInfo.connId} + {connectionInfo.sessionId && ( + <> + {" "}| Session: {connectionInfo.sessionId} + + )} +
+ )} +
+
+ + {/* Statistics */} + {stats && ( +
+

SSE Statistics

+
+
+
{stats.activeConnections}
+
Active Connections
+
+
+
{stats.totalEventsSent}
+
Events Sent
+
+
+
{stats.totalErrors}
+
Errors
+
+
+
{Math.floor(stats.uptime / 1000)}s
+
Uptime
+
+
+
+ )} + + {/* Test Controls */} +
+

Test Controls

+ +
+
+ + +
+ +
+ + setTestTopic(e.target.value)} + className="w-full px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-blue-500" + placeholder="Enter topic name" + /> +
+
+ +
+ + setTestMessage(e.target.value)} + className="w-full px-3 py-2 border border-gray-300 rounded-md focus:outline-none focus:ring-2 focus:ring-blue-500" + placeholder="Enter your message" + /> +
+ +
+ + + + + + + + + + + + + +
+
+ + {/* Messages */} +
+

Received Messages ({messages.length})

+
+ {messages.length === 0 ? ( +
No messages received yet
+ ) : ( +
+ {messages.map((message, index) => ( +
+
+ + {message.event} + + + {formatTimestamp(message.timestamp)} + +
+ 
+                      {JSON.stringify(message.data, null, 2)}
+
+
+ ))} +
+ )} +
+
+ + {/* My Connections */} + {getMyConnections.data && getMyConnections.data.length > 0 && ( +
+

My Connections ({getMyConnections.data.length})

+
+
+ {getMyConnections.data.map((conn: ConnectionInfo, index: number) => ( +
+
+
+
Connection {index + 1}
+
ID: {conn.connId}
+
+ Topics: {conn.topics.join(', ') || 'None'} +
+
+
+
+ Connected: {formatTimestamp(conn.connectedAt)} +
+
+ Last Activity: {formatTimestamp(conn.lastActivity)} +
+
+
+
+ ))} +
+
+
+ )} +
+
+ ); +} diff --git a/src/app/(public)/page.tsx b/src/app/(public)/page.tsx index 9726c0ed..7bcbed76 100644 --- a/src/app/(public)/page.tsx +++ b/src/app/(public)/page.tsx @@ -14,12 +14,19 @@ const LandingPage = () => ( - - {"Sign in"} - +
+ + {"Sign in"} + + +
+

🚀 SSE Testing Ready!

+

Use username login to test multiple profiles and real-time notifications

+
+
); diff --git a/src/app/(public)/signin/page.tsx b/src/app/(public)/signin/page.tsx new file mode 100644 index 00000000..004d374d --- /dev/null +++ b/src/app/(public)/signin/page.tsx @@ -0,0 +1,156 @@ +"use client"; + +import { useState } from "react"; +import { signIn, getProviders } from "next-auth/react"; +import { useRouter } from "next/navigation"; + +export default function SignInPage() { + const [username, setUsername] = useState(""); + const [password, setPassword] = useState(""); + const [isLoading, setIsLoading] = useState(false); + const [error, setError] = useState(""); + const router = useRouter(); + + const handleUsernameLogin = async (e: React.FormEvent) => { + e.preventDefault(); + setIsLoading(true); + setError(""); + + try { + const result = await signIn("username", { + username, + password, + redirect: false, + }); + + if (result?.error) { + setError("Invalid credentials. Use any username with password 'test123'"); + } else if (result?.ok) { + router.push("/home"); + } + } catch (error) { + setError("An error occurred during sign in"); + } finally { + setIsLoading(false); + } + }; + + const handleDiscordLogin = async () => { + setIsLoading(true); + setError(""); + + try { + await signIn("discord", { callbackUrl: "/home" }); + } catch (error) { + setError("Failed to sign in with Discord"); + setIsLoading(false); + } + }; + + return ( +
+
+
+

+ Sign in to your account +

+

+ Test with multiple profiles using username login +

+
+ +
+ {/* Username Login Form */} +
+
+
+ +
+ setUsername(e.target.value)} + className="appearance-none block w-full px-3 py-2 border border-gray-300 rounded-md placeholder-gray-400 focus:outline-none focus:ring-blue-500 focus:border-blue-500 sm:text-sm" + placeholder="Enter any username" + /> +
+
+ +
+ +
+ setPassword(e.target.value)} + className="appearance-none block w-full px-3 py-2 border border-gray-300 rounded-md placeholder-gray-400 focus:outline-none focus:ring-blue-500 focus:border-blue-500 sm:text-sm" + placeholder="Use 'test123' for any username" + /> +
+
+ + {error && ( +
{error}
+ )} + +
+ +
+
+ +
+
+
+
+
+
+ Or continue with +
+
+ +
+ +
+
+
+ + {/* Testing Instructions */} +
+

Testing Instructions

+
    +
  • â€ĸ Use any username with password "test123"
    • +
  • â€ĸ Open multiple Chrome profiles to test different users
    • +
  • â€ĸ Each profile will have a unique session for SSE testing
    • +
  • â€ĸ After testing, we'll remove this simple login system
    • +
+
+
+
+
+ ); +} diff --git a/src/app/api/auth/[...nextauth]/route.ts b/src/app/api/auth/[...nextauth]/route.ts index 789f7d22..11132817 100644 --- a/src/app/api/auth/[...nextauth]/route.ts +++ b/src/app/api/auth/[...nextauth]/route.ts @@ -1,3 +1,5 @@ import { handlers } from "@/features/auth"; +// src/app/api/auth/[...nextauth]/route.ts + export const { GET, POST } = handlers; diff --git a/src/app/api/sse/route.ts b/src/app/api/sse/route.ts new file mode 100644 index 00000000..b5d83df2 --- /dev/null +++ b/src/app/api/sse/route.ts @@ -0,0 +1,160 @@ +import { NextRequest } from "next/server"; +import { auth as getServerSession } from "@/features/auth/handlers"; +import { sse as manager } from "@/server/sse"; +import { randomUUID } from "crypto"; +import { log } from "@/lib/logger"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +function sseLine(evt: { event?: string; data?: any; id?: string; retry?: number }) { + let out = ""; + if (evt.id) out += `id: ${evt.id}\n`; + if (evt.retry) out += `retry: ${evt.retry}\n`; + if (evt.event) out += `event: ${evt.event}\n`; + if (evt.data !== undefined) out += `data: ${typeof evt.data === "string" ? evt.data : JSON.stringify(evt.data)}\n`; + return out + "\n"; +} + +export async function GET(req: NextRequest) { + try { + const session = await getServerSession(); + if (!session?.user?.id) { + log.warn("Unauthorized SSE connection attempt"); + return new Response("Unauthorized", { status: 401 }); + } + + const { searchParams } = new URL(req.url); + const topics = (searchParams.get("topic") ?? "") + .split(",").map(s => s.trim()).filter(Boolean); + + const connId = randomUUID(); + const sessionId = (session as any).sessionToken as string | undefined; + const userAgent = req.headers.get("user-agent") || undefined; + const ip = req.headers.get("x-forwarded-for") || req.headers.get("x-real-ip") || "unknown"; + + log.info("SSE connection request", { + connId, + userId: session.user.id, + sessionId, + topics, + userAgent, + ip + }); + + const stream = new ReadableStream({ + start: async (controller) => { + const enc = new TextEncoder(); + const write = (s: string) => { + try { + controller.enqueue(enc.encode(s)); + } catch (error) { + log.error("Failed to write to SSE stream", { connId, error }); + } + }; + + // Send initial connection setup + write(sseLine({ retry: 5000 })); + + try { + const unregister = await manager.register({ + connId, + userId: session.user.id, + sessionId, + topics, + userAgent, + ip, + send: (evt) => { + try { + write(sseLine({ + event: evt.event, + data: evt.data, + id: evt.id + })); + } catch (error) { + log.error("Failed to send SSE event to client", { + connId, + event: evt.event, + error + }); + } + }, + }); + + // Send welcome message + write(sseLine({ + event: "welcome", + data: { + connId, + userId: session.user.id, + sessionId, + timestamp: Date.now() + } + })); + + // Set up heartbeat + const heartbeatInterval = setInterval(() => { + try { + write(`: heartbeat ${Date.now()}\n\n`); + } catch (error) { + log.error("Failed to send heartbeat", { connId, error }); + } + }, 30000); + + // Handle connection cleanup + const cleanup = async () => { + clearInterval(heartbeatInterval); + await unregister(); + try { + controller.close(); + } catch (error) { + log.error("Error closing SSE stream", { connId, error }); + } + }; + + // Listen for abort signal + req.signal.addEventListener("abort", cleanup); + + // Handle stream errors + req.signal.addEventListener("error", (error) => { + log.error("SSE stream error", { connId, error }); + cleanup(); + }); + + } catch (error) { + log.error("Failed to register SSE connection", { connId, error }); + write(sseLine({ + event: "error", + data: { + message: "Failed to establish connection", + code: "CONNECTION_ERROR" + } + })); + controller.close(); + } + }, + cancel: async () => { + try { + await manager.unregister(connId); + log.info("SSE connection cancelled", { connId }); + } catch (error) { + log.error("Error during SSE connection cancellation", { connId, error }); + } + }, + }); + + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Headers": "Cache-Control", + }, + }); + } catch (error) { + log.error("SSE route error", { error }); + return new Response("Internal Server Error", { status: 500 }); + } +} diff --git a/src/app/api/webhooks/sse-example/route.ts b/src/app/api/webhooks/sse-example/route.ts new file mode 100644 index 00000000..51eaaccd --- /dev/null +++ b/src/app/api/webhooks/sse-example/route.ts @@ -0,0 +1,74 @@ +import { NextRequest } from "next/server"; +import { notifyUser, notifyTopic, broadcast } from "@/server/sse"; +import { log } from "@/lib/logger"; + +export async function POST(req: NextRequest) { + try { + const payload = await req.json(); + + log.info("SSE Example webhook received", { + type: payload.type, + userId: payload.userId + }); + + // Example: Process different webhook types + switch (payload.type) { + case "user_notification": + // Send notification to specific user + if (payload.userId) { + await notifyUser(payload.userId, "webhook_notification", { + type: payload.type, + message: payload.message, + timestamp: Date.now(), + source: "webhook" + }); + } + break; + + case "system_alert": + // Broadcast system alert to all users + await broadcast("system_alert", { + level: payload.level || "info", + message: payload.message, + timestamp: Date.now(), + source: "webhook" + }); + break; + + case "topic_update": + // Send update to topic subscribers + if (payload.topic) { + await notifyTopic(payload.topic, "topic_update", { + data: payload.data, + timestamp: Date.now(), + source: "webhook" + }); + } + break; + + case "live_data": + // Send live data updates + await notifyTopic("live-data", "data_update", { + data: payload.data, + timestamp: Date.now(), + source: "webhook" + }); + break; + + default: + log.warn("Unknown webhook type", { type: payload.type }); + } + + return Response.json({ + success: true, + message: "Webhook processed and SSE notifications sent" + }); + + } catch (error) { + log.error("Error processing SSE example webhook", { error }); + return Response.json( + { success: false, error: "Failed to process webhook" }, + { status: 500 } + ); + } +} diff --git a/src/app/providers.tsx b/src/app/providers.tsx new file mode 100644 index 00000000..92e4d6b0 --- /dev/null +++ b/src/app/providers.tsx @@ -0,0 +1,73 @@ +'use client'; + +import type { QueryClient } from '@tanstack/react-query'; +import { QueryClientProvider } from '@tanstack/react-query'; +import { ReactQueryDevtools } from '@tanstack/react-query-devtools'; +import { + httpBatchStreamLink, + httpSubscriptionLink, + loggerLink, + splitLink, +} from '@trpc/client'; +import { createQueryClient } from '..//lib/query-client'; +import { trpc } from '../lib/trpc'; +import { useState } from 'react'; +import SuperJSON from 'superjson'; + +let clientQueryClientSingleton: QueryClient | undefined = undefined; +const getQueryClient = () => { + if (typeof window === 'undefined') { + // Server: always make a new query client + return createQueryClient(); + } else { + // Browser: use singleton pattern to keep the same query client + return (clientQueryClientSingleton ??= createQueryClient()); + } +}; + +const getUrl = () => { + const base = (() => { + if (typeof window !== 'undefined') return window.location.origin; + if (process.env.APP_URL) return process.env.APP_URL; + return `http://localhost:${process.env.PORT ?? 3000}`; + })(); + + return `${base}/api/trpc`; +}; + +export function TRPCProviders(props: Readonly<{ children: React.ReactNode }>) { + const queryClient = getQueryClient(); + const [trpcClient] = useState(() => + trpc.createClient({ + links: [ + // adds pretty logs to your console in development and logs errors in production + loggerLink(), + splitLink({ + condition: (op) => op.type === 'subscription', + true: httpSubscriptionLink({ + url: getUrl(), + /** + * @see https://trpc.io/docs/v11/data-transformers + */ + transformer: SuperJSON, + }), + false: httpBatchStreamLink({ + url: getUrl(), + /** + * @see https://trpc.io/docs/v11/data-transformers + */ + transformer: SuperJSON, + }), + }), + ], + }), + ); + return ( + + + {props.children} + + + + ); +} diff --git a/src/config/routes.ts b/src/config/routes.ts index 8908c756..347dee64 100644 --- a/src/config/routes.ts +++ b/src/config/routes.ts @@ -29,8 +29,10 @@ import type { RouteData } from "@/types/routes"; // Ensure every route in the app has an entry here export const paths = { landingPage: "/", + signinPage: "/signin", homePage: "/home", reelsUploadPage: "/reels/upload", + sseDemoPage: "/sse-demo", } as const; // ⚠ī¸ DEFINE METADATA FOR NEW ROUTES HERE ⚠ī¸ @@ -41,6 +43,11 @@ export const routes: Record = { path: paths.landingPage, accessType: "public", }, + signinPage: { + name: "Sign In Page", + path: paths.signinPage, + accessType: "public", + }, homePage: { name: "Home Page", path: paths.homePage, @@ -51,4 +58,9 @@ export const routes: Record = { path: paths.reelsUploadPage, accessType: "protected", }, + sseDemoPage: { + name: "SSE Demo Page", + path: paths.sseDemoPage, + accessType: "protected", + }, }; diff --git a/src/env.js b/src/env.js index 4564a09f..ddb6ea63 100644 --- a/src/env.js +++ b/src/env.js @@ -11,31 +11,43 @@ export const env = createEnv({ process.env.NODE_ENV === "production" ? z.string() : z.string().optional(), - AUTH_DISCORD_ID: z.string(), - AUTH_DISCORD_SECRET: z.string(), + // Align with Discord provider used in auth config. Optional in dev. + AUTH_DISCORD_ID: + process.env.NODE_ENV === "production" ? z.string() : z.string().optional(), + AUTH_DISCORD_SECRET: + process.env.NODE_ENV === "production" ? z.string() : z.string().optional(), DATABASE_URL: z.string().url(), NODE_ENV: z .enum(["development", "test", "production"]) .default("development"), - // Mux environment variables - MUX_TOKEN_ID: z.string().uuid(), - MUX_TOKEN_SECRET: z.string().length(75), - MUX_SIGNING_KEY_ID: z.string().length(45), - MUX_SIGNING_KEY_SECRET: z - .string() - .min(200) - .refine((val) => val.endsWith("==")), - MUX_VIDEO_QUALITY: z.enum(["basic", "plus", "premium"]), - MUX_WEBHOOK_SECRET: z.string().min(32), + // Mux environment variables (optional in dev) + MUX_TOKEN_ID: + process.env.NODE_ENV === "production" ? z.string().uuid() : z.string().uuid().optional(), + MUX_TOKEN_SECRET: + process.env.NODE_ENV === "production" ? z.string().length(75) : z.string().length(75).optional(), + MUX_SIGNING_KEY_ID: + process.env.NODE_ENV === "production" ? z.string().length(45) : z.string().length(45).optional(), + MUX_SIGNING_KEY_SECRET: + process.env.NODE_ENV === "production" + ? z.string().min(200).refine((val) => val.endsWith("==")) + : z.string().min(200).refine((val) => val.endsWith("==")).optional(), + MUX_VIDEO_QUALITY: + process.env.NODE_ENV === "production" ? z.enum(["basic", "plus", "premium"]) : z.enum(["basic", "plus", "premium"]).optional(), + MUX_WEBHOOK_SECRET: + process.env.NODE_ENV === "production" ? z.string().min(32) : z.string().min(32).optional(), - // Meilisearch environment variables - MEILISEARCH_HOST: z.string().url(), - MEILISEARCH_API_KEY: z.string(), + // Meilisearch environment variables (optional in dev) + MEILISEARCH_HOST: + process.env.NODE_ENV === "production" ? z.string().url() : z.string().url().optional(), + MEILISEARCH_API_KEY: + process.env.NODE_ENV === "production" ? z.string() : z.string().optional(), - // Resend environment variables - RESEND_API_KEY: z.string().length(36), - RESEND_TO_DEV_ADDRESS: z.string().email(), + // Resend environment variables (optional in dev) + RESEND_API_KEY: + process.env.NODE_ENV === "production" ? z.string().length(36) : z.string().length(36).optional(), + RESEND_TO_DEV_ADDRESS: + process.env.NODE_ENV === "production" ? z.string().email() : z.string().email().optional(), RESEND_FROM_EMAIL: z.string().refine((val) => { // The "from" email address can be in the format "Name " or just "my@email.com". // We need to extract the email part and validate it. diff --git a/src/features/auth/config/next-auth.ts b/src/features/auth/config/next-auth.ts index aaf11523..90dc57f2 100644 --- a/src/features/auth/config/next-auth.ts +++ b/src/features/auth/config/next-auth.ts @@ -1,6 +1,8 @@ import { PrismaAdapter } from "@auth/prisma-adapter"; import { type DefaultSession, type NextAuthConfig } from "next-auth"; import DiscordProvider from "next-auth/providers/discord"; +import CredentialsProvider from "next-auth/providers/credentials"; +import { z } from "zod"; import { db } from "@/lib/db"; @@ -30,19 +32,69 @@ declare module "next-auth" { * * @see https://next-auth.js.org/configuration/options */ +const providers: NextAuthConfig["providers"] = []; + +// Always include simple username login for testing multiple profiles +providers.push( + CredentialsProvider({ + id: "username", + name: "Username Login", + credentials: { + username: { label: "Username", type: "text", placeholder: "Enter username" }, + password: { label: "Password", type: "password", placeholder: "Enter password" } + }, + async authorize(input) { + const creds = z + .object({ + username: z.string().min(1), + password: z.string().min(1) + }) + .safeParse(input); + if (!creds.success) return null; + + const { username, password } = creds.data; + + // Simple test credentials - in production, you'd hash and check against database + // For now, accept any username with password "test123" + if (password === "test123") { + return { + id: `user-${username}`, + name: username, + email: `${username}@test.local`, + image: `https://api.dicebear.com/7.x/initials/svg?seed=${username}` + }; + } + + return null; + }, + }) +); + +// Include Discord if env vars are set +const hasDiscordEnv = Boolean(process.env.AUTH_DISCORD_ID && process.env.AUTH_DISCORD_SECRET); +if (hasDiscordEnv) { + providers.push(DiscordProvider); +} else { + // Fallback dev credentials when Discord not configured + providers.push( + CredentialsProvider({ + id: "dev", + name: "Dev Login", + credentials: { name: { label: "Name", type: "text" } }, + async authorize(input) { + const creds = z + .object({ name: z.string().min(1) }) + .safeParse(input); + if (!creds.success) return null; + const name = creds.data.name; + return { id: name, name, email: `${name}@example.test` }; + }, + }) + ); +} + export const nextAuthConfig = { - providers: [ - DiscordProvider, - /** - * ...add more providers here. - * - * Most other providers require a bit more work than the Discord provider. For example, the - * GitHub provider requires you to add the `refresh_token_expires_in` field to the Account - * model. Refer to the NextAuth.js docs for the provider you want to use. Example: - * - * @see https://next-auth.js.org/providers/github - */ - ], + providers, adapter: PrismaAdapter(db), callbacks: { session: ({ session, user }) => ({ diff --git a/src/hooks/useSSE.ts b/src/hooks/useSSE.ts new file mode 100644 index 00000000..8b0d24f3 --- /dev/null +++ b/src/hooks/useSSE.ts @@ -0,0 +1,275 @@ +import { useEffect, useRef, useState, useCallback } from 'react'; + +interface SSEOptions { + topics?: string[]; + onMessage?: (event: string, data: any) => void; + onConnect?: () => void; + onDisconnect?: () => void; + onError?: (error: Event) => void; + autoReconnect?: boolean; + reconnectInterval?: number; + maxReconnectAttempts?: number; +} + +interface SSEState { + status: 'connecting' | 'connected' | 'disconnected' | 'error'; + connectionInfo: any; + error: string | null; + reconnectAttempts: number; +} + +export function useSSE(options: SSEOptions = {}) { + const { + topics = [], + onMessage, + onConnect, + onDisconnect, + onError, + autoReconnect = true, + reconnectInterval = 5000, + maxReconnectAttempts = 5 + } = options; + + const [state, setState] = useState({ + status: 'disconnected', + connectionInfo: null, + error: null, + reconnectAttempts: 0 + }); + + const eventSourceRef = useRef(null); + const reconnectTimeoutRef = useRef(null); + const reconnectAttemptsRef = useRef(0); + + const connect = useCallback(() => { + if (eventSourceRef.current) { + eventSourceRef.current.close(); + } + + setState(prev => ({ ...prev, status: 'connecting', error: null })); + + try { + const topicParam = topics.length > 0 ? `?topic=${topics.join(',')}` : ''; + const eventSource = new EventSource(`/api/sse${topicParam}`, { + withCredentials: true + }); + + eventSource.addEventListener('open', () => { + setState(prev => ({ + ...prev, + status: 'connected', + error: null, + reconnectAttempts: 0 + })); + reconnectAttemptsRef.current = 0; + onConnect?.(); + }); + + eventSource.addEventListener('message', (e) => { + try { + const data = JSON.parse(e.data); + onMessage?.('message', data); + } catch (error) { + onMessage?.('message', e.data); + } + }); + + eventSource.addEventListener('welcome', (e) => { + try { + const data = JSON.parse(e.data); + setState(prev => ({ ...prev, connectionInfo: data })); + onMessage?.('welcome', data); + } catch (error) { + console.error('Failed to parse welcome message:', error); + } + }); + + // Listen for custom events + const customEvents = [ + 'test_message', 'topic_message', 'broadcast_message', + 'new_message', 'system_update', 'live_update', + 'webhook_notification', 'system_alert', 'topic_update', + 'data_update' + ]; + + customEvents.forEach(eventType => { + eventSource.addEventListener(eventType, (e) => { + try { + const data = JSON.parse(e.data); + onMessage?.(eventType, data); + } catch (error) { + onMessage?.(eventType, e.data); + } + }); + }); + + eventSource.addEventListener('error', (e) => { + setState(prev => ({ + ...prev, + status: 'error', + error: 'Connection error occurred' + })); + onError?.(e); + }); + + eventSource.addEventListener('close', () => { + setState(prev => ({ ...prev, status: 'disconnected' })); + onDisconnect?.(); + }); + + eventSourceRef.current = eventSource; + + } catch (error) { + setState(prev => ({ + ...prev, + status: 'error', + error: 'Failed to create connection' + })); + onError?.(error as Event); + } + }, [topics, onMessage, onConnect, onDisconnect, onError]); + + const disconnect = useCallback(() => { + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } + + if (eventSourceRef.current) { + eventSourceRef.current.close(); + eventSourceRef.current = null; + } + + setState(prev => ({ + ...prev, + status: 'disconnected', + error: null, + reconnectAttempts: 0 + })); + reconnectAttemptsRef.current = 0; + }, []); + + const reconnect = useCallback(() => { + if (!autoReconnect || reconnectAttemptsRef.current >= maxReconnectAttempts) { + setState(prev => ({ + ...prev, + status: 'error', + error: 'Max reconnection attempts reached' + })); + return; + } + + reconnectAttemptsRef.current++; + setState(prev => ({ + ...prev, + reconnectAttempts: reconnectAttemptsRef.current + })); + + const delay = reconnectInterval * Math.pow(2, reconnectAttemptsRef.current - 1); + + reconnectTimeoutRef.current = setTimeout(() => { + connect(); + }, delay); + }, [autoReconnect, maxReconnectAttempts, reconnectInterval, connect]); + + // Auto-reconnect on error + useEffect(() => { + if (state.status === 'error' && autoReconnect) { + reconnect(); + } + }, [state.status, autoReconnect, reconnect]); + + // Initial connection + useEffect(() => { + connect(); + + return () => { + disconnect(); + }; + }, [connect, disconnect]); + + // Cleanup on unmount + useEffect(() => { + return () => { + disconnect(); + }; + }, [disconnect]); + + return { + status: state.status, + connectionInfo: state.connectionInfo, + error: state.error, + reconnectAttempts: state.reconnectAttempts, + connect, + disconnect, + reconnect + }; +} + +// Hook for listening to specific events +export function useSSEEvent(eventType: string, callback: (data: any) => void) { + const [lastEvent, setLastEvent] = useState(null); + + const handleMessage = useCallback((event: string, data: any) => { + if (event === eventType) { + setLastEvent(data); + callback(data); + } + }, [eventType, callback]); + + const sse = useSSE({ onMessage: handleMessage }); + + return { + ...sse, + lastEvent + }; +} + +// Hook for user-specific notifications +export function useUserNotifications(userId: string) { + const [notifications, setNotifications] = useState([]); + + const handleMessage = useCallback((event: string, data: any) => { + if (event === 'new_message' || event === 'webhook_notification') { + setNotifications(prev => [data, ...prev.slice(0, 49)]); // Keep last 50 + } + }, []); + + const sse = useSSE({ onMessage: handleMessage }); + + const clearNotifications = useCallback(() => { + setNotifications([]); + }, []); + + return { + ...sse, + notifications, + clearNotifications + }; +} + +// Hook for topic subscriptions +export function useTopicSubscription(topic: string) { + const [updates, setUpdates] = useState([]); + + const handleMessage = useCallback((event: string, data: any) => { + if (event === 'topic_update' || event === 'live_update' || event === 'data_update') { + setUpdates(prev => [data, ...prev.slice(0, 49)]); // Keep last 50 + } + }, []); + + const sse = useSSE({ + topics: [topic], + onMessage: handleMessage + }); + + const clearUpdates = useCallback(() => { + setUpdates([]); + }, []); + + return { + ...sse, + updates, + clearUpdates + }; +} diff --git a/src/lib/logger.ts b/src/lib/logger.ts new file mode 100644 index 00000000..97d0156c --- /dev/null +++ b/src/lib/logger.ts @@ -0,0 +1,6 @@ + +export const log = { + info: console.log, + error: console.error, + warn: console.warn, + }; \ No newline at end of file diff --git a/src/lib/query-client.ts b/src/lib/query-client.ts new file mode 100644 index 00000000..58333eb2 --- /dev/null +++ b/src/lib/query-client.ts @@ -0,0 +1,12 @@ +import { QueryClient } from '@tanstack/react-query'; + +export const createQueryClient = () => + new QueryClient({ + defaultOptions: { + queries: { + // With SSR, we usually want to set some default staleTime + // above 0 to avoid refetching immediately on the client + staleTime: 30 * 1000, + }, + }, + }); diff --git a/src/lib/trpc/root.ts b/src/lib/trpc/root.ts index 129dcabc..a9f27414 100644 --- a/src/lib/trpc/root.ts +++ b/src/lib/trpc/root.ts @@ -1,5 +1,6 @@ import { createCallerFactory, createTRPCRouter } from "@/lib/trpc"; import { searchRouter } from "@/features/search"; +import { sseRouter } from "@/server/api/routers/sse"; /** * This is the primary router for your server. @@ -8,6 +9,7 @@ import { searchRouter } from "@/features/search"; */ export const appRouter = createTRPCRouter({ search: searchRouter, + sse: sseRouter, }); // export type definition of API diff --git a/src/server/api/routers/sse.ts b/src/server/api/routers/sse.ts new file mode 100644 index 00000000..33bea193 --- /dev/null +++ b/src/server/api/routers/sse.ts @@ -0,0 +1,188 @@ +import { createTRPCRouter, protectedProcedure } from "@/lib/trpc"; +import { z } from "zod"; +import { + notifyUser, + broadcast, + notifyTopic, + notifySession, + notifyConnection, + notifyNewMessage, + notifySystemUpdate, + notifyUserStatus, + notifyLiveUpdate, + getStats, + getConnectionInfo, + getUserConnections, + getTopicConnections, + healthCheck +} from "@/server/sse"; + +export const sseRouter = createTRPCRouter({ + // Basic test endpoints + debugSend: protectedProcedure.mutation(async ({ ctx }) => { + await notifyUser(ctx.session.user.id, "test", { ok: true, at: Date.now() }); + await broadcast("test", { msg: "Hello all" }); + return { ok: true }; + }), + + // Send test message to current user + sendTestToUser: protectedProcedure + .input(z.object({ + message: z.string().optional(), + event: z.string().optional() + })) + .mutation(async ({ ctx, input }) => { + const event = input.event || "test_message"; + const message = input.message || "Hello from SSE!"; + + await notifyUser(ctx.session.user.id, event, { + message, + timestamp: Date.now(), + userId: ctx.session.user.id + }); + + return { success: true, event, message }; + }), + + // Send test message to a topic + sendTestToTopic: protectedProcedure + .input(z.object({ + topic: z.string(), + message: z.string().optional(), + event: z.string().optional() + })) + .mutation(async ({ input }) => { + const event = input.event || "topic_message"; + const message = input.message || "Hello from topic!"; + + await notifyTopic(input.topic, event, { + message, + timestamp: Date.now(), + topic: input.topic + }); + + return { success: true, event, message, topic: input.topic }; + }), + + // Send test broadcast + sendTestBroadcast: protectedProcedure + .input(z.object({ + message: z.string().optional(), + event: z.string().optional() + })) + .mutation(async ({ input }) => { + const event = input.event || "broadcast_message"; + const message = input.message || "Hello everyone!"; + + await broadcast(event, { + message, + timestamp: Date.now() + }); + + return { success: true, event, message }; + }), + + // Send test message to specific session + sendTestToSession: protectedProcedure + .input(z.object({ + sessionId: z.string(), + message: z.string().optional(), + event: z.string().optional() + })) + .mutation(async ({ input }) => { + const event = input.event || "session_message"; + const message = input.message || "Hello from session!"; + + await notifySession(input.sessionId, event, { + message, + timestamp: Date.now(), + sessionId: input.sessionId + }); + + return { success: true, event, message, sessionId: input.sessionId }; + }), + + // Utility notification endpoints + sendNewMessage: protectedProcedure + .input(z.object({ + userId: z.string(), + message: z.object({ + id: z.string(), + content: z.string(), + sender: z.string(), + roomId: z.string().optional() + }) + })) + .mutation(async ({ input }) => { + await notifyNewMessage(input.userId, input.message); + return { success: true }; + }), + + sendSystemUpdate: protectedProcedure + .input(z.object({ + event: z.string(), + data: z.any() + })) + .mutation(async ({ input }) => { + await notifySystemUpdate(input.event, input.data); + return { success: true }; + }), + + sendUserStatus: protectedProcedure + .input(z.object({ + userId: z.string(), + status: z.string() + })) + .mutation(async ({ input }) => { + await notifyUserStatus(input.userId, input.status); + return { success: true }; + }), + + sendLiveUpdate: protectedProcedure + .input(z.object({ + topic: z.string(), + data: z.any() + })) + .mutation(async ({ input }) => { + await notifyLiveUpdate(input.topic, input.data); + return { success: true }; + }), + + // Monitoring and management endpoints + getStats: protectedProcedure.query(() => { + return getStats(); + }), + + getHealth: protectedProcedure.query(() => { + return healthCheck(); + }), + + getConnectionInfo: protectedProcedure + .input(z.object({ + connId: z.string() + })) + .query(({ input }) => { + return getConnectionInfo(input.connId); + }), + + getUserConnections: protectedProcedure + .input(z.object({ + userId: z.string() + })) + .query(({ input }) => { + return getUserConnections(input.userId); + }), + + getTopicConnections: protectedProcedure + .input(z.object({ + topic: z.string() + })) + .query(({ input }) => { + return getTopicConnections(input.topic); + }), + + // Get current user's connections + getMyConnections: protectedProcedure.query(({ ctx }) => { + return getUserConnections(ctx.session.user.id); + }), +}); diff --git a/src/server/context.ts b/src/server/context.ts new file mode 100644 index 00000000..3a753441 --- /dev/null +++ b/src/server/context.ts @@ -0,0 +1,18 @@ +import type { FetchCreateContextFnOptions } from '@trpc/server/adapters/fetch'; +import { auth } from './auth'; + +/** + * Creates context for an incoming request + * @see https://trpc.io/docs/v11/context + */ +export const createContext = async (opts: FetchCreateContextFnOptions) => { + const session = await auth(); + + console.log('createContext for', session?.user?.name ?? 'unknown user'); + + return { + session, + }; +}; + +export type Context = Awaited>; diff --git a/src/server/db/client.ts b/src/server/db/client.ts new file mode 100644 index 00000000..3f6b05e5 --- /dev/null +++ b/src/server/db/client.ts @@ -0,0 +1,11 @@ +import { drizzle } from 'drizzle-orm/postgres-js'; +import postgres from 'postgres'; +import * as schema from './schema'; + +const DB_URL = process.env.DATABASE_URL ?? process.env.POSTGRES_URL; +if (!DB_URL) { + throw new Error('Missing POSTGRES_URL or DATABASE_URL environment variable'); +} + +const queryClient = postgres(DB_URL); +export const db = drizzle(queryClient, { schema }); diff --git a/src/server/db/migrations/0000_lyrical_khan.sql b/src/server/db/migrations/0000_lyrical_khan.sql new file mode 100644 index 00000000..20ff7883 --- /dev/null +++ b/src/server/db/migrations/0000_lyrical_khan.sql @@ -0,0 +1,21 @@ +CREATE TABLE IF NOT EXISTS "sse-chat_channel" ( + "id" text PRIMARY KEY NOT NULL, + "name" text, + "created_at" timestamp (3) with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp (3) with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +CREATE TABLE IF NOT EXISTS "sse-chat_post" ( + "id" text PRIMARY KEY NOT NULL, + "channel_id" text, + "name" text, + "text" text, + "created_at" timestamp (3) with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp (3) with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "sse-chat_post" ADD CONSTRAINT "sse-chat_post_channel_id_sse-chat_channel_id_fk" FOREIGN KEY ("channel_id") REFERENCES "public"."sse-chat_channel"("id") ON DELETE no action ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; diff --git a/src/server/db/migrations/0001_wet_tarantula.sql b/src/server/db/migrations/0001_wet_tarantula.sql new file mode 100644 index 00000000..e6737f46 --- /dev/null +++ b/src/server/db/migrations/0001_wet_tarantula.sql @@ -0,0 +1,3 @@ +ALTER TABLE "sse-chat_post" ALTER COLUMN "channel_id" SET NOT NULL;--> statement-breakpoint +ALTER TABLE "sse-chat_post" ALTER COLUMN "name" SET NOT NULL;--> statement-breakpoint +ALTER TABLE "sse-chat_post" ALTER COLUMN "text" SET NOT NULL; \ No newline at end of file diff --git a/src/server/db/migrations/meta/0000_snapshot.json b/src/server/db/migrations/meta/0000_snapshot.json new file mode 100644 index 00000000..2bcb25e7 --- /dev/null +++ b/src/server/db/migrations/meta/0000_snapshot.json @@ -0,0 +1,109 @@ +{ + "id": "726a54a4-0cb9-44fa-bce5-aee86c5ae30f", + "prevId": "00000000-0000-0000-0000-000000000000", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.sse-chat_channel": { + "name": "sse-chat_channel", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {} + }, + "public.sse-chat_post": { + "name": "sse-chat_post", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "channel_id": { + "name": "channel_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "text": { + "name": "text", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "sse-chat_post_channel_id_sse-chat_channel_id_fk": { + "name": "sse-chat_post_channel_id_sse-chat_channel_id_fk", + "tableFrom": "sse-chat_post", + "tableTo": "sse-chat_channel", + "columnsFrom": ["channel_id"], + "columnsTo": ["id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {} + } + }, + "enums": {}, + "schemas": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/src/server/db/migrations/meta/0001_snapshot.json b/src/server/db/migrations/meta/0001_snapshot.json new file mode 100644 index 00000000..b29cb554 --- /dev/null +++ b/src/server/db/migrations/meta/0001_snapshot.json @@ -0,0 +1,109 @@ +{ + "id": "720da587-6309-4c5a-83ed-230a080c0b88", + "prevId": "726a54a4-0cb9-44fa-bce5-aee86c5ae30f", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.sse-chat_channel": { + "name": "sse-chat_channel", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {} + }, + "public.sse-chat_post": { + "name": "sse-chat_post", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "channel_id": { + "name": "channel_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "text": { + "name": "text", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp (3) with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "sse-chat_post_channel_id_sse-chat_channel_id_fk": { + "name": "sse-chat_post_channel_id_sse-chat_channel_id_fk", + "tableFrom": "sse-chat_post", + "tableTo": "sse-chat_channel", + "columnsFrom": ["channel_id"], + "columnsTo": ["id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {} + } + }, + "enums": {}, + "schemas": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/src/server/db/migrations/meta/_journal.json b/src/server/db/migrations/meta/_journal.json new file mode 100644 index 00000000..3dc62870 --- /dev/null +++ b/src/server/db/migrations/meta/_journal.json @@ -0,0 +1,20 @@ +{ + "version": "7", + "dialect": "postgresql", + "entries": [ + { + "idx": 0, + "version": "7", + "when": 1718001586910, + "tag": "0000_lyrical_khan", + "breakpoints": true + }, + { + "idx": 1, + "version": "7", + "when": 1721401000711, + "tag": "0001_wet_tarantula", + "breakpoints": true + } + ] +} diff --git a/src/server/db/schema.ts b/src/server/db/schema.ts new file mode 100644 index 00000000..f96b75a7 --- /dev/null +++ b/src/server/db/schema.ts @@ -0,0 +1,66 @@ +import type { InferSelectModel } from 'drizzle-orm'; +import { relations } from 'drizzle-orm'; +import { pgTableCreator, text, timestamp } from 'drizzle-orm/pg-core'; + +const pgTable = pgTableCreator((name) => `sse-chat_${name}`); + +export const Post = pgTable('post', { + id: text('id') + .primaryKey() + .$defaultFn(() => crypto.randomUUID()), + channelId: text('channel_id') + .notNull() + .references(() => Channel.id), + + name: text('name').notNull(), + text: text('text').notNull(), + + createdAt: timestamp('created_at', { + mode: 'date', + precision: 3, + withTimezone: true, + }) + .notNull() + .defaultNow(), + updatedAt: timestamp('updated_at', { + mode: 'date', + precision: 3, + withTimezone: true, + }) + .notNull() + .defaultNow() + .$onUpdateFn(() => new Date()), +}); +export type PostType = InferSelectModel; + +export const Channel = pgTable('channel', { + id: text('id') + .primaryKey() + .$defaultFn(() => crypto.randomUUID()), + name: text('name'), + + createdAt: timestamp('created_at', { + mode: 'date', + precision: 3, + withTimezone: true, + }) + .notNull() + .defaultNow(), + updatedAt: timestamp('updated_at', { + mode: 'date', + precision: 3, + withTimezone: true, + }) + .notNull() + .defaultNow() + .$onUpdateFn(() => new Date()), +}); +export type ChannelType = InferSelectModel; + +export const PostRelations = relations(Post, ({ one }) => ({ + channel: one(Channel, { fields: [Post.channelId], references: [Channel.id] }), +})); + +export const ChannelRelations = relations(Channel, ({ many }) => ({ + posts: many(Post), +})); diff --git a/src/server/routers/_app.ts b/src/server/routers/_app.ts new file mode 100644 index 00000000..e72c85c1 --- /dev/null +++ b/src/server/routers/_app.ts @@ -0,0 +1,39 @@ +/** + * This file contains the root router of your tRPC-backend + */ +import { observable } from '@trpc/server/observable'; +import { createCallerFactory } from '@trpc/server/unstable-core-do-not-import'; +import { cache } from 'react'; +import { auth } from '../auth'; +import type { Context } from '../context'; +import { publicProcedure, router } from '../trpc'; +import { channelRouter } from './channel'; +import { postRouter } from './post'; + +export const appRouter = router({ + healthcheck: publicProcedure.query(() => 'yay!'), + + channel: channelRouter, + post: postRouter, + + randomNumber: publicProcedure.subscription(() => { + return observable((emit) => { + const int = setInterval(() => { + emit.next(Math.random()); + }, 500); + return () => { + clearInterval(int); + }; + }); + }), +}); + +export type AppRouter = typeof appRouter; + +const createCallerContext = cache( + async (): Promise => ({ + session: await auth(), + }), +); + +export const caller = createCallerFactory()(appRouter)(createCallerContext); diff --git a/src/server/routers/channel.ts b/src/server/routers/channel.ts new file mode 100644 index 00000000..93a517b1 --- /dev/null +++ b/src/server/routers/channel.ts @@ -0,0 +1,122 @@ +import EventEmitter, { on } from 'node:events'; +import type { TRPCRouterRecord } from '@trpc/server'; +import { db } from '~/server/db/client'; +import type { PostType } from '~/server/db/schema'; +import { Channel } from '~/server/db/schema'; +import { authedProcedure, publicProcedure } from '~/server/trpc'; +import { z } from 'zod'; + +export type WhoIsTyping = Record; + +type EventMap = Record; +class IterableEventEmitter> extends EventEmitter { + toIterable( + eventName: TEventName, + opts?: NonNullable[2]>, + ): AsyncIterable { + return on(this as any, eventName, opts) as any; + } +} + +export interface MyEvents { + add: [channelId: string, data: PostType]; + isTypingUpdate: [channelId: string, who: WhoIsTyping]; +} + +// In a real app, you'd probably use Redis or something +export const ee = new IterableEventEmitter(); + +// who is currently typing for each channel, key is `name` +export const currentlyTyping: Record = Object.create(null); + +// every 1s, clear old "isTyping" +setInterval(() => { + const updatedChannels = new Set(); + const now = Date.now(); + for (const [channelId, typers] of Object.entries(currentlyTyping)) { + for (const [key, value] of Object.entries(typers ?? {})) { + if (now - value.lastTyped.getTime() > 3e3) { + delete typers[key]; + updatedChannels.add(channelId); + } + } + } + updatedChannels.forEach((channelId) => { + ee.emit('isTypingUpdate', channelId, currentlyTyping[channelId] ?? {}); + }); +}, 3e3).unref(); + +export const channelRouter = { + list: publicProcedure.query(() => { + return db.query.Channel.findMany(); + }), + + create: authedProcedure + .input(z.object({ name: z.string().trim().min(2) })) + .mutation(async ({ ctx, input }) => { + const [channel] = await db + .insert(Channel) + .values({ + name: input.name, + }) + .returning(); + + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + return channel!.id; + }), + + isTyping: authedProcedure + .input(z.object({ channelId: z.string().uuid(), typing: z.boolean() })) + .mutation(async (opts) => { + const { name } = opts.ctx.user; + const { channelId } = opts.input; + + currentlyTyping[channelId] ??= {}; + + if (!opts.input.typing) { + delete currentlyTyping[channelId][name]; + } else { + currentlyTyping[channelId][name] = { + lastTyped: new Date(), + }; + } + ee.emit('isTypingUpdate', channelId, currentlyTyping[channelId]); + }), + + whoIsTyping: publicProcedure + .input( + z.object({ + channelId: z.string().uuid(), + }), + ) + .subscription(async function* (opts) { + const { channelId } = opts.input; + + let lastIsTyping = ''; + + /** + * yield who is typing if it has changed + * won't yield if it's the same as last time + */ + function* maybeYield(who: WhoIsTyping) { + const idx = Object.keys(who).toSorted().toString(); + if (idx === lastIsTyping) { + return; + } + yield Object.keys(who); + + lastIsTyping = idx; + } + + // emit who is currently typing + yield* maybeYield(currentlyTyping[channelId] ?? {}); + + for await (const [channelId, who] of ee.toIterable('isTypingUpdate', { + signal: opts.signal, + })) { + if (channelId === opts.input.channelId) { + yield* maybeYield(who); + } + } + }), +} satisfies TRPCRouterRecord; diff --git a/src/server/routers/post.ts b/src/server/routers/post.ts new file mode 100644 index 00000000..732f2912 --- /dev/null +++ b/src/server/routers/post.ts @@ -0,0 +1,142 @@ +import { tracked } from '@trpc/server'; +import { db } from '~/server/db/client'; +import { Post, type PostType } from '~/server/db/schema'; +import { z } from 'zod'; +import { authedProcedure, publicProcedure, router } from '../trpc'; +import { currentlyTyping, ee } from './channel'; + +export const postRouter = router({ + add: authedProcedure + .input( + z.object({ + id: z.string().uuid().optional(), + channelId: z.string().uuid(), + text: z.string().trim().min(1), + }), + ) + .mutation(async (opts) => { + const { channelId } = opts.input; + + const [post] = await db + .insert(Post) + .values({ + id: opts.input.id, + text: opts.input.text, + name: opts.ctx.user.name, + channelId, + }) + .returning(); + + const channelTyping = currentlyTyping[channelId]; + if (channelTyping) { + delete channelTyping[opts.ctx.user.name]; + ee.emit('isTypingUpdate', channelId, channelTyping); + } + + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const defPost = post!; + ee.emit('add', channelId, defPost); + + return post; + }), + + infinite: publicProcedure + .input( + z.object({ + channelId: z.string().uuid(), + cursor: z.date().nullish(), + take: z.number().min(1).max(50).nullish(), + }), + ) + .query(async (opts) => { + const take = opts.input.take ?? 20; + const cursor = opts.input.cursor; + + const page = await db.query.Post.findMany({ + orderBy: (fields, ops) => ops.desc(fields.createdAt), + where: (fields, ops) => + ops.and( + ops.eq(fields.channelId, opts.input.channelId), + cursor ? ops.lte(fields.createdAt, cursor) : undefined, + ), + limit: take + 1, + }); + + const items = page.reverse(); + let nextCursor: typeof cursor | null = null; + if (items.length > take) { + const prev = items.shift(); + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + nextCursor = prev!.createdAt; + } + return { + items, + nextCursor, + }; + }), + + onAdd: publicProcedure + .input( + z.object({ + channelId: z.string().uuid(), + // lastEventId is the last event id that the client has received + // On the first call, it will be whatever was passed in the initial setup + // If the client reconnects, it will be the last event id that the client received + lastEventId: z.string().nullish(), + }), + ) + .subscription(async function* (opts) { + // We start by subscribing to the event emitter so that we don't miss any new events while fetching + const iterable = ee.toIterable('add', { + signal: opts.signal, + }); + + // Fetch the last message createdAt based on the last event id + let lastMessageCreatedAt = await (async () => { + const lastEventId = opts.input.lastEventId; + if (!lastEventId) return null; + + const itemById = await db.query.Post.findFirst({ + where: (fields, ops) => ops.eq(fields.id, lastEventId), + }); + return itemById?.createdAt ?? null; + })(); + + const newPostsSinceLastMessage = await db.query.Post.findMany({ + where: (fields, ops) => + ops.and( + ops.eq(fields.channelId, opts.input.channelId), + lastMessageCreatedAt + ? ops.gt(fields.createdAt, lastMessageCreatedAt) + : undefined, + ), + orderBy: (fields, ops) => ops.asc(fields.createdAt), + }); + + function* maybeYield(post: PostType) { + if (post.channelId !== opts.input.channelId) { + // ignore posts from other channels - the event emitter can emit from other channels + return; + } + if (lastMessageCreatedAt && post.createdAt <= lastMessageCreatedAt) { + // ignore posts that we've already sent - happens if there is a race condition between the query and the event emitter + return; + } + + yield tracked(post.id, post); + + // update the cursor so that we don't send this post again + lastMessageCreatedAt = post.createdAt; + } + + // yield the posts we fetched from the db + for (const post of newPostsSinceLastMessage) { + yield* maybeYield(post); + } + + // yield any new posts from the event emitter + for await (const [channelId, post] of iterable) { + yield* maybeYield(post); + } + }), +}); diff --git a/src/server/sse/ACCEPTANCE.md b/src/server/sse/ACCEPTANCE.md new file mode 100644 index 00000000..a837c75f --- /dev/null +++ b/src/server/sse/ACCEPTANCE.md @@ -0,0 +1,162 @@ +# SSE Acceptance Guide + +This document captures the acceptance criteria, flow, verification steps, and edge cases for the SSE feature: "SSE endpoint implemented to accept client connections and maintain open streams." + +## Scope + +- Server supports establishing Server-Sent Events (SSE) connections from authenticated clients. +- Connection remains open and continuously streams events until the client disconnects or an error occurs. +- Heartbeats are sent to keep the connection active through proxies. +- Proper cleanup happens on disconnect or error. + +## Acceptance Criteria (Verified) + +1. SSE endpoint implemented to accept client connections and maintain open streams + - Evidence: + - Endpoint implemented at `src/app/api/sse/route.ts` using a `ReadableStream` and returning appropriate SSE headers (`text/event-stream`, `keep-alive`, `no-cache`). + +2. Clients can subscribe and receive events pushed from the server + - Evidence: + - Demo client in `src/app/(protected)/sse-demo/page.tsx` establishes `EventSource` to `/api/sse`, listens for `message`, `welcome`, and named events. + - Reusable hook in `src/hooks/useSSE.ts` supports topics and event handlers. + +3. Server code can send arbitrary named events with JSON payloads to individual or multiple clients + - Evidence: + - API helpers in `src/server/sse/index.ts`: `notifyUser`, `notifySession`, `notifyTopic`, `broadcast`, `notifyConnection` accept an event name and arbitrary data. + - tRPC router in `src/server/api/routers/sse.ts` exposes test mutations allowing custom `event` names and payloads. + +4. Heartbeat/ping mechanism in place to keep connections alive + - Evidence: + - Route-level heartbeat comment lines every 30s in `src/app/api/sse/route.ts`. + - Manager-level broadcast heartbeat every 30s in `src/server/sse/manager.ts`. + +5. Proper handling of client disconnects with cleanup of server resources + - Evidence: + - Abort handling and cleanup in `src/app/api/sse/route.ts` unregisters connections and closes the stream. + - Manager `unregister` disposes of bus subscriptions and removes connection (`src/server/sse/manager.ts`). + +6. Error handling and logging included + - Evidence: + - Logging via `src/lib/logger.ts` in route and manager for connection attempts, sends, failures, cleanup, and errors. + - Stream write and event send operations wrapped with try/catch and error logs. + +7. Well-documented usage for backend integration + - Evidence: + - Developer documentation in `src/server/sse/README.md` (Quick Start, API Reference, Integration Examples). + - This `ACCEPTANCE.md` provides criteria, flow, verification, and edge cases. + +## Components Involved + +- Endpoint: `GET /api/sse` in `src/app/api/sse/route.ts` +- Manager: `ConnectionManager` in `src/server/sse/manager.ts` +- API helpers: `notifyUser`, `notifyTopic`, `broadcast`, etc. in `src/server/sse/index.ts` +- Demo Client: `src/app/(protected)/sse-demo/page.tsx` + +## Functional Acceptance Criteria + +1. Authentication + - If the user is not authenticated, the endpoint returns `401 Unauthorized` and does not create a stream. + - If the user is authenticated, the server proceeds to establish an SSE stream. + +2. SSE Response + - Response headers include: + - `Content-Type: text/event-stream` + - `Cache-Control: no-cache, no-transform` + - `Connection: keep-alive` + - `X-Accel-Buffering: no` + - The response body is a `ReadableStream` that remains open. + +3. Connection Lifecycle + - On connect, server sends an initial `retry` line and a `welcome` event with connection metadata. + - Server registers the connection with `ConnectionManager`, subscribing to broadcast, user, session, and optional topic channels. + - Heartbeats are sent at a fixed interval to keep intermediaries from closing the stream. + - On client disconnect or error, the server unregisters the connection and closes the stream. + +4. Event Delivery + - Events published via `notifyUser`, `notifySession`, `notifyTopic`, or `broadcast` are delivered to matching open connections. + - Event data is serialized as JSON and emitted using SSE format (`event`, `data`, optional `id`, and blank line terminator). + +5. Observability + - Connection registrations/unregistrations and send failures are logged via `src/lib/logger.ts`. + - Statistics are available via tRPC router (`src/server/api/routers/sse.ts` → `getStats`, `getMyConnections`, etc.). + +## Non-Functional Acceptance Criteria + +- Heartbeat interval keeps streams active across proxies/CDNs. +- Resource cleanup on disconnect prevents connection leaks. +- Basic CORS headers are present. For cross-origin with credentials, explicit origin and `Access-Control-Allow-Credentials: true` should be used (not `*`). + +## End-to-End Flow + +1. Client (browser) creates an `EventSource` to `/api/sse`, optionally with `?topic=a,b,c`. +2. Server authenticates the session. If valid, it creates a `ReadableStream` and registers the connection in `ConnectionManager`. +3. Server immediately sends `retry` metadata and a `welcome` event containing `connId`, `userId`, `sessionId`. +4. `ConnectionManager` subscribes the connection to channels: `broadcast`, `user:{userId}`, `session:{sessionId}`, and `topic:{topic}` for each requested topic. +5. Events published through the SSE API helpers are routed through the bus to subscribers and written to the stream. +6. Heartbeats are sent periodically. On abort/error, cleanup is performed and the stream is closed. + +## How to Verify (Manual) + +Precondition: You are signed in to the app. + +1. Navigate to the demo page `/(protected)/sse-demo`. + - Expected: Connection status transitions to `CONNECTED`. + - Expected: A `welcome` event appears with a `connId` and `sessionId` (if available). + +2. Send test events using the demo controls: + - "To User": emits your selected event type to your user; verify it appears in the messages list. + - "To Topic": choose a topic (e.g., `test-topic`); verify receipt when subscribed to that topic. + - "Broadcast": verify all open demo tabs receive the event. + +3. Disconnect behavior: + - Close the tab or navigate away; the server should log unregistration, and stats should show fewer active connections. + +4. Unauthorized check: + - Open the SSE endpoint in an incognito window without logging in (e.g., navigate to `/api/sse`). + - Expected: `401 Unauthorized` response. + +## Minimal Code Snippets (Reference) + +Client connection example (browser console): + +```javascript +const es = new EventSource('/api/sse?topic=test-topic'); +es.addEventListener('open', () => console.log('open')); +es.addEventListener('welcome', (e) => console.log('welcome', JSON.parse(e.data))); +es.addEventListener('message', (e) => console.log('message', e.data)); +es.addEventListener('error', (e) => console.error('error', e)); +``` + +Publishing events (server-side): + +```ts +import { notifyUser, notifyTopic, broadcast } from '@/server/sse'; + +await notifyUser('user-id', 'test_message', { message: 'hello' }); +await notifyTopic('test-topic', 'topic_message', { message: 'hello topic' }); +await broadcast('broadcast_message', { message: 'hello everyone' }); +``` + +## Edge Cases + +- Multiple tabs per user: all tabs receive user-targeted events; session-targeted events go only to the matching tab. +- Empty topic list: stream still opens; only broadcast/user/session channels are subscribed. +- Network interruptions: browser will auto-retry; server supports reconnect via `retry` line. If you need resume with `Last-Event-ID`, add event `id`s in the stream. + +## Definition of Done + +- Authenticated clients can connect to `/api/sse` and receive a `welcome` event. +- Response uses `text/event-stream` and remains open with periodic heartbeats. +- Events published through the API helpers are delivered to connected clients as expected. +- Disconnections trigger cleanup and stats reflect accurate counts. +- Unauthorized requests return `401` without opening a stream. + +## Related Files + +- `src/app/api/sse/route.ts` +- `src/server/sse/manager.ts` +- `src/server/sse/index.ts` +- `src/server/api/routers/sse.ts` +- `src/app/(protected)/sse-demo/page.tsx` + + diff --git a/src/server/sse/README.md b/src/server/sse/README.md new file mode 100644 index 00000000..0da95381 --- /dev/null +++ b/src/server/sse/README.md @@ -0,0 +1,411 @@ +# Server-Sent Events (SSE) Implementation + +This directory contains a comprehensive Server-Sent Events (SSE) implementation for real-time, server-to-client notifications across the application. + +## Overview + +The SSE layer provides: +- **Centralized connection management** - Track active client connections per user/session +- **Flexible event routing** - Send events to specific users, sessions, topics, or broadcast to all +- **Automatic heartbeat** - Keep connections alive with periodic ping messages +- **Comprehensive error handling** - Graceful handling of disconnections and errors +- **Real-time monitoring** - Statistics and connection tracking +- **Clean API** - Easy integration for backend modules + +## Architecture + +``` +┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ +│ Client (UI) │ │ SSE Manager │ │ Event Bus │ +│ │ │ │ │ │ +│ EventSource │◄──â–ē│ ConnectionMgr │◄──â–ē│ InMemoryBus │ +│ │ │ │ │ │ +└─────────────────┘ └─────────────────┘ └─────────────────┘ + │ + â–ŧ + ┌─────────────────┐ + │ tRPC API │ + │ │ + │ SSE Endpoints │ + └─────────────────┘ +``` + +## Core Components + +### 1. Connection Manager (`manager.ts`) +- Manages active SSE connections +- Handles connection lifecycle (connect, disconnect, errors) +- Provides statistics and monitoring +- Implements automatic heartbeat mechanism + +### 2. Event Bus (`bus.ts`) +- Abstract event publishing/subscribing system +- In-memory implementation for single-server deployments +- Redis implementation available for multi-server scaling + +### 3. Types (`types.ts`) +- Comprehensive type definitions +- Connection information interfaces +- Error handling types +- Statistics interfaces + +### 4. API Layer (`index.ts`) +- High-level API functions for sending events +- Utility functions for common notification patterns +- Management and monitoring functions + +## Quick Start + +### 1. Connect from Client + +```typescript +// Basic SSE connection +const eventSource = new EventSource("/api/sse?topic=notifications,updates", { + withCredentials: true +}); + +// Listen for events +eventSource.addEventListener("message", (e) => { + console.log("Received:", e.data); +}); + +eventSource.addEventListener("welcome", (e) => { + const data = JSON.parse(e.data); + console.log("Connected with ID:", data.connId); +}); + +// Handle errors +eventSource.addEventListener("error", (e) => { + console.error("SSE error:", e); +}); +``` + +### 2. Send Events from Server + +```typescript +import { notifyUser, broadcast, notifyTopic } from "@/server/sse"; + +// Send to specific user +await notifyUser("user123", "new_message", { + message: "Hello!", + timestamp: Date.now() +}); + +// Broadcast to all connected clients +await broadcast("system_update", { + message: "Server maintenance in 5 minutes", + type: "warning" +}); + +// Send to topic subscribers +await notifyTopic("live-scores", "score_update", { + team: "Home Team", + score: 2 +}); +``` + +## API Reference + +### Core Functions + +#### `notifyUser(userId: string, event: string, data: any)` +Send an event to all connections of a specific user. + +#### `notifySession(sessionId: string, event: string, data: any)` +Send an event to a specific browser session/tab. + +#### `notifyTopic(topic: string, event: string, data: any)` +Send an event to all clients subscribed to a topic. + +#### `broadcast(event: string, data: any)` +Send an event to all connected clients. + +#### `notifyConnection(connId: string, event: string, data: any)` +Send an event to a specific connection. + +### Utility Functions + +#### `notifyNewMessage(userId: string, message: any)` +Send a new message notification to a user. + +#### `notifySystemUpdate(event: string, data: any)` +Broadcast a system update to all clients. + +#### `notifyUserStatus(userId: string, status: string)` +Notify about user status changes. + +#### `notifyLiveUpdate(topic: string, data: any)` +Send live data updates to topic subscribers. + +### Management Functions + +#### `getStats(): SseStats` +Get comprehensive SSE statistics. + +#### `getConnectionInfo(connId: string): ConnectionInfo | null` +Get information about a specific connection. + +#### `getUserConnections(userId: string): ConnectionInfo[]` +Get all connections for a user. + +#### `getTopicConnections(topic: string): ConnectionInfo[]` +Get all connections for a topic. + +#### `cleanup(): Promise` +Clean up all connections (useful for graceful shutdown). + +#### `healthCheck()` +Get health status of the SSE system. + +## Integration Examples + +### Webhook Handler Integration + +```typescript +// In a webhook handler +export async function POST(req: Request) { + const payload = await req.json(); + + // Process webhook + const result = await processWebhook(payload); + + // Notify relevant users + if (result.userId) { + await notifyUser(result.userId, "webhook_processed", { + type: payload.type, + status: "success", + timestamp: Date.now() + }); + } + + return Response.json({ success: true }); +} +``` + +### Background Job Integration + +```typescript +// In a background job processor +export async function processJob(jobData: any) { + try { + const result = await performJob(jobData); + + // Notify user about job completion + await notifyUser(jobData.userId, "job_completed", { + jobId: jobData.id, + result, + timestamp: Date.now() + }); + + } catch (error) { + // Notify about job failure + await notifyUser(jobData.userId, "job_failed", { + jobId: jobData.id, + error: error.message, + timestamp: Date.now() + }); + } +} +``` + +### Real-time Chat Integration + +```typescript +// In a chat message handler +export async function sendMessage(message: ChatMessage) { + // Save message to database + const savedMessage = await saveMessage(message); + + // Notify all users in the room + await notifyTopic(`chat:${message.roomId}`, "new_message", { + message: savedMessage, + timestamp: Date.now() + }); + + // Notify sender about delivery + await notifyUser(message.senderId, "message_sent", { + messageId: savedMessage.id, + timestamp: Date.now() + }); +} +``` + +## Monitoring and Debugging + +### SSE Statistics + +The system provides comprehensive statistics: + +```typescript +const stats = getStats(); +console.log({ + activeConnections: stats.activeConnections, + totalEventsSent: stats.totalEventsSent, + totalErrors: stats.totalErrors, + uptime: stats.uptime, + connectionsByUser: stats.connectionsByUser, + connectionsByTopic: stats.connectionsByTopic +}); +``` + +### Connection Monitoring + +```typescript +// Get all connections for a user +const userConnections = getUserConnections("user123"); +console.log(`User has ${userConnections.length} active connections`); + +// Get all connections for a topic +const topicConnections = getTopicConnections("live-scores"); +console.log(`Topic has ${topicConnections.length} subscribers`); +``` + +### Health Check + +```typescript +const health = healthCheck(); +console.log({ + status: health.status, + activeConnections: health.activeConnections, + uptime: health.uptime +}); +``` + +## Error Handling + +The SSE system includes comprehensive error handling: + +### Connection Errors +- Automatic reconnection with exponential backoff +- Error logging and monitoring +- Graceful cleanup of failed connections + +### Event Delivery Errors +- Failed event delivery is logged +- Connection status tracking +- Automatic retry mechanisms + +### Server Errors +- Graceful degradation +- Error reporting to monitoring systems +- Automatic recovery + +## Scaling Considerations + +### Single Server (Current Implementation) +- Uses in-memory event bus +- Suitable for small to medium applications +- All connections handled by single process + +### Multi-Server Scaling +- Switch to Redis event bus for horizontal scaling +- Multiple server instances can share connections +- Load balancing across SSE endpoints + +```typescript +// For production scaling, replace InMemoryBus with RedisBus +import { RedisBus } from "./bus"; +import Redis from "ioredis"; + +const pub = new Redis(process.env.REDIS_URL); +const sub = new Redis(process.env.REDIS_URL); +const bus = new RedisBus(sub, pub); +export const sse = new ConnectionManager(bus); +``` + +## Security Considerations + +### Authentication +- SSE connections require valid session +- User-specific event routing +- Session-based access control + +### Rate Limiting +- Implement rate limiting on SSE endpoints +- Prevent abuse and resource exhaustion +- Monitor connection patterns + +### Data Validation +- Validate all event data before sending +- Sanitize user inputs +- Implement proper error boundaries + +## Testing + +### Demo Interface +Visit `/sse-demo` to test all SSE functionality: +- Connection status monitoring +- Event sending and receiving +- Statistics and metrics +- Different event types and routing + +### Automated Testing +```typescript +// Example test +describe("SSE", () => { + it("should send events to connected clients", async () => { + // Setup SSE connection + const eventSource = new EventSource("/api/sse"); + + // Send test event + await notifyUser("test-user", "test", { message: "Hello" }); + + // Verify event received + // ... test assertions + }); +}); +``` + +## Best Practices + +1. **Use appropriate event types** - Choose the right routing method for your use case +2. **Handle reconnections** - Implement proper reconnection logic in clients +3. **Monitor performance** - Track connection counts and event volumes +4. **Clean up resources** - Properly close connections when components unmount +5. **Error boundaries** - Implement error handling for SSE failures +6. **Rate limiting** - Prevent abuse of SSE endpoints +7. **Logging** - Log important events for debugging and monitoring + +## Troubleshooting + +### Common Issues + +1. **Connections not receiving events** + - Check user authentication + - Verify topic subscriptions + - Check connection status + +2. **High memory usage** + - Monitor connection counts + - Implement connection limits + - Check for connection leaks + +3. **Events not being sent** + - Verify SSE manager is initialized + - Check event bus configuration + - Review error logs + +### Debug Commands + +```typescript +// Check connection status +const stats = getStats(); +console.log("Active connections:", stats.activeConnections); + +// Check specific user connections +const userConnections = getUserConnections("user123"); +console.log("User connections:", userConnections); + +// Test event sending +await notifyUser("user123", "test", { message: "Debug test" }); +``` + +## Performance Optimization + +1. **Connection pooling** - Reuse connections when possible +2. **Event batching** - Batch multiple events when appropriate +3. **Selective subscriptions** - Only subscribe to necessary topics +4. **Connection limits** - Implement per-user connection limits +5. **Memory monitoring** - Track memory usage and connection counts + +This SSE implementation provides a robust foundation for real-time features in your application. The modular design allows for easy extension and customization based on your specific requirements. diff --git a/src/server/sse/bus.ts b/src/server/sse/bus.ts new file mode 100644 index 00000000..0b5a7dda --- /dev/null +++ b/src/server/sse/bus.ts @@ -0,0 +1,34 @@ +export interface EventBus { + publish(channel: string, msg: any): Promise; + subscribe(channel: string, handler: (msg: any) => void): Promise<() => Promise>; + } + + // Dev/local adapter + import { EventEmitter } from "events"; + export class InMemoryBus implements EventBus { + private ee = new EventEmitter(); + async publish(ch: string, msg: any) { this.ee.emit(ch, msg); } + async subscribe(ch: string, handler: (msg: any) => void) { + const h = (m: unknown) => { try { handler(m); } catch {} }; + this.ee.on(ch, h); + return async () => { this.ee.off(ch, h); }; + } + } + + // Redis adapter (wire later if you run multiple instances) + export class RedisBus implements EventBus { + constructor(private sub: any, private pub: any) {} + async publish(ch: string, msg: any) { + await this.pub.publish(ch, JSON.stringify(msg)); + } + async subscribe(ch: string, handler: (msg: any) => void) { + const onMessage = (channel: string, payload: string) => { + if (channel !== ch) return; + try { handler(JSON.parse(payload)); } catch {} + }; + await this.sub.subscribe(ch); + this.sub.on("message", onMessage); + return async () => { this.sub.off("message", onMessage); await this.sub.unsubscribe(ch); }; + } + } + \ No newline at end of file diff --git a/src/server/sse/index.ts b/src/server/sse/index.ts new file mode 100644 index 00000000..8f8496ca --- /dev/null +++ b/src/server/sse/index.ts @@ -0,0 +1,110 @@ +import { InMemoryBus /*, RedisBus */ } from "./bus"; +import { ConnectionManager } from "./manager"; +import type { SseStats, ConnectionInfo } from "./types"; +import { log } from "@/lib/logger"; + +// For prod fan-out: create ioredis pub/sub clients and pass into RedisBus. +const bus = new InMemoryBus(); +export const sse = new ConnectionManager(bus); + +/** + * Core SSE API functions for sending events to clients + */ + +// Send to specific user (all their connections) +export const notifyUser = (userId: string, event: string, data: any) => + sse.send({ type: "user", userId }, event, data); + +// Send to specific session (single browser tab) +export const notifySession = (sessionId: string, event: string, data: any) => + sse.send({ type: "session", sessionId }, event, data); + +// Send to all clients subscribed to a topic +export const notifyTopic = (topic: string, event: string, data: any) => + sse.send({ type: "topic", topic }, event, data); + +// Broadcast to all connected clients +export const broadcast = (event: string, data: any) => + sse.send({ type: "broadcast" }, event, data); + +// Send to a specific connection +export const notifyConnection = (connId: string, event: string, data: any) => + sse.sendToConnection(connId, event, data); + +/** + * Utility functions for common notification patterns + */ + +// Notify user about a new message +export const notifyNewMessage = (userId: string, message: any) => { + return notifyUser(userId, "new_message", { + message, + timestamp: Date.now() + }); +}; + +// Notify about system updates +export const notifySystemUpdate = (event: string, data: any) => { + return broadcast("system_update", { + event, + data, + timestamp: Date.now() + }); +}; + +// Notify about user status changes +export const notifyUserStatus = (userId: string, status: string) => { + return notifyUser(userId, "status_change", { + status, + timestamp: Date.now() + }); +}; + +// Notify about real-time updates (e.g., live data) +export const notifyLiveUpdate = (topic: string, data: any) => { + return notifyTopic(topic, "live_update", { + data, + timestamp: Date.now() + }); +}; + +/** + * Management and monitoring functions + */ + +// Get connection statistics +export const getStats = (): SseStats => sse.getStats(); + +// Get information about a specific connection +export const getConnectionInfo = (connId: string): ConnectionInfo | null => + sse.getConnectionInfo(connId); + +// Get all connections for a user +export const getUserConnections = (userId: string): ConnectionInfo[] => + sse.getConnectionsByUser(userId); + +// Get all connections for a topic +export const getTopicConnections = (topic: string): ConnectionInfo[] => + sse.getConnectionsByTopic(topic); + +// Clean up all connections (useful for graceful shutdown) +export const cleanup = async () => { + log.info("Starting SSE cleanup"); + await sse.cleanup(); + log.info("SSE cleanup completed"); +}; + +/** + * Health check function + */ +export const healthCheck = () => { + const stats = getStats(); + return { + status: "healthy", + uptime: stats.uptime, + activeConnections: stats.activeConnections, + totalConnections: stats.totalConnections, + totalEventsSent: stats.totalEventsSent, + totalErrors: stats.totalErrors + }; +}; diff --git a/src/server/sse/manager.ts b/src/server/sse/manager.ts new file mode 100644 index 00000000..717df895 --- /dev/null +++ b/src/server/sse/manager.ts @@ -0,0 +1,284 @@ +import type { EventBus } from "./bus"; +import type { Target, SseEventEnvelope, ConnectionInfo, SseError, ConnectionStatus, SseStats } from "./types"; +import { log } from "@/lib/logger"; + +type ConnId = string; + +interface Connection { + info: ConnectionInfo; + status: ConnectionStatus; + send: (evt: SseEventEnvelope) => void; + dispose: () => Promise; + error?: SseError; +} + +export class ConnectionManager { + private connections = new Map(); + private startTime = Date.now(); + private totalEventsSent = 0; + private totalErrors = 0; + private heartbeatInterval?: NodeJS.Timeout; + + constructor(private bus: EventBus) { + this.startHeartbeat(); + } + + private startHeartbeat() { + // Send heartbeat every 30 seconds to keep connections alive + this.heartbeatInterval = setInterval(() => { + this.broadcastHeartbeat(); + }, 30000); + } + + private async broadcastHeartbeat() { + try { + await this.send({ type: "broadcast" }, "heartbeat", { timestamp: Date.now() }); + } catch (error) { + log.error("Failed to send heartbeat", { error }); + } + } + + async register(opts: { + connId: ConnId; + userId?: string; + sessionId?: string; + topics?: string[]; + send: (evt: SseEventEnvelope) => void; + userAgent?: string; + ip?: string; + }): Promise<() => Promise> { + try { + const subs: Array<() => Promise> = []; + const sub = async (ch: string) => { + const unsubscribe = await this.bus.subscribe(ch, (evt) => { + try { + opts.send(evt); + this.totalEventsSent++; + this.updateConnectionActivity(opts.connId); + } catch (error) { + log.error("Failed to send event to connection", { + connId: opts.connId, + error, + event: evt + }); + this.handleConnectionError(opts.connId, { + code: "SEND_ERROR", + message: "Failed to send event to client", + details: error, + timestamp: Date.now() + }); + } + }); + subs.push(unsubscribe); + }; + + // Subscribe to channels + await sub("broadcast"); + if (opts.userId) await sub(`user:${opts.userId}`); + if (opts.sessionId) await sub(`session:${opts.sessionId}`); + for (const topic of (opts.topics ?? [])) { + await sub(`topic:${topic}`); + } + + const connectionInfo: ConnectionInfo = { + connId: opts.connId, + userId: opts.userId, + sessionId: opts.sessionId, + topics: opts.topics ?? [], + connectedAt: Date.now(), + lastActivity: Date.now(), + userAgent: opts.userAgent, + ip: opts.ip, + }; + + const connection: Connection = { + info: connectionInfo, + status: 'connected', + send: opts.send, + dispose: async () => { + await Promise.all(subs.map(s => s())); + } + }; + + this.connections.set(opts.connId, connection); + + log.info("SSE connection registered", { + connId: opts.connId, + userId: opts.userId, + sessionId: opts.sessionId, + topics: opts.topics, + totalConnections: this.connections.size + }); + + // Send welcome message + await this.sendToConnection(opts.connId, "connected", { + connId: opts.connId, + timestamp: Date.now() + }); + + return async () => this.unregister(opts.connId); + } catch (error) { + log.error("Failed to register SSE connection", { + connId: opts.connId, + error + }); + throw error; + } + } + + async unregister(connId: ConnId) { + const connection = this.connections.get(connId); + if (!connection) { + log.warn("Attempted to unregister non-existent connection", { connId }); + return; + } + + try { + await connection.dispose(); + this.connections.delete(connId); + + log.info("SSE connection unregistered", { + connId, + userId: connection.info.userId, + duration: Date.now() - connection.info.connectedAt, + totalConnections: this.connections.size + }); + } catch (error) { + log.error("Error during connection cleanup", { connId, error }); + } + } + + private updateConnectionActivity(connId: ConnId) { + const connection = this.connections.get(connId); + if (connection) { + connection.info.lastActivity = Date.now(); + } + } + + private handleConnectionError(connId: ConnId, error: SseError) { + const connection = this.connections.get(connId); + if (connection) { + connection.status = 'error'; + connection.error = error; + this.totalErrors++; + + log.error("SSE connection error", { + connId, + error: error.message, + code: error.code + }); + } + } + + async send(target: Target, event: string, data: any) { + try { + const envelope: SseEventEnvelope = { + event, + data, + timestamp: Date.now() + }; + + const channel = + target.type === "broadcast" ? "broadcast" : + target.type === "user" ? `user:${target.userId}` : + target.type === "session" ? `session:${target.sessionId}` : + `topic:${target.topic}`; + + await this.bus.publish(channel, envelope); + + log.info("SSE event sent", { + target, + event, + channel, + dataSize: JSON.stringify(data).length + }); + } catch (error) { + log.error("Failed to send SSE event", { target, event, error }); + throw error; + } + } + + async sendToConnection(connId: ConnId, event: string, data: any) { + const connection = this.connections.get(connId); + if (!connection) { + throw new Error(`Connection ${connId} not found`); + } + + try { + const envelope: SseEventEnvelope = { + event, + data, + timestamp: Date.now() + }; + + connection.send(envelope); + this.totalEventsSent++; + this.updateConnectionActivity(connId); + } catch (error) { + this.handleConnectionError(connId, { + code: "SEND_ERROR", + message: "Failed to send event to specific connection", + details: error, + timestamp: Date.now() + }); + throw error; + } + } + + getConnectionInfo(connId: ConnId): ConnectionInfo | null { + const connection = this.connections.get(connId); + return connection ? connection.info : null; + } + + getConnectionsByUser(userId: string): ConnectionInfo[] { + return Array.from(this.connections.values()) + .filter(conn => conn.info.userId === userId) + .map(conn => conn.info); + } + + getConnectionsByTopic(topic: string): ConnectionInfo[] { + return Array.from(this.connections.values()) + .filter(conn => conn.info.topics.includes(topic)) + .map(conn => conn.info); + } + + getStats(): SseStats { + const connectionsByUser: Record = {}; + const connectionsByTopic: Record = {}; + + for (const connection of this.connections.values()) { + if (connection.info.userId) { + connectionsByUser[connection.info.userId] = + (connectionsByUser[connection.info.userId] || 0) + 1; + } + + for (const topic of connection.info.topics) { + connectionsByTopic[topic] = + (connectionsByTopic[topic] || 0) + 1; + } + } + + return { + totalConnections: this.connections.size, + activeConnections: Array.from(this.connections.values()) + .filter(conn => conn.status === 'connected').length, + totalEventsSent: this.totalEventsSent, + totalErrors: this.totalErrors, + uptime: Date.now() - this.startTime, + connectionsByUser, + connectionsByTopic + }; + } + + async cleanup() { + if (this.heartbeatInterval) { + clearInterval(this.heartbeatInterval); + } + + // Clean up all connections + const connIds = Array.from(this.connections.keys()); + await Promise.all(connIds.map(connId => this.unregister(connId))); + + log.info("SSE manager cleaned up", { totalConnections: connIds.length }); + } +} diff --git a/src/server/sse/types.ts b/src/server/sse/types.ts new file mode 100644 index 00000000..df52c1cd --- /dev/null +++ b/src/server/sse/types.ts @@ -0,0 +1,49 @@ +import { z } from "zod"; + +export const SseEventEnvelope = z.object({ + event: z.string().min(1), + data: z.any(), + id: z.string().optional(), + userId: z.string().optional(), + sessionId: z.string().optional(), + topic: z.string().optional(), + timestamp: z.number().optional(), + retry: z.number().optional(), +}); +export type SseEventEnvelope = z.infer; + +export type Target = + | { type: "broadcast" } + | { type: "user"; userId: string } + | { type: "session"; sessionId: string } + | { type: "topic"; topic: string }; + +export interface ConnectionInfo { + connId: string; + userId?: string; + sessionId?: string; + topics: string[]; + connectedAt: number; + lastActivity: number; + userAgent?: string; + ip?: string; +} + +export interface SseError { + code: string; + message: string; + details?: any; + timestamp: number; +} + +export type ConnectionStatus = 'connected' | 'disconnected' | 'error'; + +export interface SseStats { + totalConnections: number; + activeConnections: number; + totalEventsSent: number; + totalErrors: number; + uptime: number; + connectionsByUser: Record; + connectionsByTopic: Record; +} diff --git a/src/server/trpc.ts b/src/server/trpc.ts new file mode 100644 index 00000000..a00afe5e --- /dev/null +++ b/src/server/trpc.ts @@ -0,0 +1,96 @@ +/** + * This is your entry point to setup the root configuration for tRPC on the server. + * - `initTRPC` should only be used once per app. + * - We export only the functionality that we use so we can enforce which base procedures should be used + * + * Learn how to create protected base procedures and other things below: + * @see https://trpc.io/docs/v11/router + * @see https://trpc.io/docs/v11/procedures + */ + +import { initTRPC, TRPCError } from '@trpc/server'; +import superjson from 'superjson'; +import type { Context } from './context'; + +const t = initTRPC.context().create({ + /** + * @see https://trpc.io/docs/v11/data-transformers + */ + transformer: superjson, + /** + * @see https://trpc.io/docs/v11/error-formatting + */ + errorFormatter({ shape }) { + return shape; + }, + sse: { + maxDurationMs: 5 * 60 * 1_000, // 5 minutes + ping: { + enabled: true, + intervalMs: 3_000, + }, + client: { + reconnectAfterInactivityMs: 5_000, + }, + }, +}); + +export const createCallerFactory = t.createCallerFactory; + +/** + * Create a router + * @see https://trpc.io/docs/v11/router + */ +export const router = t.router; + +/** + * Create an unprotected procedure + * @see https://trpc.io/docs/v11/procedures + **/ +export const publicProcedure = t.procedure.use( + async function artificialDelayInDevelopment(opts) { + const res = opts.next(opts); + + if (process.env.NODE_ENV === 'development') { + const randomNumber = (min: number, max: number) => + Math.floor(Math.random() * (max - min + 1)) + min; + + const delay = randomNumber(300, 1_000); + console.debug( + 'ℹī¸ doing artificial delay of', + delay, + 'ms before returning', + opts.path, + ); + + await new Promise((resolve) => setTimeout(resolve, delay)); + } + + return res; + }, +); + +/** + * @see https://trpc.io/docs/v11/merging-routers + */ +export const mergeRouters = t.mergeRouters; + +/** + * Protected base procedure + */ +export const authedProcedure = publicProcedure.use(function isAuthed(opts) { + const user = opts.ctx.session?.user; + + if (!user?.name) { + throw new TRPCError({ code: 'UNAUTHORIZED' }); + } + + return opts.next({ + ctx: { + user: { + ...user, + name: user.name, + }, + }, + }); +});