snapshot of https://github.com/docmost/docmost.git @ 6205bbeb908fe846f87dd6a2cbf562e24777db38 (2026-09-15)

apps/server/src/collaboration/adapter/collab-ws.adapter.ts

@@ -0,0 +1,51 @@
+import { WebSocketServer } from 'ws';
+
+export class CollabWsAdapter {
+  private readonly wss: WebSocketServer;
+
+  constructor() {
+    this.wss = new WebSocketServer({ noServer: true });
+  }
+
+  handleUpgrade(path: string, httpServer: any) {
+    httpServer.on('upgrade', (request: any, socket: any, head: any) => {
+      try {
+        const baseUrl = 'ws://' + request.headers.host + '/';
+        const pathname = new URL(request.url, baseUrl).pathname;
+
+        if (pathname === path) {
+          this.wss.handleUpgrade(request, socket, head, (ws) => {
+            this.wss.emit('connection', ws, request);
+          });
+        } else if (pathname === '/socket.io/') {
+          return;
+        } else {
+          socket.destroy();
+        }
+      } catch (err) {
+        socket.end('HTTP/1.1 400\r\n' + (err as Error).message);
+      }
+    });
+
+    return this.wss;
+  }
+
+  public close() {
+    try {
+      this.wss.close();
+    } catch (err) {
+      console.error(err);
+    }
+  }
+
+  public destroy() {
+    try {
+      this.wss.close();
+      this.wss.clients.forEach((client) => {
+        client.terminate();
+      });
+    } catch (err) {
+      console.error(err);
+    }
+  }
+}

apps/server/src/collaboration/collaboration.gateway.ts

@@ -0,0 +1,194 @@
+import { Hocuspocus } from '@hocuspocus/server';
+import { IncomingMessage } from 'http';
+import WebSocket from 'ws';
+import { AuthenticationExtension } from './extensions/authentication.extension';
+import { PersistenceExtension } from './extensions/persistence.extension';
+import { Injectable } from '@nestjs/common';
+import { EnvironmentService } from '../integrations/environment/environment.service';
+import {
+  createRetryStrategy,
+  parseRedisUrl,
+  RedisConfig,
+} from '../common/helpers';
+import { LoggerExtension } from './extensions/logger.extension';
+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';
+import { nanoid } from 'nanoid';
+import * as os from 'node:os';
+import { CollabWsAdapter } from './adapter/collab-ws.adapter';
+import {
+  CollaborationHandler,
+  CollabEventHandlers,
+} from './collaboration.handler';
+
+@Injectable()
+export class CollaborationGateway {
+  private readonly hocuspocus: Hocuspocus;
+  private redisConfig: RedisConfig;
+  // @ts-ignore
+  private readonly redisSync: RedisSyncExtension<CollabEventHandlers> | null =
+    null;
+  private readonly withRedis: boolean;
+
+  constructor(
+    private authenticationExtension: AuthenticationExtension,
+    private persistenceExtension: PersistenceExtension,
+    private loggerExtension: LoggerExtension,
+    private environmentService: EnvironmentService,
+    private collabEventsService: CollaborationHandler,
+  ) {
+    this.redisConfig = parseRedisUrl(this.environmentService.getRedisUrl());
+    this.withRedis = !this.environmentService.isCollabDisableRedis();
+
+    this.hocuspocus = new Hocuspocus({
+      debounce: 10000,
+      maxDebounce: 45000,
+      unloadImmediately: false,
+      extensions: [
+        this.authenticationExtension,
+        this.persistenceExtension,
+        this.loggerExtension,
+      ],
+    });
+
+    if (this.withRedis) {
+      // @ts-ignore
+      this.redisSync = new RedisSyncExtension({
+        redis: new RedisClient({
+          host: this.redisConfig.host,
+          port: this.redisConfig.port,
+          username: this.redisConfig.username,
+          password: this.redisConfig.password,
+          db: this.redisConfig.db,
+          family: this.redisConfig.family,
+          tls: this.redisConfig.tls,
+          retryStrategy: createRetryStrategy(),
+        }),
+        serverId: `collab-${os?.hostname()}-${nanoid(10)}`,
+        prefix: 'collab',
+        pack,
+        unpack,
+        // @ts-ignore
+        customEvents: this.collabEventsService.getHandlers(this.hocuspocus),
+      });
+      this.hocuspocus.configuration.extensions.push(this.redisSync);
+      // @ts-ignore
+      this.redisSync.onConfigure({ instance: this.hocuspocus });
+    }
+  }
+
+  private serializeRequest(request: IncomingMessage): SerializedHTTPRequest {
+    return {
+      method: request.method ?? 'GET',
+      url: request.url ?? '/',
+      headers: {
+        'sec-websocket-key': request.headers['sec-websocket-key'] ?? '',
+        'sec-websocket-protocol':
+          request.headers['sec-websocket-protocol'] ?? '',
+      },
+      socket: { remoteAddress: request.socket?.remoteAddress ?? '' },
+    };
+  }
+
+  handleConnection(client: WebSocket, request: IncomingMessage): any {
+    if (this.redisSync) {
+      const serializedHTTPRequest = this.serializeRequest(request);
+      const socketId = serializedHTTPRequest.headers['sec-websocket-key'];
+
+      const wrappedSocket = new WsSocketWrapper(client);
+
+      // Route through RedisSync extension (this calls handleConnection internally)
+      this.redisSync.onSocketOpen(wrappedSocket, serializedHTTPRequest);
+
+      client.on('message', (data: ArrayBuffer) => {
+        this.redisSync!.onSocketMessage(serializedHTTPRequest, data);
+      });
+
+      client.on('close', (code: number, reason: Buffer) => {
+        this.redisSync!.onSocketClose(
+          socketId,
+          code,
+          new Uint8Array(reason).buffer,
+        );
+      });
+    } else {
+      // Fallback to direct Hocuspocus connection
+      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() });
+      });
+    }
+  }
+
+  getConnectionCount() {
+    return this.hocuspocus.getConnectionsCount();
+  }
+
+  getDocumentCount() {
+    return this.hocuspocus.getDocumentsCount();
+  }
+
+  handleYjsEvent<TName extends keyof CollabEventHandlers>(
+    eventName: TName,
+    documentName: string,
+    payload: Parameters<CollabEventHandlers[TName]>[1],
+  ) {
+    return this.redisSync?.handleEvent(eventName, documentName, payload);
+  }
+
+  openDirectConnection(documentName: string, context?: any) {
+    return this.hocuspocus.openDirectConnection(documentName, context);
+  }
+
+  /*
+   *Can be used before calling openDirectConnection directly
+   */
+  async lockDocument(documentName: string) {
+    return this.redisSync.lockDocument(documentName);
+  }
+
+  /*
+   *Releases a document lock and stops the interval that maintains it.
+   */
+  async releaseLock(documentName: string) {
+    return this.redisSync.releaseLock(documentName);
+  }
+
+  async destroy(collabWsAdapter: CollabWsAdapter): Promise<void> {
+    // eslint-disable-next-line no-async-promise-executor
+    await new Promise(async (resolve) => {
+      try {
+        // Wait for all documents to unload
+        this.hocuspocus.configuration.extensions.push({
+          async afterUnloadDocument({ instance }) {
+            if (instance.getDocumentsCount() === 0) resolve('');
+          },
+        });
+
+        collabWsAdapter?.close();
+
+        if (this.hocuspocus.getDocumentsCount() === 0) resolve('');
+        this.hocuspocus.closeConnections();
+        this.hocuspocus.flushPendingStores();
+      } catch (error) {
+        console.error(error);
+      }
+    });
+
+    await this.hocuspocus.hooks('onDestroy', { instance: this.hocuspocus });
+  }
+}

apps/server/src/collaboration/collaboration.handler.ts

