diff --git a/apps/server/src/collaboration/collaboration.gateway.ts b/apps/server/src/collaboration/collaboration.gateway.ts index b46c13c8a..b799ce366 100644 --- a/apps/server/src/collaboration/collaboration.gateway.ts +++ b/apps/server/src/collaboration/collaboration.gateway.ts @@ -15,6 +15,7 @@ import { RedisSyncExtension, SerializedHTTPRequest, } from './extensions/redis-sync'; +import { toWebRequest } from './extensions/redis-sync/redis-sync.types'; import { WsSocketWrapper } from './extensions/redis-sync/ws-socket-wrapper'; import RedisClient from 'ioredis'; import { pack, unpack } from 'msgpackr'; @@ -98,34 +99,36 @@ export class CollaborationGateway { const serializedHTTPRequest = this.serializeRequest(request); const socketId = serializedHTTPRequest.headers['sec-websocket-key']; - // Create wrapper socket that only receives events via emit() - // This prevents double-handling since Hocuspocus won't listen to raw WebSocket events const wrappedSocket = new WsSocketWrapper(client); // Route through RedisSync extension (this calls handleConnection internally) - this.redisSync.onSocketOpen(wrappedSocket as any, serializedHTTPRequest); + this.redisSync.onSocketOpen(wrappedSocket, serializedHTTPRequest); - // Forward raw WebSocket messages to the extension client.on('message', (data: ArrayBuffer) => { - this.redisSync!.onSocketMessage( - wrappedSocket as any, - serializedHTTPRequest, - data, - ); + this.redisSync!.onSocketMessage(serializedHTTPRequest, data); }); - // Forward close events client.on('close', (code: number, reason: Buffer) => { - this.redisSync!.onSocketClose(socketId, code, reason.buffer as ArrayBuffer); - }); - - // Forward pong events for keepalive - client.on('pong', (data: Buffer) => { - wrappedSocket.emit('pong', data); + this.redisSync!.onSocketClose( + socketId, + code, + new Uint8Array(reason).buffer, + ); }); } else { // Fallback to direct Hocuspocus connection - this.hocuspocus.handleConnection(client, request); + const clientConnection = this.hocuspocus.handleConnection( + client, + toWebRequest(this.serializeRequest(request)), + ); + + client.on('message', (data: Buffer) => { + clientConnection.handleMessage(new Uint8Array(data)); + }); + + client.on('close', (code: number, reason: Buffer) => { + clientConnection.handleClose({ code, reason: reason.toString() }); + }); } } @@ -178,6 +181,7 @@ export class CollaborationGateway { if (this.hocuspocus.getDocumentsCount() === 0) resolve(''); this.hocuspocus.closeConnections(); + this.hocuspocus.flushPendingStores(); } catch (error) { console.error(error); } diff --git a/apps/server/src/collaboration/extensions/redis-sync/ws-socket-wrapper.ts b/apps/server/src/collaboration/extensions/redis-sync/ws-socket-wrapper.ts index 714111108..0ce14ed1d 100644 --- a/apps/server/src/collaboration/extensions/redis-sync/ws-socket-wrapper.ts +++ b/apps/server/src/collaboration/extensions/redis-sync/ws-socket-wrapper.ts @@ -2,8 +2,9 @@ import type WebSocket from 'ws'; import type { WebSocketLike } from '@hocuspocus/server'; /** - * Wrapper around ws WebSocket that only receives events via emit(). - * This prevents double-handling when used with RedisSyncExtension. + * Wrapper around ws WebSocket that Hocuspocus only writes to. + * Incoming socket events are forwarded separately by the gateway, + * which prevents double-handling with RedisSyncExtension. */ export class WsSocketWrapper implements WebSocketLike { private ws: WebSocket;