Add socket.io realtime infrastructure backed by Redis

components/providers.tsx

@@ -7,6 +7,7 @@ import { ThemeProvider } from "@/components/theme-provider"
 import { PageViewTracker } from "@/components/page-view-tracker"
 import { CookieBanner } from "@/components/cookie-banner"
 import { Logger } from "@/components/logger"
+import { RealtimeProvider } from "@/components/realtime-provider"
 import { Toaster } from "@/components/ui/toaster"
 
 interface ProvidersProps {
@@ -19,7 +20,7 @@ export function Providers({ children }: ProvidersProps) {
       <ThemeProvider attribute="class" defaultTheme="system" enableSystem disableTransitionOnChange>
         <Logger />
         <PageViewTracker />
-        {children}
+        <RealtimeProvider>{children}</RealtimeProvider>
         <Toaster />
         <CookieBanner />
       </ThemeProvider>

components/realtime-provider.tsx

@@ -0,0 +1,50 @@
+"use client"
+
+import type { ReactNode } from "react"
+import { useEffect } from "react"
+import { useRouter } from "next/navigation"
+import { getRealtimeClient } from "@/lib/realtime-client"
+
+interface RealtimeProviderProps {
+  children: ReactNode
+}
+
+export function RealtimeProvider({ children }: RealtimeProviderProps) {
+  const router = useRouter()
+
+  useEffect(() => {
+    const controller = new AbortController()
+
+    fetch("/api/socket/io", { signal: controller.signal }).catch(() => {
+      // The socket route will respond immediately once the server is ready.
+    })
+
+    const socket = getRealtimeClient()
+
+    if (!socket) {
+      controller.abort()
+      return
+    }
+
+    const handleEvent = () => {
+      router.refresh()
+    }
+
+    const handleConnect = () => {
+      router.refresh()
+    }
+
+    socket.on("connect", handleConnect)
+
+    socket.on("realtime:event", handleEvent)
+
+    return () => {
+      controller.abort()
+      socket.off("realtime:event", handleEvent)
+      socket.off("connect", handleConnect)
+    }
+  }, [router])
+
+  return <>{children}</>
+}
+

lib/db.ts

@@ -1,5 +1,7 @@
 import type { PrismaClient as PrismaClientType } from "@prisma/client"
 
+import { registerPrismaRealtime } from "@/lib/realtime-events"
+
 type PrismaClientConstructor = new (...args: unknown[]) => PrismaClientType
 
 const globalForPrisma = globalThis as unknown as { prisma?: PrismaClientType }
@@ -108,6 +110,10 @@ const prisma =
       })
     : createStubClient())
 
+if (PrismaClientCtor) {
+  registerPrismaRealtime(prisma as PrismaClientType)
+}
+
 if (process.env.NODE_ENV !== "production") {
   globalForPrisma.prisma = prisma
 }

lib/realtime-client.ts

@@ -0,0 +1,24 @@
+"use client"
+
+import { io, type Socket } from "socket.io-client"
+
+let socketInstance: Socket | null = null
+
+export function getRealtimeClient() {
+  if (typeof window === "undefined") {
+    return null
+  }
+
+  if (!socketInstance) {
+    socketInstance = io({
+      path: "/api/socket/io",
+      transports: ["websocket", "polling"],
+      autoConnect: true,
+      reconnection: true,
+      reconnectionDelay: 1000,
+      reconnectionDelayMax: 5000,
+    })
+  }
+
+  return socketInstance
+}

lib/realtime-events.ts