@@ -0,0 +1,134 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Hocuspocus, Document } from '@hocuspocus/server';
+import { TiptapTransformer } from '@hocuspocus/transformer';
+import {
+  prosemirrorNodeToYElement,
+  tiptapExtensions,
+} from './collaboration.util';
+import { setYjsMark, updateYjsMarkAttribute, YjsSelection } from './yjs.util';
+import * as Y from 'yjs';
+import { User } from '@docmost/db/types/entity.types';
+
+export type CollabEventHandlers = ReturnType<
+  CollaborationHandler['getHandlers']
+>;
+
+@Injectable()
+export class CollaborationHandler {
+  private readonly logger = new Logger(CollaborationHandler.name);
+
+  constructor() {}
+
+  getHandlers(hocuspocus: Hocuspocus) {
+    return {
+      alterState: async (documentName: string, payload: { pageId: string }) => {
+        // dummy
+        // this.logger.log('Processing', documentName, payload);
+        // await this.withYdocConnection(hocuspocus, documentName, {}, (doc) => {
+        //   const fragment = doc.getXmlFragment('default');
+        //});
+      },
+      setCommentMark: async (
+        documentName: string,
+        payload: {
+          yjsSelection: YjsSelection;
+          commentId: string;
+          resolved: boolean;
+          user: User;
+        },
+      ) => {
+        const { yjsSelection, commentId, resolved, user } = payload;
+        await this.withYdocConnection(
+          hocuspocus,
+          documentName,
+          { user },
+          (doc) => {
+            const fragment = doc.getXmlFragment('default');
+            setYjsMark(doc, fragment, yjsSelection, 'comment', {
+              commentId,
+              resolved,
+            });
+          },
+        );
+      },
+      resolveCommentMark: async (
+        documentName: string,
+        payload: {
+          commentId: string;
+          resolved: boolean;
+          user: User;
+        },
+      ) => {
+        const { commentId, resolved, user } = payload;
+        await this.withYdocConnection(
+          hocuspocus,
+          documentName,
+          { user },
+          (doc) => {
+            const fragment = doc.getXmlFragment('default');
+            updateYjsMarkAttribute(
+              fragment,
+              'comment',
+              { name: 'commentId', value: commentId },
+              { resolved },
+            );
+          },
+        );
+      },
+      updatePageContent: async (
+        documentName: string,
+        payload: {
+          prosemirrorJson: any;
+          operation: string;
+          user: User;
+        },
+      ) => {
+        const { prosemirrorJson, operation, user } = payload;
+        this.logger.debug('Updating page content via yjs', documentName);
+        await this.withYdocConnection(
+          hocuspocus,
+          documentName,
+          { user },
+          (doc) => {
+            const fragment = doc.getXmlFragment('default');
+
+            if (operation === 'replace') {
+              if (fragment.length > 0) {
+                fragment.delete(0, fragment.length);
+              }
+
+              const newDoc = TiptapTransformer.toYdoc(
+                prosemirrorJson,
+                'default',
+                tiptapExtensions,
+              );
+              Y.applyUpdate(doc, Y.encodeStateAsUpdate(newDoc));
+            } else {
+              const newContent = prosemirrorJson.content || [];
+              const yElements = newContent.map(prosemirrorNodeToYElement);
+              const position = operation === 'prepend' ? 0 : fragment.length;
+              fragment.insert(position, yElements);
+            }
+          },
+        );
+      },
+    };
+  }
+
+  async withYdocConnection(
+    hocuspocus: Hocuspocus,
+    documentName: string,
+    context: any = {},
+    fn: (doc: Document) => void,
+  ): Promise<void> {
+    const connection = await hocuspocus.openDirectConnection(
+      documentName,
+      context,
+    );
+    try {
+      await connection.transact(fn);
+    } finally {
+      await connection.disconnect();
+    }
+  }
+}

apps/server/src/collaboration/collaboration.module.ts

@@ -0,0 +1,74 @@
+import { Logger, Module, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { AuthenticationExtension } from './extensions/authentication.extension';
+import { PersistenceExtension } from './extensions/persistence.extension';
+import { CollaborationGateway } from './collaboration.gateway';
+import { HttpAdapterHost } from '@nestjs/core';
+import { CollabWsAdapter } from './adapter/collab-ws.adapter';
+import { IncomingMessage } from 'http';
+import { WebSocket } from 'ws';
+import { TokenModule } from '../core/auth/token.module';
+import { HistoryProcessor } from './processors/history.processor';
+import { LoggerExtension } from './extensions/logger.extension';
+import { CollaborationHandler } from './collaboration.handler';
+import { CollabHistoryService } from './services/collab-history.service';
+import { WatcherModule } from '../core/watcher/watcher.module';
+import { TransclusionService } from '../core/page/transclusion/transclusion.service';
+import { TransclusionModule } from '../core/page/transclusion/transclusion.module';
+import { StorageModule } from '../integrations/storage/storage.module';
+import { EnvironmentModule } from '../integrations/environment/environment.module';
+
+@Module({
+  providers: [
+    CollaborationGateway,
+    AuthenticationExtension,
+    PersistenceExtension,
+    LoggerExtension,
+    HistoryProcessor,
+    CollabHistoryService,
+    CollaborationHandler,
+    TransclusionService,
+  ],
+  exports: [CollaborationGateway],
+  imports: [
+    TokenModule,
+    WatcherModule,
+    StorageModule.forRootAsync({
+      imports: [EnvironmentModule],
+    }),
+    TransclusionModule,
+  ],
+})
+export class CollaborationModule implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(CollaborationModule.name);
+  private collabWsAdapter: CollabWsAdapter;
+  private path = '/collab';
+
+  constructor(
+    private readonly collaborationGateway: CollaborationGateway,
+    private readonly httpAdapterHost: HttpAdapterHost,
+  ) {}
+
+  onModuleInit() {
+    this.collabWsAdapter = new CollabWsAdapter();
+    const httpServer = this.httpAdapterHost.httpAdapter.getHttpServer();
+
+    const wss = this.collabWsAdapter.handleUpgrade(this.path, httpServer);
+
+    wss.on('connection', (client: WebSocket, request: IncomingMessage) => {
+      this.collaborationGateway.handleConnection(client, request);
+
+      client.on('error', (error) => {
+        this.logger.error('WebSocket client error:', error);
+      });
+    });
+
+    wss.on('error', (error) =>
+      this.logger.error('WebSocket server error:', error),
+    );
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.collaborationGateway?.destroy(this.collabWsAdapter);
+    this.collabWsAdapter?.destroy();
+  }
+}

apps/server/src/collaboration/collaboration.util.ts

@@ -0,0 +1,249 @@
+import { StarterKit } from '@tiptap/starter-kit';
+import { Document } from '@tiptap/extension-document';
+import { TextAlign } from '@tiptap/extension-text-align';
+import { Superscript } from '@tiptap/extension-superscript';
+import SubScript from '@tiptap/extension-subscript';
+import { Typography } from '@tiptap/extension-typography';
+import { TextStyle } from '@tiptap/extension-text-style';
+import { Color } from '@tiptap/extension-color';
+import { Youtube } from '@tiptap/extension-youtube';
+import { TaskList, TaskItem } from '@tiptap/extension-list';
+import {
+  Heading,
+  Callout,
+  Comment,
+  CustomCodeBlock,
+  Details,
+  DetailsContent,
+  DetailsSummary,
+  LinkExtension,
+  MathBlock,
+  MathInline,
+  TableHeader,
+  TableCell,
+  TableRow,
+  CustomTable,
+  TiptapImage,
+  TiptapVideo,
+  TiptapAudio,
+  TiptapPdf,
+  PageBreak,
+  TrailingNode,
+  Attachment,
+  Drawio,
+  Excalidraw,
+  Embed,
+  Mention,
+  Subpages,
+  Highlight,
+  Indent,
+  UniqueID,
+  Columns,
+  Column,
+  Status,
+  addUniqueIdsToDoc,
+  htmlToMarkdown,
+  TransclusionSource,
+  TransclusionReference,
+  BaseEmbed,
+  Footnotes,
+  Footnote,
+  FootnoteReference,
+} from '@docmost/editor-ext';
+import {
+  extensions as coreExtensions,
+  generateText,
+  getSchema,
+  JSONContent,
+} from '@tiptap/core';
+import { generateHTML, generateJSON } from '../common/helpers/prosemirror/html';
+import { collapseBlankLines } from '../common/helpers';
+// @tiptap/html library works best for generating prosemirror json state but not HTML
+// see: https://github.com/ueberdosis/tiptap/issues/5352
+// see:https://github.com/ueberdosis/tiptap/issues/4089
+//import { generateJSON } from '@tiptap/html';
+import { Node, Schema } from '@tiptap/pm/model';
+import * as Y from 'yjs';
+import { Logger } from '@nestjs/common';
+
+export const tiptapExtensions = [
+  coreExtensions.TextDirection.configure({ direction: 'auto' }),
+  StarterKit.configure({
+    document: false,
+    codeBlock: false,
+    link: false,
+    trailingNode: false,
+    heading: false,
+  }),
+  Document.extend({
+    content: 'block+ footnotes?',
+  }),
+  Heading,
+  UniqueID.configure({
+    types: ['heading', 'paragraph', 'transclusionSource'],
+  }),
+  Comment,
+  TextAlign.configure({ types: ['heading', 'paragraph'] }),
+  Indent,
+  TaskList,
+  TaskItem.configure({
+    nested: true,
+  }),
+  LinkExtension,
+  Superscript,
+  SubScript,
+  Highlight,
+  Typography,
+  TrailingNode,
+  TextStyle,
+  Color,
+  MathInline,
+  MathBlock,
+  Details,
+  DetailsContent,
+  DetailsSummary,
+  CustomTable,
+  TableCell,
+  TableRow,
+  TableHeader,
+  Youtube,
+  TiptapImage,
+  TiptapVideo,
+  TiptapAudio,
+  TiptapPdf,
+  PageBreak,
+  Callout,
+  Attachment,
+  CustomCodeBlock,
+  Drawio,
+  Excalidraw,
+  Embed,
+  Mention,
+  Subpages,
+  Columns,
+  Column,
+  Status,
+  TransclusionSource,
+  TransclusionReference,
+  BaseEmbed,
+  Footnotes,
+  Footnote,
+  FootnoteReference,
+] as any;
+
+export function jsonToHtml(tiptapJson: any) {
+  return generateHTML(tiptapJson, tiptapExtensions);
+}
+
+export function htmlToJson(html: string) {
+  const pmJson = generateJSON(html, tiptapExtensions);
+
+  try {
+    return addUniqueIdsToDoc(pmJson, tiptapExtensions);
+  } catch (error) {
+    console.warn('failed to add unique ids to doc', error);
+    return pmJson;
+  }
+}
+
+export function jsonToText(tiptapJson: JSONContent) {
+  return collapseBlankLines(generateText(tiptapJson, tiptapExtensions));
+}
+
+export function jsonToNode(tiptapJson: JSONContent) {
+  const schema = getSchema(tiptapExtensions);
+  try {
+    return Node.fromJSON(schema, tiptapJson);
+  } catch (error) {
+    if (
+      error instanceof RangeError &&
+      error.message.includes('Unknown node type')
+    ) {
+      Logger.warn('Stripping unknown node types from document:', error.message);
+      const cleanedJson = stripUnknownNodes(tiptapJson, schema);
+      return Node.fromJSON(schema, cleanedJson);
+    }
+    throw error;
+  }
+}
+
+export function getPageId(documentName: string) {
+  return documentName.split('.')[1];
+}
+
+export function isEmptyParagraphDoc(tiptapJson: JSONContent): boolean {
+  if (!tiptapJson || tiptapJson.type !== 'doc') return false;
+  const content = tiptapJson.content;
+  if (!Array.isArray(content) || content.length !== 1) return false;
+  const child = content[0];
+  if (!child || child.type !== 'paragraph') return false;
+  return (
+    !child.content ||
+    (Array.isArray(child.content) && child.content.length === 0)
+  );
+}
+
+function stripUnknownNodes(
+  json: JSONContent,
+  schema: Schema,
+): JSONContent | null {
+  if (!json || typeof json !== 'object') return json;
+
+  // Recursively clean children first, flattening any unwrapped content
+  if (json.content && Array.isArray(json.content)) {
+    const newContent: JSONContent[] = [];
+    for (const child of json.content) {
+      const cleaned = stripUnknownNodes(child, schema);
+      if (Array.isArray(cleaned)) {
+        newContent.push(...cleaned);
+      } else if (cleaned) {
+        newContent.push(cleaned);
+      }
+    }
+    json.content = newContent;
+  }
+
+  // Check if this node is unknown AFTER processing children
+  if (json.type && !schema.nodes[json.type]) {
+    // Unwrap: return cleaned children directly instead of wrapping
+    return (
+      json.content && json.content.length > 0 ? json.content : null
+    ) as any;
+  }
+
+  return json;
+}
+
+export function prosemirrorNodeToYElement(node: any): Y.XmlElement | Y.XmlText {
+  if (node.type === 'text') {
+    const ytext = new Y.XmlText();
+    ytext.insert(0, node.text || '');
+    if (node.marks?.length > 0) {
+      const attrs: Record<string, any> = {};
+      for (const mark of node.marks) {
+        attrs[mark.type] = mark.attrs || true;
+      }
+      ytext.format(0, node.text?.length || 0, attrs);
+    }
+    return ytext;
+  }
+
+  const element = new Y.XmlElement(node.type);
+  if (node.attrs) {
+    for (const [key, value] of Object.entries(node.attrs)) {
+      if (value !== null && value !== undefined) {
+        element.setAttribute(key, value as any);
+      }
+    }
+  }
+  if (node.content?.length > 0) {
+    const children = node.content.map(prosemirrorNodeToYElement);
+    element.insert(0, children);
+  }
+  return element;
+}
+
+export function jsonToMarkdown(tiptapJson: any): string {
+  const html = jsonToHtml(tiptapJson);
+  return htmlToMarkdown(html);
+}

