feat(hocuspocus): scope history queries to the collab document room

Resolve version list and watch data using the WebSocket document name
instead of trusting client-supplied ids, and reply on the same
connection with unicast stateless payloads.

Made-with: Cursor

Author: HMarzban

packages/hocuspocus.server/src/hocuspocus.server.ts

@@ -1,90 +1,106 @@
+import type { Connection } from '@hocuspocus/server'
 import { Server } from '@hocuspocus/server'
 
 import HocuspocusConfig from './config/hocuspocus.config'
+import { handleHistoryStateless } from './lib/history-stateless'
 import { logger } from './lib/logger'
-import { prisma, shutdownDatabase } from './lib/prisma'
+import { shutdownDatabase } from './lib/prisma'
 import { disconnectRedis } from './lib/redis'
-import type { HistoryPayload } from './types'
+import type { HistoryPayload } from './types/document.types'
 import { verifyJWT } from './utils'
 
 process.env.NODE_ENV = process.env.NODE_ENV || 'development'
 
-// Create logger for WebSocket server
 const wsLogger = logger.child({ service: 'websocket' })
 
-async function handleHistoryEvents(payload: HistoryPayload, _context: any, _document: any) {
-  const { type, documentId } = payload
-
-  switch (type) {
-    case 'history.list': {
-      const docs = await prisma.documents.findMany({
-        where: { documentId },
-        orderBy: { createdAt: 'desc' },
-        select: { version: true, commitMessage: true, createdAt: true }
-      })
-      return docs
-    }
-
-    case 'history.watch': {
-      const doc = await prisma.documents.findFirst({
-        where: { documentId, version: payload.version },
-        select: { data: true, version: true, commitMessage: true, createdAt: true }
-      })
-
-      if (!doc) return null
-
-      // Convert Buffer to Base64 string for transport
-      return {
-        data: Buffer.from(doc.data).toString('base64'),
-        version: doc.version,
-        commitMessage: doc.commitMessage,
-        createdAt: doc.createdAt
-      }
-    }
-
-    case 'history.prev':
-      return prisma.documents.findFirst({
-        where: { documentId, version: { lt: payload.currentVersion || 0 } },
-        orderBy: { version: 'desc' }
-      })
-
-    case 'history.next':
-      return prisma.documents.findFirst({
-        where: { documentId, version: { gt: payload.currentVersion || 0 } },
-        orderBy: { version: 'asc' }
-      })
-
-    default:
-      return payload
-  }
+function sendHistoryResponse(
+  connection: Connection,
+  type: string,
+  response: unknown,
+  error?: string
+) {
+  connection.sendStateless(
+    JSON.stringify({
+      msg: 'history.response',
+      type,
+      response,
+      ...(error ? { error } : {})
+    })
+  )
 }
 
-const broadcastToAll = (document: any, payload: any) => {
-  document.broadcastStateless(JSON.stringify(payload))
+/** Hocuspocus document name === provider `name` / room id; never trust client `documentId` for Prisma. */
+function roomDocumentId(document: { name?: string }): string | null {
+  const n = document?.name
+  return typeof n === 'string' && n.length > 0 ? n : null
 }
 
 const statelessExtension = {
-  async onStateless({ payload, context, document, _connection }: any) {
-    const parsedPayload = JSON.parse(payload)
+  async onStateless({
+    payload,
+    connection,
+    document
+  }: {
+    payload: string
+    connection: Connection
+    document: { name?: string; broadcastStateless: (p: string) => void }
+  }) {
+    let parsedPayload: {
+      msg?: string
+      type?: string
+      documentId?: string
+      version?: number
+      currentVersion?: number
+    }
+    try {
+      parsedPayload = JSON.parse(payload) as typeof parsedPayload
+    } catch {
+      return
+    }
 
     if (parsedPayload.msg === 'history') {
+      const canonicalId = roomDocumentId(document)
+      const type = parsedPayload.type
+
+      if (!canonicalId || !type) {
+        wsLogger.warn(
+          { parsedPayload, hasRoomName: Boolean(canonicalId) },
+          'history stateless missing room or type'
+        )
+        if (type) sendHistoryResponse(connection, type, null, 'history_failed')
+        return
+      }
+
+      if (parsedPayload.documentId != null && parsedPayload.documentId !== canonicalId) {
+        wsLogger.warn(
+          { clientDocumentId: parsedPayload.documentId, canonicalId },
+          'history stateless documentId does not match connection room'
+        )
+        sendHistoryResponse(connection, type, null, 'history_failed')
+        return
+      }
+
+      const historyPayload: HistoryPayload = {
+        type,
+        documentId: canonicalId,
+        version: parsedPayload.version,
+        currentVersion: parsedPayload.currentVersion
+      }
+
       try {
-        const response = await handleHistoryEvents(parsedPayload, context, document)
-        broadcastToAll(document, {
-          msg: 'history.response',
-          type: parsedPayload.type,
-          response
-        })
+        const response = await handleHistoryStateless(historyPayload)
+        sendHistoryResponse(connection, type, response)
       } catch (error) {
         wsLogger.error({ err: error }, 'Error handling history event')
+        sendHistoryResponse(connection, type, null, 'history_failed')
       }
-    } else {
-      document.broadcastStateless(JSON.stringify(parsedPayload))
+      return
     }
+
+    document.broadcastStateless(JSON.stringify(parsedPayload))
   }
 }
 
-// Get the base configuration and add our stateless extension
 const baseConfig = HocuspocusConfig()
 const serverConfig = {
   ...baseConfig,
@@ -98,7 +114,6 @@ const serverConfig = {
 
     try {
       const tokenData = JSON.parse(token)
-      // Extract deviceType from token (sent by webapp)
       const deviceType = tokenData.deviceType || 'desktop'
 
       if (tokenData.accessToken) {
@@ -109,7 +124,6 @@ const serverConfig = {
           return { user, slug: tokenData.slug || '', documentId: documentName, deviceType }
         }
 
-        // Token invalid
         if (process.env.NODE_ENV === 'production') {
           throw new Error('Invalid authentication token')
         }
@@ -118,7 +132,6 @@ const serverConfig = {
         return { user: null, slug: tokenData.slug || '', documentId: documentName, deviceType }
       }
 
-      // No accessToken, allow with slug only
       return { user: null, slug: tokenData.slug || '', documentId: documentName, deviceType }
     } catch (error) {
       wsLogger.error({ err: error, documentName }, 'Auth error')
@@ -132,10 +145,8 @@ const serverConfig = {
   }
 }
 
-// Configure and start the server
 const server = new Server(serverConfig)
 
-// Start listening
 server.listen()
 
 wsLogger.info({
@@ -145,7 +156,6 @@ wsLogger.info({
   url: `ws://localhost:${baseConfig.port}`
 })
 
-// Graceful shutdown
 const shutdown = async () => {
   wsLogger.info('Shutting down WebSocket server gracefully...')
 

packages/hocuspocus.server/src/lib/history-stateless.ts

@@ -0,0 +1,96 @@
+import type { HistoryPayload } from '../types/document.types'
+import { prisma } from './prisma'
+
+/** Metadata rows for the version sidebar (no Yjs payload). */
+export type HistoryVersionMeta = {
+  version: number
+  commitMessage: string | null
+  createdAt: Date
+}
+
+/** Full row for editor hydration (base64 Yjs). */
+export type HistorySnapshot = {
+  data: string
+  version: number
+  commitMessage: string | null
+  createdAt: Date
+}
+
+/** New list shape: sidebar list + latest body in one response (one network round-trip). */
+export type HistoryListResult = {
+  versions: HistoryVersionMeta[]
+  latestSnapshot: HistorySnapshot | null
+}
+
+function toSnapshot(doc: {
+  data: Buffer | Uint8Array
+  version: number
+  commitMessage: string | null
+  createdAt: Date
+}): HistorySnapshot {
+  return {
+    data: Buffer.from(doc.data).toString('base64'),
+    version: doc.version,
+    commitMessage: doc.commitMessage,
+    createdAt: doc.createdAt
+  }
+}
+
+/**
+ * Document version history over Hocuspocus stateless channel.
+ * `history.prev` / `history.next` are reserved for future UI; not used by the webapp today.
+ */
+export async function handleHistoryStateless(payload: HistoryPayload): Promise<unknown> {
+  const { type, documentId } = payload
+
+  switch (type) {
+    case 'history.list': {
+      const versions = await prisma.documents.findMany({
+        where: { documentId },
+        orderBy: { createdAt: 'desc' },
+        select: { version: true, commitMessage: true, createdAt: true }
+      })
+
+      if (versions.length === 0) {
+        return { versions: [], latestSnapshot: null } satisfies HistoryListResult
+      }
+
+      const latestVersion = versions[0].version
+      const full = await prisma.documents.findFirst({
+        where: { documentId, version: latestVersion },
+        select: { data: true, version: true, commitMessage: true, createdAt: true }
+      })
+
+      return {
+        versions,
+        latestSnapshot: full ? toSnapshot(full) : null
+      } satisfies HistoryListResult
+    }
+
+    case 'history.watch': {
+      const doc = await prisma.documents.findFirst({
+        where: { documentId, version: payload.version },
+        select: { data: true, version: true, commitMessage: true, createdAt: true }
+      })
+
+      if (!doc) return null
+
+      return toSnapshot(doc)
+    }
+
+    case 'history.prev':
+      return prisma.documents.findFirst({
+        where: { documentId, version: { lt: payload.currentVersion || 0 } },
+        orderBy: { version: 'desc' }
+      })
+
+    case 'history.next':
+      return prisma.documents.findFirst({
+        where: { documentId, version: { gt: payload.currentVersion || 0 } },
+        orderBy: { version: 'asc' }
+      })
+
+    default:
+      return payload
+  }
+}