@@ -0,0 +1,55 @@
+import type { PrismaClient } from "@prisma/client"
+
+import { enqueueRealtimeEvent } from "@/lib/realtime-queue"
+
+const MUTATION_ACTIONS = new Set([
+  "create",
+  "createMany",
+  "update",
+  "updateMany",
+  "upsert",
+  "delete",
+  "deleteMany",
+])
+
+const globalForRealtime = globalThis as unknown as {
+  prismaRealtimeRegistered?: boolean
+}
+
+export function registerPrismaRealtime(prisma: PrismaClient) {
+  if (globalForRealtime.prismaRealtimeRegistered) {
+    return
+  }
+
+  prisma.$use(async (params, next) => {
+    const result = await next(params)
+
+    if (MUTATION_ACTIONS.has(params.action)) {
+      const model = params.model ?? "unknown"
+
+      try {
+        await enqueueRealtimeEvent(`prisma:${model}:${params.action}`, {
+          model,
+          action: params.action,
+          args: params.args,
+          result,
+          timestamp: Date.now(),
+        })
+      } catch (error) {
+        console.error("Failed to enqueue realtime Prisma event", error)
+      }
+    }
+
+    return result
+  })
+
+  globalForRealtime.prismaRealtimeRegistered = true
+}
+
+export async function broadcastRealtimeEvent(event: string, payload: unknown) {
+  try {
+    await enqueueRealtimeEvent(event, payload)
+  } catch (error) {
+    console.error("Failed to enqueue realtime event", error)
+  }
+}

lib/realtime-queue.ts

@@ -0,0 +1,83 @@
+import { Queue, Worker } from "bullmq"
+import type { JobsOptions } from "bullmq"
+
+import { createRedisConnection, getRedisUrl } from "@/lib/redis"
+import { getRealtimeIO } from "@/lib/realtime"
+
+const queueName = "realtime-events"
+
+const globalForQueue = globalThis as unknown as {
+  realtimeQueue?: Queue
+  realtimeWorker?: Worker
+  realtimeQueueConnection?: ReturnType<typeof createRedisConnection>
+  realtimeWorkerConnection?: ReturnType<typeof createRedisConnection>
+}
+
+function ensureConnections() {
+  if (!globalForQueue.realtimeQueueConnection) {
+    globalForQueue.realtimeQueueConnection = createRedisConnection()
+  }
+
+  if (!globalForQueue.realtimeWorkerConnection) {
+    globalForQueue.realtimeWorkerConnection = createRedisConnection()
+  }
+}
+
+export function getRealtimeQueue() {
+  ensureConnections()
+
+  if (!globalForQueue.realtimeQueue) {
+    globalForQueue.realtimeQueue = new Queue(queueName, {
+      connection: globalForQueue.realtimeQueueConnection,
+    })
+  }
+
+  return globalForQueue.realtimeQueue
+}
+
+function ensureWorker() {
+  ensureConnections()
+
+  if (!globalForQueue.realtimeWorker) {
+    globalForQueue.realtimeWorker = new Worker(
+      queueName,
+      async job => {
+        const io = getRealtimeIO()
+
+        if (io) {
+          io.emit("realtime:event", {
+            event: job.name,
+            payload: job.data,
+            jobId: job.id,
+            timestamp: Date.now(),
+          })
+
+          io.emit(job.name, job.data)
+        }
+      },
+      { connection: globalForQueue.realtimeWorkerConnection },
+    )
+
+    globalForQueue.realtimeWorker.on("error", error => {
+      console.error("Realtime worker error", error)
+    })
+  }
+}
+
+export async function enqueueRealtimeEvent(event: string, payload: unknown, options?: JobsOptions) {
+  const queue = getRealtimeQueue()
+  ensureWorker()
+
+  await queue.add(event, payload, {
+    removeOnComplete: { age: 60, count: 1000 },
+    removeOnFail: { age: 60 * 60, count: 100 },
+    ...options,
+  })
+}
+
+export function describeRealtimeQueue() {
+  return {
+    name: queueName,
+    redis: getRedisUrl(),
+  }
+}

lib/realtime.ts