apps/server/src/collaboration/constants.ts

@@ -0,0 +1,3 @@
+export const HISTORY_INTERVAL = 5 * 60 * 1000;
+export const HISTORY_FAST_INTERVAL = 60 * 1000;
+export const HISTORY_FAST_THRESHOLD = 5 * 60 * 1000;

apps/server/src/collaboration/extensions/authentication.extension.ts

@@ -0,0 +1,110 @@
+import { Extension, onAuthenticatePayload } from '@hocuspocus/server';
+import {
+  Injectable,
+  Logger,
+  NotFoundException,
+  UnauthorizedException,
+} from '@nestjs/common';
+import { TokenService } from '../../core/auth/services/token.service';
+import { UserRepo } from '@docmost/db/repos/user/user.repo';
+import { PageRepo } from '@docmost/db/repos/page/page.repo';
+import { SpaceMemberRepo } from '@docmost/db/repos/space/space-member.repo';
+import { PagePermissionRepo } from '@docmost/db/repos/page/page-permission.repo';
+import { findHighestUserSpaceRole } from '@docmost/db/repos/space/utils';
+import { SpaceRole } from '../../common/helpers/types/permission';
+import { isUserDisabled } from '../../common/helpers';
+import { getPageId } from '../collaboration.util';
+import { JwtCollabPayload, JwtType } from '../../core/auth/dto/jwt-payload';
+
+@Injectable()
+export class AuthenticationExtension implements Extension {
+  private readonly logger = new Logger(AuthenticationExtension.name);
+
+  constructor(
+    private tokenService: TokenService,
+    private userRepo: UserRepo,
+    private pageRepo: PageRepo,
+    private readonly spaceMemberRepo: SpaceMemberRepo,
+    private readonly pagePermissionRepo: PagePermissionRepo,
+  ) {}
+
+  async onAuthenticate(data: onAuthenticatePayload) {
+    const { documentName, token } = data;
+    const pageId = getPageId(documentName);
+
+    let jwtPayload: JwtCollabPayload;
+
+    try {
+      jwtPayload = await this.tokenService.verifyJwt(token, JwtType.COLLAB);
+    } catch (error) {
+      throw new UnauthorizedException('Invalid collab token');
+    }
+
+    const userId = jwtPayload.sub;
+    const workspaceId = jwtPayload.workspaceId;
+
+    const user = await this.userRepo.findById(userId, workspaceId);
+
+    if (!user) {
+      throw new UnauthorizedException();
+    }
+
+    if (isUserDisabled(user)) {
+      throw new UnauthorizedException();
+    }
+
+    const page = await this.pageRepo.findById(pageId);
+    if (!page) {
+      this.logger.debug(`Page not found: ${pageId}`);
+      throw new NotFoundException('Page not found');
+    }
+
+    const userSpaceRoles = await this.spaceMemberRepo.getUserSpaceRoles(
+      user.id,
+      page.spaceId,
+    );
+
+    const userSpaceRole = findHighestUserSpaceRole(userSpaceRoles);
+
+    if (!userSpaceRole) {
+      this.logger.warn(`User not authorized to access page: ${pageId}`);
+      throw new UnauthorizedException();
+    }
+
+    // Check page-level permissions
+    const { hasAnyRestriction, canAccess, canEdit } =
+      await this.pagePermissionRepo.canUserEditPage(user.id, page.id);
+
+    if (hasAnyRestriction) {
+      if (!canAccess) {
+        this.logger.warn(
+          `User ${user.id} denied page-level access to page: ${pageId}`,
+        );
+        throw new UnauthorizedException();
+      }
+
+      if (!canEdit) {
+        data.connectionConfig.readOnly = true;
+        this.logger.debug(
+          `User ${user.id} granted readonly access to restricted page: ${pageId}`,
+        );
+      }
+    } else {
+      // No restrictions - use space-level permissions
+      if (userSpaceRole === SpaceRole.READER) {
+        data.connectionConfig.readOnly = true;
+        this.logger.debug(`User granted readonly access to page: ${pageId}`);
+      }
+    }
+
+    if (page.deletedAt) {
+      data.connectionConfig.readOnly = true;
+    }
+
+    this.logger.debug(`Authenticated user ${user.id} on page ${pageId}`);
+
+    return {
+      user,
+    };
+  }
+}

apps/server/src/collaboration/extensions/logger.extension.ts

@@ -0,0 +1,19 @@
+import {
+  Extension,
+  onDisconnectPayload,
+  onLoadDocumentPayload,
+} from '@hocuspocus/server';
+import { Injectable, Logger } from '@nestjs/common';
+
+@Injectable()
+export class LoggerExtension implements Extension {
+  private readonly logger = new Logger('Collab' + LoggerExtension.name);
+
+  async afterUnloadDocument(data: onLoadDocumentPayload) {
+    this.logger.debug('Unloaded ' + data.documentName + ' from memory');
+  }
+
+  async onDisconnect(data: onDisconnectPayload) {
+    this.logger.debug('User disconnected from ' + data.documentName);
+  }
+}

apps/server/src/collaboration/extensions/persistence.extension.ts

