From 971b7e099d1d75fe3e3627275ed7285e85a7192f Mon Sep 17 00:00:00 2001 From: Anthony LC Date: Thu, 13 Feb 2025 16:45:52 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(y-provider)=20collaboration=20polling?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We can now interact with the collaboration server using http requests. It will be used as a fallback when the websocket is not working. 2 kind of requests: - to send messages to the server we use POST requests - to get messages from the server we use a GET request using SSE (Server Sent Events) --- .../y-provider/__tests__/PollSync.test.ts | 185 ++++++++++++++ .../collaborationPollHandler.test.ts | 193 +++++++++++++++ .../servers/y-provider/jest.config.js | 1 + src/frontend/servers/y-provider/package.json | 1 + .../src/handlers/collaborationPollHandler.ts | 132 ++++++++++ .../servers/y-provider/src/handlers/index.ts | 3 +- .../servers/y-provider/src/helpers.ts | 6 +- .../servers/y-provider/src/libs/PollSync.ts | 227 ++++++++++++++++++ src/frontend/servers/y-provider/src/routes.ts | 2 + .../y-provider/src/servers/appServer.ts | 20 +- src/frontend/servers/y-provider/src/utils.ts | 13 +- src/frontend/yarn.lock | 2 +- 12 files changed, 777 insertions(+), 8 deletions(-) create mode 100644 src/frontend/servers/y-provider/__tests__/PollSync.test.ts create mode 100644 src/frontend/servers/y-provider/__tests__/collaborationPollHandler.test.ts create mode 100644 src/frontend/servers/y-provider/src/handlers/collaborationPollHandler.ts create mode 100644 src/frontend/servers/y-provider/src/libs/PollSync.ts diff --git a/src/frontend/servers/y-provider/__tests__/PollSync.test.ts b/src/frontend/servers/y-provider/__tests__/PollSync.test.ts new file mode 100644 index 00000000..8cfb69a4 --- /dev/null +++ b/src/frontend/servers/y-provider/__tests__/PollSync.test.ts @@ -0,0 +1,185 @@ +/* eslint-disable @typescript-eslint/no-require-imports */ +import { EventEmitter } from 'events'; + +import { MessageType } from '@hocuspocus/server'; +import { Response } from 'express'; +import * as Y from 'yjs'; + +import { PollSync, PollSyncRequest } from '../src/libs/PollSync'; + +const { base64ToYDoc } = require('@/utils'); +const { logger } = require('@/utils'); + +const mockEncodeStateAsUpdate = jest.fn(); +jest.mock('yjs', () => ({ + ...jest.requireActual('yjs'), + encodeStateAsUpdate: () => mockEncodeStateAsUpdate(), +})); + +jest.mock('@/utils', () => ({ + base64ToYDoc: jest.fn((_b64: string) => new Y.Doc()), + toBase64: jest.fn((data: Uint8Array) => Buffer.from(data).toString('base64')), + logger: jest.fn(), +})); + +jest.mock('@hocuspocus/server', () => { + const originalModule = jest.requireActual('@hocuspocus/server'); + return { + __esModule: true, + ...originalModule, + IncomingMessage: jest.fn().mockImplementation((buf: Buffer) => ({ + buffer: buf, + readVarString: jest.fn(() => 'testRoom'), + readVarUint: jest.fn(() => MessageType.Sync), + writeVarUint: jest.fn(), + decoder: {}, + encoder: {}, + })), + }; +}); + +describe('PollSync', () => { + let req: PollSyncRequest; + let res: Response; + let dummyDoc: any; + let pollSync: PollSync; + let mockHocuspocusServer: any; + + beforeEach(() => { + req = Object.assign(new EventEmitter(), { + on: jest.fn((event, cb) => { + // For 'close' event, store the callback for manual trigger in tests. + if (event === 'close') { + req.destroy = cb; + } + }), + }) as unknown as PollSyncRequest; + + res = { + write: jest.fn(), + end: jest.fn(), + } as unknown as Response; + + // Create a dummy document with required methods/properties + dummyDoc = { + name: 'testRoom', + merge: jest.fn((_other: Y.Doc) => { + // Simulate merging by returning a new Y.Doc instance + return new Y.Doc(); + }), + getConnections: jest.fn(() => [{ handleMessage: jest.fn() }]), + awareness: { + on: jest.fn(), + off: jest.fn(), + }, + addDirectConnection: jest.fn(), + removeDirectConnection: jest.fn(), + on: jest.fn(), + off: jest.fn(), + }; + + pollSync = new PollSync(req, 'testRoom', true); + // Pre-set the document for non-init tests + (pollSync as any)._hpDocument = dummyDoc; + + // Create a dummy Hocuspocus server + mockHocuspocusServer = { + loadingDocuments: { get: jest.fn() }, + documents: new Map(), + createDocument: jest.fn(() => ({ + name: 'newDoc', + merge: (doc: Y.Doc) => doc, + })), + }; + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + describe('initHocuspocusDocument', () => { + it('should return document from loadingDocuments when available', async () => { + mockHocuspocusServer.loadingDocuments.get.mockResolvedValue(dummyDoc); + pollSync = new PollSync(req, 'testRoom', true); + const doc = await pollSync.initHocuspocusDocument(mockHocuspocusServer); + expect(doc).toBe(dummyDoc); + expect(mockHocuspocusServer.loadingDocuments.get).toHaveBeenCalledWith( + 'testRoom', + ); + }); + + it('should return document from documents when available', async () => { + mockHocuspocusServer.loadingDocuments.get.mockResolvedValue(undefined); + mockHocuspocusServer.documents.set('testRoom', dummyDoc); + pollSync = new PollSync(req, 'testRoom', false); + const doc = await pollSync.initHocuspocusDocument(mockHocuspocusServer); + expect(doc).toBe(dummyDoc); + }); + + it('should create a new document when none exists and canEdit is true', async () => { + mockHocuspocusServer.loadingDocuments.get.mockResolvedValue(undefined); + pollSync = new PollSync(req, 'testRoom', true); + const newDoc = { name: 'newDoc', merge: (doc: any) => doc }; + mockHocuspocusServer.createDocument.mockResolvedValue(newDoc); + const doc = await pollSync.initHocuspocusDocument(mockHocuspocusServer); + expect(doc).toBe(newDoc); + expect(mockHocuspocusServer.createDocument).toHaveBeenCalled(); + }); + }); + + describe('sync', () => { + it('should encode state without merging when canEdit is false', () => { + // When user cannot edit, merge should not be called. + pollSync = new PollSync(req, 'testRoom', false); + (pollSync as any)._hpDocument = dummyDoc; + mockEncodeStateAsUpdate.mockReturnValue(Uint8Array.from([1, 2, 3])); + + const result = pollSync.sync('dummyLocalDoc64'); + expect(dummyDoc.merge).not.toHaveBeenCalled(); + expect(result).toBe( + Buffer.from(Uint8Array.from([1, 2, 3])).toString('base64'), + ); + }); + + it('should merge local doc when canEdit is true', () => { + pollSync = new PollSync(req, 'testRoom', true); + (pollSync as any)._hpDocument = dummyDoc; + const localDoc = new Y.Doc(); + // Mock base64ToYDoc to return our localDoc + base64ToYDoc.mockReturnValue(localDoc); + mockEncodeStateAsUpdate.mockReturnValue(Uint8Array.from([4, 5, 6])); + pollSync.sync('localDoc64'); + expect(dummyDoc.merge).toHaveBeenCalledWith(localDoc); + }); + }); + + describe('sendClientsMessages', () => { + it('should log an error if room names do not match', () => { + // Set doc name different from what IncomingMessage returns ('testRoom') + dummyDoc.name = 'differentRoom'; + (pollSync as any)._hpDocument = dummyDoc; + const fakeMessage = Buffer.from('fakeMessage').toString('base64'); + pollSync.sendClientsMessages(fakeMessage); + expect(logger).toHaveBeenCalled(); + }); + }); + + describe('pullClientsMessages', () => { + it('should register event listeners and cleanup on request close', () => { + (pollSync as any)._hpDocument = dummyDoc; + pollSync.pullClientsMessages(res); + expect(dummyDoc.addDirectConnection).toHaveBeenCalled(); + expect(dummyDoc.on).toHaveBeenCalled(); + expect(dummyDoc.awareness.on).toHaveBeenCalled(); + + // Simulate a 'close' event on the request + if (req.destroy) { + req.destroy(); + } + expect(dummyDoc.removeDirectConnection).toHaveBeenCalled(); + // Verify that document listeners are removed (off was called) + expect(dummyDoc.off).toHaveBeenCalled(); + expect(dummyDoc.awareness.off).toHaveBeenCalled(); + }); + }); +}); diff --git a/src/frontend/servers/y-provider/__tests__/collaborationPollHandler.test.ts b/src/frontend/servers/y-provider/__tests__/collaborationPollHandler.test.ts new file mode 100644 index 00000000..f8b99e95 --- /dev/null +++ b/src/frontend/servers/y-provider/__tests__/collaborationPollHandler.test.ts @@ -0,0 +1,193 @@ +import { Response } from 'express'; + +import { + collaborationPollPostMessageHandler, + collaborationPollSSEMessageHandler, + collaborationPollSyncDocHandler, +} from '../src/handlers/collaborationPollHandler'; + +const mockInitHocuspocusDocument = jest.fn(); +const mockSendClientsMessages = jest.fn(); +const mockSync = jest.fn(); +const mockPullClientsMessages = jest.fn(); + +jest.mock('@/libs/PollSync', () => { + return { + PollSync: jest.fn().mockImplementation(() => ({ + initHocuspocusDocument: mockInitHocuspocusDocument, + sendClientsMessages: mockSendClientsMessages, + sync: mockSync, + pullClientsMessages: mockPullClientsMessages, + })), + }; +}); + +jest.mock('@/servers/hocusPocusServer', () => ({ + hocusPocusServer: {}, +})); + +// Helper function to create a mock response +function createResponse() { + const res: Partial> = {}; + res.status = jest.fn().mockReturnValue(res); + res.json = jest.fn().mockReturnValue(res); + res.setHeader = jest.fn(); + res.write = jest.fn(); + res.headersSent = false; + return res as Response; +} + +describe('collaborationPollPostMessageHandler', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockInitHocuspocusDocument.mockResolvedValue({ doc: 'exists' }); + }); + + it('should return 403 if user is not allowed to edit', async () => { + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'False' }, + body: { message64: 'testMessage' }, + } as any; + const res = createResponse(); + await collaborationPollPostMessageHandler(req, res); + expect(res.status).toHaveBeenCalledWith(403); + expect(res.json).toHaveBeenCalledWith({ error: 'Forbidden' }); + }); + + it('should return 400 if room is not provided', async () => { + const req = { + query: {}, + headers: { 'x-can-edit': 'True' }, + body: { message64: 'testMessage' }, + } as any; + const res = createResponse(); + await collaborationPollPostMessageHandler(req, res); + expect(res.status).toHaveBeenCalledWith(400); + expect(res.json).toHaveBeenCalledWith({ error: 'Room name not provided' }); + }); + + it('should return 404 if document is not found', async () => { + mockInitHocuspocusDocument.mockResolvedValue(null); + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'True' }, + body: { message64: 'testMessage' }, + } as any; + const res = createResponse(); + res.headersSent = false; + await collaborationPollPostMessageHandler(req, res); + expect(res.status).toHaveBeenCalledWith(404); + expect(res.json).toHaveBeenCalledWith({ error: 'Document not found' }); + }); + + it('should process message and return updated true when successful', async () => { + // Reset headerSent to false to simulate a proper response + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'True' }, + body: { message64: 'testMessage' }, + } as any; + const res = createResponse(); + res.headersSent = false; + await collaborationPollPostMessageHandler(req, res); + expect(mockSendClientsMessages).toHaveBeenCalledWith('testMessage'); + expect(res.status).toHaveBeenCalledWith(200); + expect(res.json).toHaveBeenCalledWith({ updated: true }); + }); +}); + +describe('collaborationPollSyncDocHandler', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockInitHocuspocusDocument.mockResolvedValue({ doc: 'exists' }); + mockSync.mockReturnValue('syncDocEncoded'); + }); + + it('should return 400 if room is not provided', async () => { + const req = { + query: {}, + headers: { 'x-can-edit': 'True' }, + body: { localDoc64: 'localDocEncoded' }, + } as any; + const res = createResponse(); + await collaborationPollSyncDocHandler(req, res); + expect(res.status).toHaveBeenCalledWith(400); + expect(res.json).toHaveBeenCalledWith({ error: 'Room name not provided' }); + }); + + it('should return 404 if document is not found', async () => { + mockInitHocuspocusDocument.mockResolvedValue(null); + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'True' }, + body: { localDoc64: 'localDocEncoded' }, + } as any; + const res = createResponse(); + await collaborationPollSyncDocHandler(req, res); + expect(res.status).toHaveBeenCalledWith(404); + expect(res.json).toHaveBeenCalledWith({ error: 'Document not found' }); + }); + + it('should sync document and return syncDoc64 when successful', async () => { + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'True' }, + body: { localDoc64: 'localDocEncoded' }, + } as any; + const res = createResponse(); + res.headersSent = false; + await collaborationPollSyncDocHandler(req, res); + expect(mockSync).toHaveBeenCalledWith('localDocEncoded'); + expect(res.status).toHaveBeenCalledWith(200); + expect(res.json).toHaveBeenCalledWith({ syncDoc64: 'syncDocEncoded' }); + }); +}); + +describe('collaborationPollSSEMessageHandler', () => { + beforeEach(() => { + jest.clearAllMocks(); + mockInitHocuspocusDocument.mockResolvedValue({ doc: 'exists' }); + }); + + it('should return 400 if room is not provided', async () => { + const req = { + query: {}, + headers: { 'x-can-edit': 'True' }, + } as any; + const res = createResponse(); + await collaborationPollSSEMessageHandler(req, res); + expect(res.status).toHaveBeenCalledWith(400); + expect(res.json).toHaveBeenCalledWith({ error: 'Room name not provided' }); + }); + + it('should return 404 if document is not found', async () => { + mockInitHocuspocusDocument.mockResolvedValue(null); + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'True' }, + } as any; + const res = createResponse(); + await collaborationPollSSEMessageHandler(req, res); + expect(res.status).toHaveBeenCalledWith(404); + expect(res.json).toHaveBeenCalledWith({ error: 'Document not found' }); + }); + + it('should set SSE headers and send connected message when successful', async () => { + const req = { + query: { room: 'test-room' }, + headers: { 'x-can-edit': 'True' }, + } as any; + const res = createResponse(); + res.headersSent = false; + await collaborationPollSSEMessageHandler(req, res); + expect(res.setHeader).toHaveBeenCalledWith( + 'Content-Type', + 'text/event-stream', + ); + expect(res.setHeader).toHaveBeenCalledWith('Cache-Control', 'no-cache'); + expect(res.setHeader).toHaveBeenCalledWith('Connection', 'keep-alive'); + expect(res.write).toHaveBeenCalledWith(': connected\n\n'); + expect(mockPullClientsMessages).toHaveBeenCalledWith(res); + }); +}); diff --git a/src/frontend/servers/y-provider/jest.config.js b/src/frontend/servers/y-provider/jest.config.js index 9ba612b3..cf854391 100644 --- a/src/frontend/servers/y-provider/jest.config.js +++ b/src/frontend/servers/y-provider/jest.config.js @@ -7,6 +7,7 @@ var config = { moduleNameMapper: { '^@/(.*)$': '/../src/$1', '^@blocknote/server-util$': '/../__mocks__/mock.js', + '^y-protocols/awareness.js$': '/../__mocks__/mock.js', }, }; export default config; diff --git a/src/frontend/servers/y-provider/package.json b/src/frontend/servers/y-provider/package.json index 1da18db2..de3a4924 100644 --- a/src/frontend/servers/y-provider/package.json +++ b/src/frontend/servers/y-provider/package.json @@ -23,6 +23,7 @@ "cors": "2.8.5", "express": "4.21.2", "express-ws": "5.0.2", + "uuid": "11.0.5", "y-protocols": "1.0.6", "yjs": "13.6.23" }, diff --git a/src/frontend/servers/y-provider/src/handlers/collaborationPollHandler.ts b/src/frontend/servers/y-provider/src/handlers/collaborationPollHandler.ts new file mode 100644 index 00000000..b5a1a769 --- /dev/null +++ b/src/frontend/servers/y-provider/src/handlers/collaborationPollHandler.ts @@ -0,0 +1,132 @@ +import { Response } from 'express'; + +import { PollSync, PollSyncRequest } from '@/libs/PollSync'; +import { hocusPocusServer } from '@/servers/hocusPocusServer'; + +interface CollaborationPollPostMessagePayload { + message64: string; +} +interface CollaborationPollPostMessageResponse { + updated?: boolean; + error?: string; +} + +export const collaborationPollPostMessageHandler = async ( + req: PollSyncRequest, + res: Response, +) => { + const room = req.query.room; + const canEdit = req.headers['x-can-edit'] === 'True'; + + // Only editors can send messages + if (!canEdit) { + res.status(403).json({ error: 'Forbidden' }); + return; + } + + if (!room) { + res.status(400).json({ error: 'Room name not provided' }); + return; + } + + const pollSynch = new PollSync( + req, + room, + canEdit, + ); + const hpDoc = await pollSynch.initHocuspocusDocument(hocusPocusServer); + + if (!res.headersSent && !hpDoc) { + res.status(404).json({ error: 'Document not found' }); + return; + } + + pollSynch.sendClientsMessages(req.body.message64); + + if (!res.headersSent) { + res.status(200).json({ updated: true }); + } +}; + +/** + * Polling way of handling collaboration + * @param req + * @param res + */ +interface CollaborationPollSyncDocResponse { + syncDoc64?: string; + error?: string; +} +interface CollaborationPollSyncDocBody { + localDoc64: string; +} + +export const collaborationPollSyncDocHandler = async ( + req: PollSyncRequest, + res: Response, +) => { + const room = req.query.room; + const canEdit = req.headers['x-can-edit'] === 'True'; + + if (!room) { + res.status(400).json({ error: 'Room name not provided' }); + return; + } + + const pollSynch = new PollSync( + req, + room, + canEdit, + ); + const hpDoc = await pollSynch.initHocuspocusDocument(hocusPocusServer); + + if (!hpDoc) { + res.status(404).json({ error: 'Document not found' }); + return; + } + + const syncDoc64 = pollSynch.sync(req.body.localDoc64); + + if (!res.headersSent) { + res.status(200).json({ syncDoc64 }); + } +}; + +/** + * SSE message handling + * @param req + * @param res + */ +interface CollaborationPollSSEMessageResponse { + updatedDoc64?: string; + stateFingerprint?: string; + awareness64?: string; + error?: string; +} +export const collaborationPollSSEMessageHandler = async ( + req: PollSyncRequest, + res: Response, +) => { + const room = req.query.room; + const canEdit = req.headers['x-can-edit'] === 'True'; + + if (!room) { + res.status(400).json({ error: 'Room name not provided' }); + return; + } + + const pollSynch = new PollSync(req, room, canEdit); + const hpDoc = await pollSynch.initHocuspocusDocument(hocusPocusServer); + + if (!hpDoc) { + res.status(404).json({ error: 'Document not found' }); + return; + } + + res.setHeader('Content-Type', 'text/event-stream'); + res.setHeader('Cache-Control', 'no-cache'); + res.setHeader('Connection', 'keep-alive'); + res.write(': connected\n\n'); + + pollSynch.pullClientsMessages(res); +}; diff --git a/src/frontend/servers/y-provider/src/handlers/index.ts b/src/frontend/servers/y-provider/src/handlers/index.ts index 75bd7f7b..5b14542b 100644 --- a/src/frontend/servers/y-provider/src/handlers/index.ts +++ b/src/frontend/servers/y-provider/src/handlers/index.ts @@ -1,3 +1,4 @@ export * from './collaborationResetConnectionsHandler'; -export * from './collaborationWSHandler'; +export * from './collaborationPollHandler'; export * from './convertMarkdownHandler'; +export * from './collaborationWSHandler'; diff --git a/src/frontend/servers/y-provider/src/helpers.ts b/src/frontend/servers/y-provider/src/helpers.ts index c23b8368..2a3bd033 100644 --- a/src/frontend/servers/y-provider/src/helpers.ts +++ b/src/frontend/servers/y-provider/src/helpers.ts @@ -1,6 +1,6 @@ -export const promiseDone = () => { - let done: (value: void | PromiseLike) => void = () => {}; - const promise = new Promise((resolve) => { +export const promiseDone = () => { + let done: (value: T | PromiseLike) => void = () => {}; + const promise = new Promise((resolve) => { done = resolve; }); diff --git a/src/frontend/servers/y-provider/src/libs/PollSync.ts b/src/frontend/servers/y-provider/src/libs/PollSync.ts new file mode 100644 index 00000000..7d262d43 --- /dev/null +++ b/src/frontend/servers/y-provider/src/libs/PollSync.ts @@ -0,0 +1,227 @@ +/* eslint-disable @typescript-eslint/no-unused-vars */ +import crypto from 'crypto'; + +import { + AwarenessUpdate, + Document, + Hocuspocus, + IncomingMessage, + MessageType, + OutgoingMessage, +} from '@hocuspocus/server'; +import { Request, Response } from 'express'; +import { v4 as uuid } from 'uuid'; +import { applyAwarenessUpdate } from 'y-protocols/awareness.js'; +import { readSyncMessage } from 'y-protocols/sync'; +import * as Y from 'yjs'; + +import { base64ToYDoc, logger, toBase64 } from '@/utils'; + +export type PollSyncRequestQuery = { + room?: string; +}; + +export type PollSyncRequest = Request< + object, + object, + T, + PollSyncRequestQuery +>; + +export class PollSync { + public readonly canEdit: boolean; + public readonly req: PollSyncRequest; + public readonly room: string; + private _hpDocument?: Document; + + constructor(req: PollSyncRequest, room: string, canEdit: boolean) { + this.room = room; + this.canEdit = canEdit; + this.req = req; + } + + public get hpDocument() { + return this._hpDocument; + } + + public async initHocuspocusDocument(hocusPocusServer: Hocuspocus) { + const { req, room, canEdit } = this; + this._hpDocument = await hocusPocusServer.loadingDocuments.get(room); + + if (this._hpDocument) { + return this._hpDocument; + } + + this._hpDocument = hocusPocusServer.documents.get(room); + + if (this._hpDocument || (!this._hpDocument && !canEdit)) { + return this._hpDocument; + } + + /** + * createDocument is used to create a new document if it does not exist. + * If the document exists, it will return the existing document. + */ + this._hpDocument = await hocusPocusServer.createDocument( + room, + req, + uuid(), + { + readOnly: false, + requiresAuthentication: false, + isAuthenticated: false, + }, + ); + + return this._hpDocument; + } + + /** + * Sync the document with the latest changes + * + * @param localDoc64 + * @returns + */ + public sync(localDoc64: string): string | undefined { + const hpDoc = this.getHpDocument(); + let syncYDoc = hpDoc; + + // Merge the coming document with the latest changes (only if the user can edit) + if (this.canEdit) { + const localDoc = base64ToYDoc(localDoc64); + syncYDoc = hpDoc.merge(localDoc); + } + + return toBase64(Y.encodeStateAsUpdate(syncYDoc)); + } + + /** + * Create a hash SHA-256 of the state vector of the document. + * Usefull to compare the state of the document. + * @param doc + * @returns + */ + protected getStateFingerprint(doc: Y.Doc): string { + const stateVector = Y.encodeStateVector(doc); + return crypto.createHash('sha256').update(stateVector).digest('base64'); // or 'hex' + } + + /** + * Send messages to other clients + */ + public sendClientsMessages(message64: string) { + const hpDoc = this.getHpDocument(); + const messageBuffer = Buffer.from(message64, 'base64'); + const message = new IncomingMessage(messageBuffer); + const room = message.readVarString(); + + if (hpDoc.name !== room) { + logger('Send messages problem, room different', room, hpDoc.name); + return; + } + + // We write the sync to the current doc - it will propagate to others by itself + const type = message.readVarUint() as MessageType; + if (type === MessageType.Sync) { + message.writeVarUint(MessageType.Sync); + readSyncMessage(message.decoder, message.encoder, hpDoc, null); + } else if (type === MessageType.Awareness) { + const awarenessUpdate = message.readVarUint8Array(); + applyAwarenessUpdate( + hpDoc.awareness, + awarenessUpdate, + hpDoc.awareness.clientID, + ); + } else { + hpDoc.getConnections().forEach((connection) => { + connection.handleMessage(messageBuffer); + }); + } + } + + /** + * Pull messages from other clients + * + * We listen 2 kind of messages: + * - Document updates (change in the document) + * - Awareness messages (cursor, selection, etc.) + * + * @param res + */ + public pullClientsMessages(res: Response) { + const hpDoc = this.getHpDocument(); + hpDoc.addDirectConnection(); + + const updateMessagesFn = ( + update: Uint8Array, + _origin: string, + updatedDoc: Y.Doc, + _transaction: Y.Transaction, + ) => { + res.write( + `data: ${JSON.stringify({ + time: new Date(), + updatedDoc64: toBase64(update), + stateFingerprint: this.getStateFingerprint(updatedDoc), + })}\n\n`, + ); + }; + + const destroyFn = (updatedDoc: Y.Doc) => { + res.write( + `data: ${JSON.stringify({ + time: new Date(), + updatedDoc64: undefined, + stateFingerprint: this.getStateFingerprint(updatedDoc), + })}\n\n`, + ); + + hpDoc.off('destroy', destroyFn); + hpDoc.off('update', updateMessagesFn); + + // Close the connection + res.end(); + }; + + const updateAwarenessFn = ({ + added, + updated, + removed, + }: AwarenessUpdate) => { + const changedClients = added.concat(updated, removed); + const awarenessMessage = new OutgoingMessage( + this.room, + ).createAwarenessUpdateMessage(hpDoc.awareness, changedClients); + + res.write( + `data: ${JSON.stringify({ + time: new Date(), + awareness64: toBase64(awarenessMessage.toUint8Array()), + stateFingerprint: this.getStateFingerprint(hpDoc), + })}\n\n`, + ); + }; + + hpDoc.awareness.off('update', updateAwarenessFn); + hpDoc.awareness.on('update', updateAwarenessFn); + hpDoc.off('update', updateMessagesFn); + hpDoc.off('destroy', destroyFn); + hpDoc.on('update', updateMessagesFn); + hpDoc.on('destroy', destroyFn); + + this.req.on('close', () => { + hpDoc.off('update', updateMessagesFn); + hpDoc.off('destroy', destroyFn); + hpDoc.awareness.off('update', updateAwarenessFn); + hpDoc.removeDirectConnection(); + }); + } + + protected getHpDocument() { + if (!this.hpDocument) { + throw new Error('HocusPocus document not initialized'); + } + + return this.hpDocument; + } +} diff --git a/src/frontend/servers/y-provider/src/routes.ts b/src/frontend/servers/y-provider/src/routes.ts index 98803b87..44b294e8 100644 --- a/src/frontend/servers/y-provider/src/routes.ts +++ b/src/frontend/servers/y-provider/src/routes.ts @@ -1,5 +1,7 @@ export const routes = { COLLABORATION_WS: '/collaboration/ws/', + COLLABORATION_POLL_MESSAGE: '/collaboration/ws/poll/message/', + COLLABORATION_POLL_SYNC: '/collaboration/ws/poll/sync/', COLLABORATION_RESET_CONNECTIONS: '/collaboration/api/reset-connections/', CONVERT_MARKDOWN: '/api/convert-markdown/', }; diff --git a/src/frontend/servers/y-provider/src/servers/appServer.ts b/src/frontend/servers/y-provider/src/servers/appServer.ts index 80077bb8..804378ab 100644 --- a/src/frontend/servers/y-provider/src/servers/appServer.ts +++ b/src/frontend/servers/y-provider/src/servers/appServer.ts @@ -6,6 +6,9 @@ import expressWebsockets from 'express-ws'; import { PORT } from '../env'; import { + collaborationPollPostMessageHandler, + collaborationPollSSEMessageHandler, + collaborationPollSyncDocHandler, collaborationResetConnectionsHandler, collaborationWSHandler, convertMarkdownHandler, @@ -27,9 +30,24 @@ export const initServer = () => { app.use(corsMiddleware); /** - * Route to handle WebSocket connections + * Routes to handle collaboration connections */ app.ws(routes.COLLABORATION_WS, wsSecurity, collaborationWSHandler); + app.get( + routes.COLLABORATION_POLL_MESSAGE, + httpSecurity, + collaborationPollSSEMessageHandler, + ); + app.post( + routes.COLLABORATION_POLL_MESSAGE, + httpSecurity, + collaborationPollPostMessageHandler, + ); + app.post( + routes.COLLABORATION_POLL_SYNC, + httpSecurity, + collaborationPollSyncDocHandler, + ); /** * Route to reset connections in a room: diff --git a/src/frontend/servers/y-provider/src/utils.ts b/src/frontend/servers/y-provider/src/utils.ts index 847c5568..2bc62bb5 100644 --- a/src/frontend/servers/y-provider/src/utils.ts +++ b/src/frontend/servers/y-provider/src/utils.ts @@ -1,3 +1,5 @@ +import * as Y from 'yjs'; + import { COLLABORATION_LOGGING } from './env'; // eslint-disable-next-line @typescript-eslint/no-explicit-any @@ -8,6 +10,13 @@ export function logger(...args: any[]) { } } -export const toBase64 = function (str: Uint8Array) { - return Buffer.from(str).toString('base64'); +export const toBase64 = function (uInt8Array: Uint8Array) { + return Buffer.from(uInt8Array).toString('base64'); +}; + +export const base64ToYDoc = (base64: string) => { + const uint8Array = Buffer.from(base64, 'base64'); + const ydoc = new Y.Doc(); + Y.applyUpdate(ydoc, uint8Array); + return ydoc; }; diff --git a/src/frontend/yarn.lock b/src/frontend/yarn.lock index 38330b15..0a8f9460 100644 --- a/src/frontend/yarn.lock +++ b/src/frontend/yarn.lock @@ -13934,7 +13934,7 @@ utils-merge@1.0.1: resolved "https://registry.yarnpkg.com/utils-merge/-/utils-merge-1.0.1.tgz#9f95710f50a267947b2ccc124741c1028427e713" integrity sha512-pMZTvIkT1d+TFGvDOqodOclx0QWkkgi6Tdoa8gC8ffGAAqz9pzPTZWAybbsHHoED/ztMtkv/VoYTYyShUn81hA== -uuid@^11.0.3: +uuid@11.0.5, uuid@^11.0.3: version "11.0.5" resolved "https://registry.yarnpkg.com/uuid/-/uuid-11.0.5.tgz#07b46bdfa6310c92c3fb3953a8720f170427fc62" integrity sha512-508e6IcKLrhxKdBbcA2b4KQZlLVp2+J5UwQ6F7Drckkc5N9ZJwFa4TgWtsww9UG8fGHbm6gbV19TdM5pQ4GaIA==