@@ -0,0 +1,14 @@
+import type { Server as IOServer } from "socket.io"
+
+const globalForRealtime = globalThis as unknown as {
+  realtimeIO?: IOServer
+}
+
+export function getRealtimeIO() {
+  return globalForRealtime.realtimeIO
+}
+
+export function setRealtimeIO(io: IOServer) {
+  globalForRealtime.realtimeIO = io
+  return io
+}

lib/redis.ts

@@ -0,0 +1,28 @@
+import IORedis from "ioredis"
+
+const redisUrl = process.env.REDIS_URL || "redis://127.0.0.1:6379"
+
+const baseOptions = {
+  maxRetriesPerRequest: null as number | null,
+  enableReadyCheck: true,
+}
+
+const globalForRedis = globalThis as unknown as {
+  redisConnection?: IORedis
+}
+
+export function createRedisConnection() {
+  return new IORedis(redisUrl, baseOptions)
+}
+
+export function getRedisConnection() {
+  if (!globalForRedis.redisConnection) {
+    globalForRedis.redisConnection = createRedisConnection()
+  }
+
+  return globalForRedis.redisConnection
+}
+
+export function getRedisUrl() {
+  return redisUrl
+}

next-env.d.ts

@@ -1,5 +1,6 @@
 /// <reference types="next" />
 /// <reference types="next/image-types/global" />
+/// <reference types="next/navigation-types/compat/navigation" />
 
 // NOTE: This file should not be edited
 // see https://nextjs.org/docs/app/api-reference/config/typescript for more information.

package.json

@@ -59,6 +59,7 @@
     "@react-three/fiber": "^9.0.0",
     "autoprefixer": "^10.4.20",
     "bcryptjs": "3.0.2",
+    "bullmq": "^5.63.0",
     "class-variance-authority": "^0.7.1",
     "clsx": "^2.1.1",
     "cmdk": "1.0.4",
@@ -67,6 +68,7 @@
     "framer-motion": "^11.11.7",
     "immer": "10.1.3",
     "input-otp": "1.4.1",
+    "ioredis": "^5.8.2",
     "lucide-react": "^0.454.0",
     "next": "15.4.7",
     "next-auth": "4.24.13",
@@ -80,6 +82,8 @@
     "react-resizable-panels": "^2.1.7",
     "react-syntax-highlighter": "^16.1.0",
     "recharts": "2.15.0",
+    "socket.io": "^4.8.1",
+    "socket.io-client": "^4.8.1",
     "sonner": "^1.7.1",
     "swr": "^2.3.3",
     "tailwind-merge": "^2.5.5",

pages/api/socket/io.ts

@@ -0,0 +1,48 @@
+import type { NextApiRequest } from "next"
+import { Server as IOServer } from "socket.io"
+
+import type { NextApiResponseServerIO } from "@/types/next"
+import { getRealtimeIO, setRealtimeIO } from "@/lib/realtime"
+
+export const config = {
+  api: {
+    bodyParser: false,
+  },
+}
+
+export default function handler(req: NextApiRequest, res: NextApiResponseServerIO) {
+  const existingServer = res.socket.server as typeof res.socket.server & { io?: IOServer }
+
+  if (!existingServer.io) {
+    const io = new IOServer(res.socket.server, {
+      path: "/api/socket/io",
+      cors: {
+        origin: process.env.NEXT_PUBLIC_SITE_URL || "*",
+      },
+    })
+
+    setRealtimeIO(io)
+    existingServer.io = io
+
+    io.on("connection", socket => {
+      socket.emit("realtime:event", {
+        event: "socket:connected",
+        socketId: socket.id,
+        timestamp: Date.now(),
+      })
+
+      socket.on("disconnect", reason => {
+        socket.broadcast.emit("realtime:event", {
+          event: "socket:disconnected",
+          socketId: socket.id,
+          reason,
+          timestamp: Date.now(),
+        })
+      })
+    })
+  } else if (!getRealtimeIO()) {
+    setRealtimeIO(existingServer.io)
+  }
+
+  res.end()
+}