@@ -0,0 +1,296 @@
+import {
+  afterUnloadDocumentPayload,
+  Extension,
+  onChangePayload,
+  onLoadDocumentPayload,
+  onStoreDocumentPayload,
+} from '@hocuspocus/server';
+import * as Y from 'yjs';
+import { Injectable, Logger } from '@nestjs/common';
+import { TiptapTransformer } from '@hocuspocus/transformer';
+import { getPageId, jsonToText, tiptapExtensions } from '../collaboration.util';
+import { PageRepo } from '@docmost/db/repos/page/page.repo';
+import { InjectKysely } from 'nestjs-kysely';
+import { KyselyDB } from '@docmost/db/types/kysely.types';
+import { executeTx } from '@docmost/db/utils';
+import { InjectQueue } from '@nestjs/bullmq';
+import { QueueJob, QueueName } from '../../integrations/queue/constants';
+import { Queue } from 'bullmq';
+import {
+  extractMentions,
+  extractUserMentions,
+} from '../../common/helpers/prosemirror/utils';
+import { isDeepStrictEqual } from 'node:util';
+import {
+  IPageHistoryJob,
+  IPageMentionNotificationJob,
+} from '../../integrations/queue/constants/queue.interface';
+import { Page } from '@docmost/db/types/entity.types';
+import { CollabHistoryService } from '../services/collab-history.service';
+import {
+  HISTORY_FAST_INTERVAL,
+  HISTORY_FAST_THRESHOLD,
+  HISTORY_INTERVAL,
+} from '../constants';
+import { TransclusionService } from '../../core/page/transclusion/transclusion.service';
+
+@Injectable()
+export class PersistenceExtension implements Extension {
+  private readonly logger = new Logger(PersistenceExtension.name);
+  private contributors: Map<string, Set<string>> = new Map();
+
+  constructor(
+    private readonly pageRepo: PageRepo,
+    @InjectKysely() private readonly db: KyselyDB,
+    @InjectQueue(QueueName.AI_QUEUE) private aiQueue: Queue,
+    @InjectQueue(QueueName.HISTORY_QUEUE) private historyQueue: Queue,
+    @InjectQueue(QueueName.NOTIFICATION_QUEUE) private notificationQueue: Queue,
+    private readonly collabHistory: CollabHistoryService,
+    private readonly transclusionService: TransclusionService,
+  ) {}
+
+  async onLoadDocument(data: onLoadDocumentPayload) {
+    const { documentName, document } = data;
+    const pageId = getPageId(documentName);
+
+    if (!document.isEmpty('default')) {
+      return;
+    }
+
+    const page = await this.pageRepo.findById(pageId, {
+      includeContent: true,
+      includeYdoc: true,
+    });
+
+    if (!page) {
+      this.logger.warn('page not found');
+      return;
+    }
+
+    if (page.ydoc) {
+      this.logger.debug(`ydoc loaded from db: ${pageId}`);
+
+      const doc = new Y.Doc();
+      const dbState = new Uint8Array(page.ydoc);
+
+      Y.applyUpdate(doc, dbState);
+      return doc;
+    }
+
+    // if no ydoc state in db convert json in page.content to Ydoc.
+    if (page.content) {
+      this.logger.debug(`converting json to ydoc: ${pageId}`);
+
+      const ydoc = TiptapTransformer.toYdoc(
+        page.content,
+        'default',
+        tiptapExtensions,
+      );
+
+      Y.encodeStateAsUpdate(ydoc);
+      return ydoc;
+    }
+
+    this.logger.debug(`creating fresh ydoc: ${pageId}`);
+    return new Y.Doc();
+  }
+
+  async onStoreDocument(data: onStoreDocumentPayload) {
+    const { documentName, document, lastContext } = data;
+
+    const pageId = getPageId(documentName);
+
+    const tiptapJson = TiptapTransformer.fromYdoc(document, 'default');
+    const ydocState = Buffer.from(Y.encodeStateAsUpdate(document));
+
+    let textContent = null;
+
+    try {
+      textContent = jsonToText(tiptapJson);
+    } catch (err) {
+      this.logger.warn('jsonToText' + err?.['message']);
+    }
+
+    let page: Page = null;
+    const editingUserIds = this.consumeContributors(documentName);
+
+    try {
+      await executeTx(this.db, async (trx) => {
+        page = await this.pageRepo.findById(pageId, {
+          withLock: true,
+          includeContent: true,
+          trx,
+        });
+
+        if (!page) {
+          this.logger.error(`Page with id ${pageId} not found`);
+          return;
+        }
+
+        if (isDeepStrictEqual(tiptapJson, page.content)) {
+          page = null;
+          return;
+        }
+
+        let contributorIds = undefined;
+        try {
+          const existingContributors = page.contributorIds || [];
+          contributorIds = Array.from(
+            new Set([
+              ...existingContributors,
+              ...editingUserIds,
+              page.creatorId,
+            ]),
+          );
+        } catch (err) {
+          //this.logger.debug('Contributors error:' + err?.['message']);
+        }
+
+        await this.pageRepo.updatePage(
+          {
+            content: tiptapJson,
+            textContent: textContent,
+            ydoc: ydocState,
+            lastUpdatedById: lastContext.user.id,
+            contributorIds: contributorIds,
+          },
+          pageId,
+          trx,
+        );
+
+        this.logger.debug(`Page updated: ${pageId} - SlugId: ${page.slugId}`);
+      });
+    } catch (err) {
+      this.logger.error(`Failed to update page ${pageId}`, err);
+    }
+
+    if (page) {
+      document.broadcastStateless(
+        JSON.stringify({
+          type: 'page.updated',
+          updatedAt: new Date().toISOString(),
+          lastUpdatedById: lastContext?.user?.id,
+          lastUpdatedBy: lastContext?.user
+            ? {
+                id: lastContext.user?.id,
+                name: lastContext.user?.name,
+                avatarUrl: lastContext.user?.avatarUrl,
+              }
+            : undefined,
+        }),
+      );
+
+      await this.syncTransclusion(pageId, page.workspaceId, tiptapJson);
+    }
+
+    if (page) {
+      await this.collabHistory.addContributors(pageId, editingUserIds);
+
+      const mentions = extractMentions(tiptapJson);
+
+      const userMentions = extractUserMentions(mentions);
+      const oldMentions = page.content ? extractMentions(page.content) : [];
+      const oldMentionedUserIds = extractUserMentions(oldMentions).map(
+        (m) => m.entityId,
+      );
+
+      if (userMentions.length > 0) {
+        await this.notificationQueue.add(QueueJob.PAGE_MENTION_NOTIFICATION, {
+          userMentions: userMentions.map((m) => ({
+            userId: m.entityId,
+            mentionId: m.id,
+            creatorId: m.creatorId,
+          })),
+          oldMentionedUserIds,
+          pageId,
+          spaceId: page.spaceId,
+          workspaceId: page.workspaceId,
+        } as IPageMentionNotificationJob);
+      }
+
+      await this.aiQueue.add(QueueJob.PAGE_CONTENT_UPDATED, {
+        pageIds: [pageId],
+        workspaceId: page.workspaceId,
+      });
+
+      await this.enqueuePageHistory(page);
+    }
+  }
+
+  async onChange(data: onChangePayload) {
+    const documentName = data.documentName;
+    const userId = data.context?.user?.id;
+
+    if (!userId) return;
+
+    if (!this.contributors.has(documentName)) {
+      this.contributors.set(documentName, new Set());
+    }
+
+    this.contributors.get(documentName).add(userId);
+  }
+
+  async afterUnloadDocument(data: afterUnloadDocumentPayload) {
+    const documentName = data.documentName;
+    this.contributors.delete(documentName);
+  }
+
+  private consumeContributors(documentName: string): string[] {
+    const contributorSet = this.contributors.get(documentName);
+    if (!contributorSet) return [];
+    const userIds = [...contributorSet];
+    this.contributors.delete(documentName);
+    return userIds;
+  }
+
+  private async enqueuePageHistory(page: Page): Promise<void> {
+    const pageAge = Date.now() - new Date(page.createdAt).getTime();
+    const delay =
+      pageAge < HISTORY_FAST_THRESHOLD
+        ? HISTORY_FAST_INTERVAL
+        : HISTORY_INTERVAL;
+
+    await this.historyQueue.add(
+      QueueJob.PAGE_HISTORY,
+      { pageId: page.id } as IPageHistoryJob,
+      { jobId: page.id, delay },
+    );
+  }
+
+  /**
+   * Refresh `page_transclusions` and `page_transclusion_references` to match
+   * the page's current content. Runs outside the page-write transaction and
+   * isolates each call so a failure here cannot affect the page save itself.
+   * The diff is idempotent — the next save converges if a round drops anything.
+   */
+  private async syncTransclusion(
+    pageId: string,
+    workspaceId: string,
+    tiptapJson: unknown,
+  ): Promise<void> {
+    try {
+      await this.transclusionService.syncPageTransclusions(
+        pageId,
+        workspaceId,
+        tiptapJson,
+      );
+    } catch (err) {
+      this.logger.error(
+        { err, pageId },
+        'Failed to sync transclusions for page',
+      );
+    }
+    try {
+      await this.transclusionService.syncPageReferences(
+        pageId,
+        workspaceId,
+        tiptapJson,
+      );
+    } catch (err) {
+      this.logger.error(
+        { err, pageId },
+        'Failed to sync transclusion references for page',
+      );
+    }
+  }
+}

apps/server/src/collaboration/extensions/redis-sync/collab-proxy-socket.ts

@@ -0,0 +1,46 @@
+import type RedisClient from 'ioredis';
+import type { WebSocketLike } from '@hocuspocus/server';
+import type { Pack, RSAMessageClose, RSAMessageSend } from './redis-sync.types';
+
+// Stands in for the client WebSocket on the server that owns the document.
+// Outgoing traffic is relayed over redis to the origin server, which holds the real socket.
+export class CollabProxySocket implements WebSocketLike {
+  private readonly replyTo: string;
+  private readonly socketId: string;
+  private pub: RedisClient;
+  private readonly pack: Pack;
+  readyState = 1;
+  onClose?: (code?: number, reason?: string) => void;
+
+  constructor(pub: RedisClient, pack: Pack, replyTo: string, socketId: string) {
+    this.replyTo = replyTo;
+    this.socketId = socketId;
+    this.pub = pub;
+    this.pack = pack;
+  }
+
+  private publish(msg: RSAMessageClose | RSAMessageSend) {
+    this.pub.publish(this.replyTo, this.pack(msg));
+  }
+
+  // The origin server already closed the real socket; stop relaying without echoing a close back
+  markClosed() {
+    this.readyState = 3;
+  }
+
+  close(code?: number, reason?: string) {
+    if (this.readyState !== 1) return;
+    this.readyState = 3;
+    this.onClose?.(code, reason);
+  }
+
+  send(message: Uint8Array) {
+    if (this.readyState !== 1) return;
+    const msg: RSAMessageSend = {
+      type: 'send',
+      socketId: this.socketId,
+      message,
+    };
+    this.publish(msg);
+  }
+}

