✨(y-provider) collaboration polling

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)

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<any>;
+  let res: Response;
+  let dummyDoc: any;
+  let pollSync: PollSync<any>;
+  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<any>;
+
+    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<string, any>(),
+      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();
+    });
+  });
+});

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<T>() {
+  const res: Partial<Response<T>> = {};
+  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<T>;
+}
+
+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);
+  });
+});

src/frontend/servers/y-provider/jest.config.js

@@ -7,6 +7,7 @@ var config = {
   moduleNameMapper: {
     '^@/(.*)$': '<rootDir>/../src/$1',
     '^@blocknote/server-util$': '<rootDir>/../__mocks__/mock.js',
+    '^y-protocols/awareness.js$': '<rootDir>/../__mocks__/mock.js',
   },
 };
 export default config;

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"
   },

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<CollaborationPollPostMessagePayload>,
+  res: Response<CollaborationPollPostMessageResponse>,
+) => {
+  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<CollaborationPollPostMessagePayload>(
+    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<CollaborationPollSyncDocBody>,
+  res: Response<CollaborationPollSyncDocResponse>,
+) => {
+  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<CollaborationPollSyncDocBody>(
+    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<void>,
+  res: Response<CollaborationPollSSEMessageResponse>,
+) => {
+  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<void>(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);
+};

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';

src/frontend/servers/y-provider/src/helpers.ts

@@ -1,6 +1,6 @@
-export const promiseDone = () => {
-  let done: (value: void | PromiseLike<void>) => void = () => {};
-  const promise = new Promise<void>((resolve) => {
+export const promiseDone = <T = void>() => {
+  let done: (value: T | PromiseLike<T>) => void = () => {};
+  const promise = new Promise<T>((resolve) => {
     done = resolve;
   });
 

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<T> = Request<
+  object,
+  object,
+  T,
+  PollSyncRequestQuery
+>;
+
+export class PollSync<T> {
+  public readonly canEdit: boolean;
+  public readonly req: PollSyncRequest<T>;
+  public readonly room: string;
+  private _hpDocument?: Document;
+
+  constructor(req: PollSyncRequest<T>, 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;
+  }
+}

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/',
 };

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:

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;
 };

src/frontend/yarn.lock

@@ -13934,7 +13934,7 @@ [email protected]:
   resolved "https://registry.yarnpkg.com/utils-merge/-/utils-merge-1.0.1.tgz#9f95710f50a267947b2ccc124741c1028427e713"
   integrity sha512-pMZTvIkT1d+TFGvDOqodOclx0QWkkgi6Tdoa8gC8ffGAAqz9pzPTZWAybbsHHoED/ztMtkv/VoYTYyShUn81hA==
 
-uuid@^11.0.3:
+[email protected], uuid@^11.0.3:
   version "11.0.5"
   resolved "https://registry.yarnpkg.com/uuid/-/uuid-11.0.5.tgz#07b46bdfa6310c92c3fb3953a8720f170427fc62"
   integrity sha512-508e6IcKLrhxKdBbcA2b4KQZlLVp2+J5UwQ6F7Drckkc5N9ZJwFa4TgWtsww9UG8fGHbm6gbV19TdM5pQ4GaIA==