apps/server/src/collaboration/extensions/redis-sync/index.ts

@@ -0,0 +1,2 @@
+export * from './redis-sync.extension';
+export type { SerializedHTTPRequest } from './redis-sync.extension';

apps/server/src/collaboration/extensions/redis-sync/redis-sync.extension.ts

@@ -0,0 +1,428 @@
+// Source https://github.com/ueberdosis/hocuspocus/pull/1008 - MIT
+import {
+  Extension,
+  Hocuspocus,
+  IncomingMessage,
+  onConfigurePayload,
+  onLoadDocumentPayload,
+  afterUnloadDocumentPayload,
+  WebSocketLike,
+} from '@hocuspocus/server';
+import { ConnectionTimeout, Unauthorized } from '@hocuspocus/common';
+import RedisClient from 'ioredis';
+import { CollabProxySocket } from './collab-proxy-socket';
+import {
+  Configuration,
+  CustomEvents,
+  Pack,
+  RSAMessage,
+  RSAMessageClose,
+  RSAMessageCloseProxy,
+  RSAMessageCustomEventComplete,
+  RSAMessageCustomEventStart,
+  RSAMessageProxy,
+  RSAMessageUnload,
+  SerializedHTTPRequest,
+  Unpack,
+  OriginConnection,
+  ProxyConnection,
+  toWebRequest,
+} from './redis-sync.types';
+
+export type { Pack, SerializedHTTPRequest } from './redis-sync.types';
+
+type ServerId = string;
+type DocumentName = string;
+type SocketId = string;
+
+export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
+  priority = 1000;
+  private readonly pub: RedisClient;
+  private sub: RedisClient;
+  private readonly pack: Pack;
+  private readonly unpack: Unpack;
+  private originConnections: Record<SocketId, OriginConnection> = {};
+  private locks: Record<DocumentName, NodeJS.Timeout> = {};
+  private lockPromises: Record<DocumentName, Promise<ServerId | null>> = {};
+  private proxyConnections: Record<SocketId, ProxyConnection> = {};
+  private readonly prefix: string;
+  private readonly lockPrefix: string;
+  private readonly msgChannel: string;
+  private readonly serverId: ServerId;
+  private readonly customEventTTL: number;
+  private readonly lockTTL: number;
+  private instance!: Hocuspocus;
+  private readonly customEvents: TCE;
+  private replyIdCounter: number = 0;
+  // @ts-ignore
+  private pendingReplies: Record<number, PromiseWithResolvers<any>['resolve']> =
+    {};
+  private deriveContext: (
+    serializedHTTPRequest: SerializedHTTPRequest,
+  ) => Record<string, any>;
+
+  constructor(configuration: Configuration<TCE>) {
+    const {
+      redis,
+      pack,
+      unpack,
+      serverId,
+      lockTTL,
+      prefix,
+      customEvents,
+      customEventTTL,
+      deriveContext,
+    } = configuration;
+    this.pub = redis.duplicate();
+    this.sub = redis.duplicate();
+    this.pack = pack;
+    this.unpack = unpack;
+    this.serverId = serverId;
+    this.lockTTL = lockTTL ?? 10_000;
+    this.customEventTTL = customEventTTL ?? 30_000;
+    this.prefix = prefix ?? 'collab';
+    this.lockPrefix = `${this.prefix}Lock`;
+    this.msgChannel = `${this.prefix}Msg`;
+    this.customEvents = (customEvents as any) ?? ({} as any as CustomEvents);
+    this.deriveContext = deriveContext ?? (() => ({}));
+    this.sub.subscribe(this.msgChannel, `${this.msgChannel}:${this.serverId}`);
+    this.sub.on('messageBuffer', this.handleRedisMessage);
+    this.pub.on('error', () => {});
+    this.sub.on('error', () => {});
+  }
+  private getKey(documentName: string) {
+    return `${this.lockPrefix}:${documentName}`;
+  }
+
+  private closeProxy(socketId: string) {
+    const entry = this.proxyConnections[socketId];
+    if (entry) {
+      delete this.proxyConnections[socketId];
+      const { socket, clientConnection } = entry;
+      // The origin socket is already gone; don't echo a close message back
+      socket.markClosed();
+      clientConnection.handleClose({
+        code: 1000,
+        reason: 'provider_initiated',
+      });
+    }
+  }
+
+  private handleProxyMessage(
+    msg: Pick<RSAMessageProxy, 'replyTo' | 'message' | 'serializedHTTPRequest'>,
+  ) {
+    const { replyTo, message, serializedHTTPRequest } = msg;
+    const { headers } = serializedHTTPRequest;
+    const socketId = headers['sec-websocket-key'];
+    let entry = this.proxyConnections[socketId];
+    if (!entry) {
+      const socket = new CollabProxySocket(
+        this.pub,
+        this.pack,
+        replyTo,
+        socketId,
+      );
+      // A proxy connection with no live documents (client left the page, auth
+      // failed, or the origin server crashed) is reaped by hocuspocus' message
+      // timeout. Dispose it silently in that case: relaying the timeout close
+      // to the origin would kill the client's real socket, which may be busy
+      // serving other documents. Genuine protocol closes are still relayed.
+      socket.onClose = (code, reason) => {
+        delete this.proxyConnections[socketId];
+        if (code !== ConnectionTimeout.code) {
+          const msg: RSAMessageClose = {
+            type: 'close',
+            code,
+            reason,
+            socketId,
+          };
+          this.pub.publish(replyTo, this.pack(msg));
+        }
+      };
+      const clientConnection = this.instance.handleConnection(
+        socket,
+        toWebRequest(serializedHTTPRequest),
+        this.deriveContext(serializedHTTPRequest),
+      );
+      entry = { clientConnection, socket };
+      this.proxyConnections[socketId] = entry;
+    }
+    entry.clientConnection.handleMessage(message);
+  }
+
+  private getLock(documentName: string) {
+    return this.pub.get(this.getKey(documentName));
+  }
+
+  private getOrClaimLock(documentName: string) {
+    const lockPromise = this.pub.set(
+      this.getKey(documentName),
+      this.serverId,
+      'PX',
+      this.lockTTL,
+      'NX',
+      'GET',
+    );
+    this.lockPromises[documentName] = lockPromise;
+    // Briefly cache the serverId that claimed the doc to reduce load on redis
+    // When the claimant unloads the doc, it will send an unload message to immediately clear this
+    // a lockTTL / 2 guarantees stale reads < lockTTL upon server crash
+    setTimeout(() => {
+      delete this.lockPromises[documentName];
+    }, this.lockTTL / 2);
+    return lockPromise;
+  }
+
+  private getOrClaimLockThrottled(documentName: string) {
+    const existingWorkerIdPromise = this.lockPromises[documentName];
+    if (existingWorkerIdPromise) return existingWorkerIdPromise;
+    return this.getOrClaimLock(documentName);
+  }
+
+  private handleRedisMessage = async (
+    _channel: Buffer,
+    packedMessage: Buffer,
+  ) => {
+    const msg = this.unpack(packedMessage) as RSAMessage;
+    const { type } = msg;
+    if (type === 'proxy') {
+      this.handleProxyMessage(msg);
+      return;
+    }
+    if (type === 'closeProxy') {
+      this.closeProxy(msg.socketId);
+      return;
+    }
+    if (type === 'unload') {
+      delete this.lockPromises[msg.documentName];
+      return;
+    }
+    if (type === 'customEventStart') {
+      const { documentName, eventName, payload, replyTo, replyId } = msg;
+      const res = await this.handleEventLocally(
+        eventName as Extract<keyof TCE, string>,
+        documentName,
+        payload,
+      );
+      const reply: RSAMessageCustomEventComplete = {
+        type: 'customEventComplete',
+        replyId,
+        payload: res,
+      };
+      this.pub.publish(`${replyTo}`, this.pack(reply));
+      return;
+    }
+    if (type === 'customEventComplete') {
+      const { replyId, payload } = msg;
+      const resolveFn = this.pendingReplies[replyId];
+      if (!resolveFn) return;
+      delete this.pendingReplies[replyId];
+      resolveFn(payload);
+      return;
+    }
+    const { socketId } = msg;
+    const entry = this.originConnections[socketId];
+    if (!entry) {
+      // origin socket already cleaned up
+      return;
+    }
+    const { socket } = entry;
+    if (type === 'close') {
+      socket.close(msg.code, msg.reason);
+    } else if (type === 'send') {
+      socket.send(msg.message);
+    }
+  };
+
+  async maintainLock(documentName: string) {
+    this.locks[documentName] = setInterval(() => {
+      this.pub.set(
+        this.getKey(documentName),
+        this.serverId,
+        'PX',
+        this.lockTTL,
+      );
+    }, this.lockTTL / 2);
+  }
+
+  async releaseLock(documentName: string) {
+    clearInterval(this.locks[documentName]);
+    delete this.locks[documentName];
+    return this.pub.del(this.getKey(documentName));
+  }
+
+  private async handleEventLocally<TName extends Extract<keyof TCE, string>>(
+    eventName: TName,
+    documentName: string,
+    payload: any,
+  ) {
+    const handler = this.customEvents[eventName];
+    if (!handler) throw new Error(`Invalid eventName: ${eventName}`);
+    const result = await handler(documentName, payload);
+    return result as Promise<ReturnType<TCE[TName]>>;
+  }
+
+  async handleEvent<TName extends Extract<keyof TCE, string>>(
+    eventName: TName,
+    documentName: string,
+    payload: any,
+    // if true, don't claim the lock. Useful for targeting pages that are currently open
+    onlyIfOpen = false,
+  ) {
+    const isDocLoadedOnInstance = this.instance.documents.has(documentName);
+
+    if (isDocLoadedOnInstance) {
+      return this.handleEventLocally(eventName, documentName, payload);
+    }
+
+    const proxyTo = await (onlyIfOpen
+      ? this.getLock(documentName)
+      : this.getOrClaimLockThrottled(documentName));
+
+    if (!proxyTo && onlyIfOpen) {
+      return;
+    }
+
+    if (proxyTo && proxyTo !== this.serverId) {
+      ++this.replyIdCounter; // bug in biome thinks this.replyIdCounter is not used if written on the line below
+      const replyId = this.replyIdCounter;
+      // another server owns the doc
+      const proxyMessage: RSAMessageCustomEventStart = {
+        eventName,
+        documentName,
+        payload,
+        replyTo: `${this.msgChannel}:${this.serverId}`,
+        replyId,
+        type: 'customEventStart',
+      };
+      const msg = this.pack(proxyMessage);
+      this.pub.publish(`${this.msgChannel}:${proxyTo}`, msg);
+      // @ts-ignore
+      const { promise, resolve, reject } = Promise.withResolvers();
+      this.pendingReplies[replyId] = resolve;
+      setTimeout(() => {
+        delete this.pendingReplies[replyId];
+        reject(new Error('TIMEOUT'));
+      }, this.customEventTTL);
+      return promise as Promise<ReturnType<TCE[TName]>>;
+    }
+    // This server owns the document, but hocuspocus hasn't loaded it yet
+    return this.handleEventLocally(eventName, documentName, payload);
+  }
+
+  async lockDocument(documentName: string) {
+    const proxyTo = await this.getOrClaimLockThrottled(documentName);
+    if (proxyTo && proxyTo !== this.serverId) {
+      throw new Error(`Could not lock document: ${documentName}`);
+    }
+    this.maintainLock(documentName);
+    return () => this.releaseLock(documentName);
+  }
+
+  /* WebSocket Server Hooks */
+  onSocketOpen(
+    ws: WebSocketLike,
+    serializedHTTPRequest: SerializedHTTPRequest,
+  ) {
+    const socketId = serializedHTTPRequest.headers['sec-websocket-key'];
+    const clientConnection = this.instance.handleConnection(
+      ws,
+      toWebRequest(serializedHTTPRequest),
+      this.deriveContext(serializedHTTPRequest),
+    );
+    this.originConnections[socketId] = { clientConnection, socket: ws };
+  }
+
+  async onSocketMessage(
+    serializedHTTPRequest: SerializedHTTPRequest,
+    detachableMsg: ArrayBuffer,
+  ) {
+    const socketId = serializedHTTPRequest.headers['sec-websocket-key'];
+    const entry = this.originConnections[socketId];
+    if (!entry) return;
+    const { clientConnection } = entry;
+
+    let message: Uint8Array;
+    let documentName: string;
+    try {
+      message = new Uint8Array(detachableMsg.slice());
+      const tmpMsg = new IncomingMessage(detachableMsg);
+      const documentNameAndSessionId = tmpMsg.readVarString();
+      // session-aware providers suffix the documentName with \0sessionId
+      const sepIdx = documentNameAndSessionId.indexOf('\0');
+      documentName =
+        sepIdx === -1
+          ? documentNameAndSessionId
+          : documentNameAndSessionId.slice(0, sepIdx);
+    } catch (error) {
+      entry.socket.close(Unauthorized.code, Unauthorized.reason);
+      return;
+    }
+    const isDocLoadedOnInstance = this.instance.documents.has(documentName);
+
+    if (isDocLoadedOnInstance) {
+      clientConnection.handleMessage(message);
+      return;
+    }
+
+    const proxyTo = await this.getOrClaimLockThrottled(documentName);
+    if (proxyTo && proxyTo !== this.serverId) {
+      // Proxied messages bypass handleMessage, so refresh the connection's
+      // liveness fields manually or hocuspocus' message timeout would reap the
+      // real socket every `timeout` ms. connectionEstablishedAt is the
+      // reference while unauthenticated (auth for remote docs is proxied too)
+      // and is private upstream.
+      clientConnection.lastMessageReceivedAt = Date.now();
+      (clientConnection as any).connectionEstablishedAt = Date.now();
+      // another server owns the doc
+      const proxyMessage: RSAMessageProxy = {
+        serializedHTTPRequest: serializedHTTPRequest,
+        replyTo: `${this.msgChannel}:${this.serverId}`,
+        message,
+        type: 'proxy',
+      };
+      const msg = this.pack(proxyMessage);
+      this.pub.publish(`${this.msgChannel}:${proxyTo}`, msg);
+      return;
+    }
+    // This server owns the document, but hocuspocus hasn't loaded it yet
+    clientConnection.handleMessage(message);
+  }
+
+  onSocketClose(socketId: string, code?: number, reason?: ArrayBuffer) {
+    const entry = this.originConnections[socketId];
+    if (!entry) return;
+    delete this.originConnections[socketId];
+    entry.clientConnection.handleClose({
+      code: code ?? 1000,
+      reason: reason ? Buffer.from(reason).toString() : '',
+    });
+    const msg: RSAMessageCloseProxy = { type: 'closeProxy', socketId };
+    this.pub.publish(this.msgChannel, this.pack(msg)).catch(() => {});
+  }
+
+  /* Hocuspocus hooks */
+  async onConfigure({ instance }: onConfigurePayload) {
+    this.instance = instance;
+  }
+
+  async onLoadDocument(data: onLoadDocumentPayload) {
+    const { documentName } = data;
+    // Refresh the lock TTL
+    this.maintainLock(documentName);
+  }
+
+  async afterUnloadDocument(data: afterUnloadDocumentPayload) {
+    const { documentName } = data;
+    this.releaseLock(documentName);
+    // Broadcast to cluster to immediately remove the cached redis value
+    const msg: RSAMessageUnload = { type: 'unload', documentName };
+    this.pub.publish(this.msgChannel, this.pack(msg));
+  }
+
+  async onDestroy() {
+    this.pendingReplies = {};
+    this.pub.disconnect(false);
+    this.sub.disconnect(false);
+  }
+}

apps/server/src/collaboration/extensions/redis-sync/redis-sync.types.ts

@@ -0,0 +1,138 @@
+import { IncomingHttpHeaders } from 'node:http2';
+import RedisClient from 'ioredis';
+import { CollabProxySocket } from './collab-proxy-socket';
+import { type Hocuspocus, type WebSocketLike } from '@hocuspocus/server';
+
+export type SecondParam<T> = T extends (
+  arg1: any,
+  arg2: infer A,
+  ...args: any[]
+) => any
+  ? A
+  : never;
+
+export type SerializedHTTPRequest = {
+  method: string;
+  url: string;
+  headers: IncomingHttpHeaders;
+  socket: { remoteAddress: string };
+};
+
+export type RSAMessageProxy = {
+  type: 'proxy';
+  replyTo: string;
+  message: Uint8Array<ArrayBufferLike>;
+  serializedHTTPRequest: SerializedHTTPRequest;
+};
+
+export type RSAMessageCloseProxy = {
+  type: 'closeProxy';
+  socketId: string;
+};
+
+export type RSAMessageUnload = {
+  type: 'unload';
+  documentName: string;
+};
+
+export type RSAMessageClose = {
+  type: 'close';
+  code?: number;
+  reason?: string;
+  socketId: string;
+};
+
+export type RSAMessageSend = {
+  type: 'send';
+  // @ts-ignore
+  message: Uint8Array<ArrayBufferLike>;
+  socketId: string;
+};
+
+export type RSAMessageCustomEventStart<TName = string, TPayload = any> = {
+  type: 'customEventStart';
+  documentName: string;
+  eventName: TName;
+  payload: TPayload;
+  replyTo: string;
+  replyId: number;
+};
+
+export type RSAMessageCustomEventComplete = {
+  type: 'customEventComplete';
+  replyId: number;
+  payload: any;
+};
+
+export type RSAMessage =
+  | RSAMessageProxy
+  | RSAMessageCloseProxy
+  | RSAMessageUnload
+  | RSAMessageClose
+  | RSAMessageSend
+  | RSAMessageCustomEventStart
+  | RSAMessageCustomEventComplete;
+
+// @ts-ignore
+export type Pack = (msg: RSAMessage) => string | Buffer<ArrayBufferLike>;
+
+export type Unpack = (
+  // @ts-ignore
+  packedMessage: Uint8Array | Buffer<ArrayBufferLike>,
+) => RSAMessage;
+
+type ServerId = string;
+type DocumentName = string;
+type CustomEventName = string;
+
+export type CustomEvents = Record<
+  CustomEventName,
+  (documentName: string, payload: any) => Promise<any>
+>;
+
+// Not exported by @hocuspocus/server
+export type ClientConnection = ReturnType<Hocuspocus['handleConnection']>;
+export type OriginConnection = {
+  clientConnection: ClientConnection;
+  socket: WebSocketLike;
+};
+export type ProxyConnection = {
+  clientConnection: ClientConnection;
+  socket: CollabProxySocket;
+};
+
+export interface Configuration<TCE> {
+  redis: RedisClient;
+  pack: Pack;
+  unpack: Unpack;
+  serverId: ServerId;
+  lockTTL?: number;
+  customEventTTL?: number;
+  prefix?: string;
+  customEvents?: TCE;
+  // Derive the hocuspocus context once per socket instead of re-deriving it in a
+  // per-document hook like onConnect/onAuthenticate. Runs on the origin server when
+  // the socket opens and on the doc owner when the first proxied message arrives.
+  deriveContext?: (
+    serializedHTTPRequest: SerializedHTTPRequest,
+  ) => Record<string, any>;
+}
+
+// Hocuspocus expects a web-standard Request, so rehydrate one from what crossed the wire
+export const toWebRequest = (serializedHTTPRequest: SerializedHTTPRequest) => {
+  const { method, url, headers } = serializedHTTPRequest;
+  const webHeaders = new Headers();
+  Object.entries(headers).forEach(([name, value]) => {
+    if (Array.isArray(value)) {
+      value.forEach((v) => {
+        webHeaders.append(name, v);
+      });
+    } else if (value !== undefined) {
+      webHeaders.set(name, value);
+    }
+  });
+  return new Request(new URL(url, 'http://localhost'), {
+    method,
+    headers: webHeaders,
+  });
+};

apps/server/src/collaboration/extensions/redis-sync/ws-socket-wrapper.ts

@@ -0,0 +1,35 @@
+import type WebSocket from 'ws';
+import type { WebSocketLike } from '@hocuspocus/server';
+
+/**
+ * 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;
+  readyState = 1;
+
+  constructor(ws: WebSocket) {
+    this.ws = ws;
+  }
+
+  close(code?: number, reason?: string) {
+    if (this.readyState !== 1) return;
+    this.readyState = 3;
+    try {
+      this.ws.close(code, reason);
+    } catch (e) {
+      // Socket already closed
+    }
+  }
+
+  send(message: Uint8Array) {
+    if (this.readyState !== 1) return;
+    try {
+      this.ws.send(message);
+    } catch (e) {
+      // Socket already closed
+    }
+  }
+}

apps/server/src/collaboration/processors/history.processor.ts

@@ -0,0 +1,142 @@
+import { Logger, OnModuleDestroy } from '@nestjs/common';
+import { OnWorkerEvent, Processor, WorkerHost } from '@nestjs/bullmq';
+import { InjectQueue } from '@nestjs/bullmq';
+import { Job, Queue } from 'bullmq';
+import { QueueJob, QueueName } from '../../integrations/queue/constants';
+import {
+  IPageBacklinkJob,
+  IPageHistoryJob,
+  IPageUpdateNotificationJob,
+} from '../../integrations/queue/constants/queue.interface';
+import {
+  extractMentions,
+  extractPageMentions,
+  extractInternalLinkSlugIds,
+} from '../../common/helpers/prosemirror/utils';
+import { PageHistoryRepo } from '@docmost/db/repos/page/page-history.repo';
+import { PageRepo } from '@docmost/db/repos/page/page.repo';
+import { isDeepStrictEqual } from 'node:util';
+import { CollabHistoryService } from '../services/collab-history.service';
+import { WatcherService } from '../../core/watcher/watcher.service';
+import { isEmptyParagraphDoc } from '../collaboration.util';
+
+@Processor(QueueName.HISTORY_QUEUE)
+export class HistoryProcessor extends WorkerHost implements OnModuleDestroy {
+  private readonly logger = new Logger(HistoryProcessor.name);
+
+  constructor(
+    private readonly pageHistoryRepo: PageHistoryRepo,
+    private readonly pageRepo: PageRepo,
+    private readonly collabHistory: CollabHistoryService,
+    private readonly watcherService: WatcherService,
+    @InjectQueue(QueueName.NOTIFICATION_QUEUE) private notificationQueue: Queue,
+    @InjectQueue(QueueName.GENERAL_QUEUE) private generalQueue: Queue,
+  ) {
+    super();
+  }
+
+  async process(job: Job<IPageHistoryJob, void>): Promise<void> {
+    if (job.name !== QueueJob.PAGE_HISTORY) return;
+
+    try {
+      const { pageId } = job.data;
+
+      const page = await this.pageRepo.findById(pageId, {
+        includeContent: true,
+      });
+
+      if (!page) {
+        this.logger.warn(`Page ${pageId} not found, skipping history`);
+        await this.collabHistory.clearContributors(pageId);
+        return;
+      }
+
+      const lastHistory = await this.pageHistoryRepo.findPageLastHistory(
+        pageId,
+        { includeContent: true },
+      );
+
+      if (!lastHistory && isEmptyParagraphDoc(page.content as any)) {
+        this.logger.debug(
+          `Skipping first history for page ${pageId}: empty content`,
+        );
+        await this.collabHistory.clearContributors(pageId);
+        return;
+      }
+
+      if (
+        !lastHistory ||
+        !isDeepStrictEqual(lastHistory.content, page.content)
+      ) {
+        const contributorIds = await this.collabHistory.popContributors(pageId);
+
+        try {
+          await this.watcherService.addPageWatchers(
+            contributorIds,
+            pageId,
+            page.spaceId,
+            page.workspaceId,
+          );
+
+          await this.pageHistoryRepo.saveHistory(page, { contributorIds });
+          this.logger.debug(`History created for page: ${pageId}`);
+        } catch (err) {
+          await this.collabHistory.addContributors(pageId, contributorIds);
+          throw err;
+        }
+
+        const mentions = extractMentions(page.content);
+        const pageMentions = extractPageMentions(mentions);
+        const internalLinkSlugIds = extractInternalLinkSlugIds(page.content);
+
+        await this.generalQueue
+          .add(QueueJob.PAGE_BACKLINKS, {
+            pageId,
+            workspaceId: page.workspaceId,
+            mentions: pageMentions,
+            internalLinkSlugIds,
+          } as IPageBacklinkJob)
+          .catch((err) => {
+            this.logger.error(
+              `Failed to queue backlinks for ${pageId}: ${err.message}`,
+            );
+          });
+
+        if (contributorIds.length > 0 && lastHistory?.content) {
+          await this.notificationQueue
+            .add(QueueJob.PAGE_UPDATED, {
+              pageId,
+              spaceId: page.spaceId,
+              workspaceId: page.workspaceId,
+              actorIds: contributorIds,
+            } as IPageUpdateNotificationJob)
+            .catch((err) => {
+              this.logger.error(
+                `Failed to queue page update notification for ${pageId}: ${err.message}`,
+              );
+            });
+        }
+      }
+    } catch (err) {
+      throw err;
+    }
+  }
+
+  @OnWorkerEvent('active')
+  onActive(job: Job) {
+    this.logger.debug(`Processing ${job.name} for page: ${job.data.pageId}`);
+  }
+
+  @OnWorkerEvent('failed')
+  onError(job: Job) {
+    this.logger.error(
+      `Failed ${job.name} for page: ${job.data.pageId}. Reason: ${job.failedReason}`,
+    );
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    if (this.worker) {
+      await this.worker.close();
+    }
+  }
+}

apps/server/src/collaboration/server/collab-app.module.ts

@@ -0,0 +1,64 @@
+import { Module } from '@nestjs/common';
+import { AppController } from '../../app.controller';
+import { AppService } from '../../app.service';
+import { EnvironmentModule } from '../../integrations/environment/environment.module';
+import { EnvironmentService } from '../../integrations/environment/environment.service';
+import { CollaborationModule } from '../collaboration.module';
+import { DatabaseModule } from '@docmost/db/database.module';
+import { QueueModule } from '../../integrations/queue/queue.module';
+import { EventEmitterModule } from '@nestjs/event-emitter';
+import { HealthModule } from '../../integrations/health/health.module';
+import { CollaborationController } from './collaboration.controller';
+import { LoggerModule } from '../../common/logger/logger.module';
+import { RedisModule } from '@nestjs-labs/nestjs-ioredis';
+import { RedisConfigService } from '../../integrations/redis/redis-config.service';
+import { CaslModule } from '../../core/casl/casl.module';
+import { CacheModule } from '@nestjs/cache-manager';
+import KeyvRedis, { defaultReconnectStrategy } from '@keyv/redis';
+import { parseRedisUrl } from '../../common/helpers';
+
+@Module({
+  imports: [
+    LoggerModule,
+    DatabaseModule,
+    EnvironmentModule,
+    CaslModule,
+    CollaborationModule,
+    QueueModule,
+    HealthModule,
+    EventEmitterModule.forRoot(),
+    RedisModule.forRootAsync({
+      useClass: RedisConfigService,
+    }),
+    CacheModule.registerAsync({
+      isGlobal: true,
+      useFactory: async (environmentService: EnvironmentService) => {
+        const redisUrl = environmentService.getRedisUrl();
+        const { family, tls } = parseRedisUrl(redisUrl);
+
+        return {
+          ttl: 5 * 1000,
+          stores: [
+            new KeyvRedis({
+              url: redisUrl,
+              socket: {
+                family,
+                reconnectStrategy: defaultReconnectStrategy,
+                ...tls,
+              },
+            }),
+          ],
+        };
+      },
+      inject: [EnvironmentService],
+    }),
+  ],
+  controllers: [
+    AppController,
+    ...(process.env.COLLAB_SHOW_STATS?.toLowerCase() === 'true'
+      ? [CollaborationController]
+      : []),
+  ],
+  providers: [AppService],
+})
+export class CollabAppModule {}

apps/server/src/collaboration/server/collab-main.ts

@@ -0,0 +1,47 @@
+import { NestFactory, Reflector } from '@nestjs/core';
+import { CollabAppModule } from './collab-app.module';
+import {
+  FastifyAdapter,
+  NestFastifyApplication,
+} from '@nestjs/platform-fastify';
+import { TransformHttpResponseInterceptor } from '../../common/interceptors/http-response.interceptor';
+import { Logger } from '@nestjs/common';
+import { Logger as PinoLogger } from 'nestjs-pino';
+import { InternalLogFilter } from '../../common/logger/internal-log-filter';
+
+async function bootstrap() {
+  const app = await NestFactory.create<NestFastifyApplication>(
+    CollabAppModule,
+    new FastifyAdapter({
+      routerOptions: {
+        maxParamLength: 1000,
+        ignoreTrailingSlash: true,
+        ignoreDuplicateSlashes: true,
+      },
+    }),
+    {
+      logger: new InternalLogFilter(),
+      bufferLogs: false,
+    },
+  );
+
+  app.useLogger(app.get(PinoLogger));
+
+  app.setGlobalPrefix('api', { exclude: ['/'] });
+
+  app.enableCors();
+
+  const reflector = app.get(Reflector);
+  app.useGlobalInterceptors(new TransformHttpResponseInterceptor(reflector));
+  app.enableShutdownHooks();
+
+  const logger = new Logger('CollabServer');
+
+  const port = process.env.COLLAB_PORT || 3001;
+  const host = process.env.HOST || '0.0.0.0';
+  await app.listen(port, host, () => {
+    logger.log(`Listening on http://127.0.0.1:${port}`);
+  });
+}
+
+bootstrap();

apps/server/src/collaboration/server/collaboration.controller.ts

@@ -0,0 +1,15 @@
+import { Controller, Get } from '@nestjs/common';
+import { CollaborationGateway } from '../collaboration.gateway';
+
+@Controller('collab')
+export class CollaborationController {
+  constructor(private readonly collaborationGateway: CollaborationGateway) {}
+
+  @Get('stats')
+  async getStats() {
+    return {
+      connections: this.collaborationGateway.getConnectionCount(),
+      documents: this.collaborationGateway.getDocumentCount(),
+    };
+  }
+}

apps/server/src/collaboration/services/collab-history.service.ts

@@ -0,0 +1,30 @@
+import { Injectable } from '@nestjs/common';
+import { RedisService } from '@nestjs-labs/nestjs-ioredis';
+import type { Redis } from 'ioredis';
+
+const REDIS_KEY_PREFIX = 'history:contributors:';
+
+@Injectable()
+export class CollabHistoryService {
+  private readonly redis: Redis;
+
+  constructor(private readonly redisService: RedisService) {
+    this.redis = this.redisService.getOrThrow();
+  }
+
+  async addContributors(pageId: string, userIds: string[]): Promise<void> {
+    if (userIds.length === 0) return;
+    await this.redis.sadd(REDIS_KEY_PREFIX + pageId, ...userIds);
+  }
+
+  async popContributors(pageId: string): Promise<string[]> {
+    const key = REDIS_KEY_PREFIX + pageId;
+    const count = await this.redis.scard(key);
+    if (count === 0) return [];
+    return await this.redis.spop(key, count);
+  }
+
+  async clearContributors(pageId: string): Promise<void> {
+    await this.redis.del(REDIS_KEY_PREFIX + pageId);
+  }
+}

apps/server/src/collaboration/yjs.util.ts

@@ -0,0 +1,177 @@
+import {
+  initProseMirrorDoc,
+  relativePositionToAbsolutePosition,
+} from '@tiptap/y-tiptap';
+import * as Y from 'yjs';
+import { Document } from '@hocuspocus/server';
+import { getSchema } from '@tiptap/core';
+import { tiptapExtensions } from './collaboration.util';
+
+export type YjsSelection = {
+  anchor: any;
+  head: any;
+};
+
+export function setYjsMark(
+  doc: Document,
+  fragment: Y.XmlFragment,
+  yjsSelection: YjsSelection,
+  markName: string,
+  markAttributes: Record<string, any>,
+) {
+  const schema = getSchema(tiptapExtensions);
+  const { mapping } = initProseMirrorDoc(fragment, schema);
+
+  // Convert JSON positions to Y.js RelativePosition objects
+  const anchorRelPos = Y.createRelativePositionFromJSON(yjsSelection.anchor);
+  const headRelPos = Y.createRelativePositionFromJSON(yjsSelection.head);
+
+  const anchor = relativePositionToAbsolutePosition(
+    doc,
+    fragment,
+    anchorRelPos,
+    mapping,
+  );
+  const head = relativePositionToAbsolutePosition(
+    doc,
+    fragment,
+    headRelPos,
+    mapping,
+  );
+
+  if (anchor === null || head === null) {
+    throw new Error(
+      'Could not resolve Y.js relative positions to absolute positions',
+    );
+  }
+
+  const from = Math.min(anchor, head);
+  const to = Math.max(anchor, head);
+
+  // Apply mark directly to Y.js XmlText nodes
+  // This bypasses updateYFragment which has compatibility issues
+  applyMarkToYFragment(fragment, from, to, markName, markAttributes);
+}
+
+function applyMarkToYFragment(
+  fragment: Y.XmlFragment,
+  from: number,
+  to: number,
+  markName: string,
+  markAttributes: Record<string, any>,
+) {
+  let pos = 0;
+
+  const processItem = (item: any, parentNodeName?: string): boolean => {
+    if (pos >= to) return false;
+
+    if (item instanceof Y.XmlText) {
+      const textLength = item.length;
+      const itemEnd = pos + textLength;
+
+      if (itemEnd > from && pos < to && parentNodeName !== 'codeBlock') {
+        const formatFrom = Math.max(0, from - pos);
+        const formatTo = Math.min(textLength, to - pos);
+        const formatLength = formatTo - formatFrom;
+
+        if (formatLength > 0) {
+          item.format(formatFrom, formatLength, { [markName]: markAttributes });
+        }
+      }
+      pos = itemEnd;
+    } else if (item instanceof Y.XmlElement) {
+      pos++; // Opening tag
+      for (let i = 0; i < item.length; i++) {
+        if (!processItem(item.get(i), item.nodeName)) return false;
+      }
+      pos++; // Closing tag
+    }
+    return true;
+  };
+
+  for (let i = 0; i < fragment.length; i++) {
+    if (!processItem(fragment.get(i))) break;
+  }
+}
+
+/**
+ * Removes a mark from all text in the fragment that has the specified attribute value.
+ * Useful for deleting comments by commentId.
+ */
+export function removeYjsMarkByAttribute(
+  fragment: Y.XmlFragment,
+  markName: string,
+  attributeName: string,
+  attributeValue: string,
+) {
+  const processItem = (item: any) => {
+    if (item instanceof Y.XmlText) {
+      // Get all formatting deltas to find ranges with this mark
+      const deltas = item.toDelta();
+      let offset = 0;
+
+      for (const delta of deltas) {
+        const length = delta.insert?.length ?? 0;
+        const attributes = delta.attributes ?? {};
+        const markAttr = attributes[markName];
+
+        if (markAttr && markAttr[attributeName] === attributeValue) {
+          // Remove the mark by setting it to null
+          item.format(offset, length, { [markName]: null });
+        }
+        offset += length;
+      }
+    } else if (item instanceof Y.XmlElement) {
+      for (let i = 0; i < item.length; i++) {
+        processItem(item.get(i));
+      }
+    }
+  };
+
+  for (let i = 0; i < fragment.length; i++) {
+    processItem(fragment.get(i));
+  }
+}
+
+/**
+ * Updates a mark's attributes for all text that has the specified attribute value.
+ * Useful for resolving/unresolving comments by commentId.
+ */
+export function updateYjsMarkAttribute(
+  fragment: Y.XmlFragment,
+  markName: string,
+  findByAttribute: { name: string; value: string },
+  newAttributes: Record<string, any>,
+) {
+  const processItem = (item: any) => {
+    if (item instanceof Y.XmlText) {
+      const deltas = item.toDelta();
+      let offset = 0;
+
+      for (const delta of deltas) {
+        const length = delta.insert?.length ?? 0;
+        const attributes = delta.attributes ?? {};
+        const markAttr = attributes[markName];
+
+        if (
+          markAttr &&
+          markAttr[findByAttribute.name] === findByAttribute.value
+        ) {
+          // Update the mark with new attributes (merge with existing)
+          item.format(offset, length, {
+            [markName]: { ...markAttr, ...newAttributes },
+          });
+        }
+        offset += length;
+      }
+    } else if (item instanceof Y.XmlElement) {
+      for (let i = 0; i < item.length; i++) {
+        processItem(item.get(i));
+      }
+    }
+  };
+
+  for (let i = 0; i < fragment.length; i++) {
+    processItem(fragment.get(i));
+  }
+}