withReadOnly

Author: AntoLC

src/frontend/apps/impress/src/features/docs/doc-management/hooks/useCollaboration.tsx

@@ -6,7 +6,11 @@ import { useBroadcastStore } from '@/stores';
 import { useProviderStore } from '../stores/useProviderStore';
 import { Base64 } from '../types';
 
-export const useCollaboration = (room?: string, initialContent?: Base64) => {
+export const useCollaboration = (
+  room?: string,
+  initialContent?: Base64,
+  canEdit?: boolean,
+) => {
   const collaborationUrl = useCollaborationUrl(room);
   const { setBroadcastProvider } = useBroadcastStore();
   const { provider, createProvider, destroyProvider } = useProviderStore();
@@ -15,11 +19,16 @@ export const useCollaboration = (room?: string, initialContent?: Base64) => {
    * Initialize the provider
    */
   useEffect(() => {
-    if (!room || !collaborationUrl || provider) {
+    if (!room || !collaborationUrl || provider || canEdit === undefined) {
       return;
     }
 
-    const newProvider = createProvider(collaborationUrl, room, initialContent);
+    const newProvider = createProvider(
+      collaborationUrl,
+      room,
+      canEdit,
+      initialContent,
+    );
     setBroadcastProvider(newProvider);
   }, [
     provider,
@@ -28,11 +37,13 @@ export const useCollaboration = (room?: string, initialContent?: Base64) => {
     initialContent,
     createProvider,
     setBroadcastProvider,
+    canEdit,
   ]);
 
   useEffect(() => {
     return () => {
       if (room) {
+        console.log('Destroy Provider');
         destroyProvider();
       }
     };

src/frontend/apps/impress/src/features/docs/doc-management/libs/CollaborationProvider.ts

@@ -27,6 +27,10 @@ export const isHocuspocusProviderConfigurationUrl = (
   return 'url' in data;
 };
 
+type CollaborationProviderConfiguration = HocuspocusProviderConfiguration & {
+  canEdit: boolean;
+};
+
 export class CollaborationProvider extends HocuspocusProvider {
   private websocketFailureCount = 0;
   private websocketMaxFailureCount = 2;
@@ -37,7 +41,7 @@ export class CollaborationProvider extends HocuspocusProvider {
   // Server-Sent Events
   private sse: EventSource | null = null;
 
-  public constructor(configuration: HocuspocusProviderConfiguration) {
+  public constructor(configuration: CollaborationProviderConfiguration) {
     const withWS = isFirefox();
 
     let url = '';
@@ -50,7 +54,10 @@ export class CollaborationProvider extends HocuspocusProvider {
 
     this.url = url;
 
-    this.on('outgoingMessage', this.onPollOutgoingMessage.bind(this));
+    if (configuration.canEdit) {
+      this.on('outgoingMessage', this.onPollOutgoingMessage.bind(this));
+    }
+
     this.configuration.websocketProvider.on(
       'connect',
       this.onWebsocketConnect.bind(this),
@@ -92,7 +99,6 @@ export class CollaborationProvider extends HocuspocusProvider {
         this.isLongPollingStarted = true;
         void this.pollSync();
         this.initCollaborationSSE();
-        //void this.longPollAwareness();
       }
     }
   }
@@ -116,14 +122,18 @@ export class CollaborationProvider extends HocuspocusProvider {
 
     //console.log('outgoingMessage', message.description);
 
-    const { updated } = await pollOutgoingMessageRequest({
-      pollUrl: this.toPollUrl('message'),
-      message64: Buffer.from(message.toUint8Array()).toString('base64'),
-    });
+    try {
+      const { updated } = await pollOutgoingMessageRequest({
+        pollUrl: this.toPollUrl('message'),
+        message64: Buffer.from(message.toUint8Array()).toString('base64'),
+      });
 
-    if (!updated) {
-      console.error('Message not updated');
-      await this.pollSync();
+      if (!updated) {
+        console.error('Message not updated');
+        await this.pollSync();
+      }
+    } catch (error: unknown) {
+      console.error('Polling message failed:', error);
     }
   }
 
@@ -134,52 +144,60 @@ export class CollaborationProvider extends HocuspocusProvider {
 
     console.log('initCollaborationSSE');
 
-    const eventSource = new EventSource(this.toPollUrl('message'), {
+    this.sse = new EventSource(this.toPollUrl('message'), {
       withCredentials: true,
     });
 
+    //eventSource.close();
+
     // 1. onmessage handles messages sent with `data:` lines
-    eventSource.onmessage = async (event) => {
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-argument
+    this.sse.onmessage = (event) => {
       const { updatedDoc64, stateFingerprint, awareness64 } = JSON.parse(
+        // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
         event.data,
       ) as {
         updatedDoc64: string;
         stateFingerprint: string;
         awareness64: string;
       };
       console.log('Received SSE event:', event.data);
+      console.log('CLineId:', this.document.clientID);
+
+      const localStateFingerprint = this.getStateFingerprint(this.document);
+      console.log('EQUAL BEF', localStateFingerprint === stateFingerprint);
 
       if (awareness64) {
         const awareness = Buffer.from(awareness64, 'base64');
 
         this.onMessage({
           data: awareness,
         } as MessageEvent);
+
+        console.log('EQUAL AWA', localStateFingerprint === stateFingerprint);
+        // if (localStateFingerprint !== stateFingerprint) {
+        //   await this.pollSync();
+        // }
       }
 
       if (updatedDoc64) {
         const uint8Array = Buffer.from(updatedDoc64, 'base64');
         Y.applyUpdate(this.document, uint8Array);
 
-        const localStateFingerprint = this.getStateFingerprint(this.document);
-        console.log('stateFingerprint', stateFingerprint);
-        console.log('localStateFingerprint', localStateFingerprint);
         console.log('EQUAL', localStateFingerprint === stateFingerprint);
 
-        if (localStateFingerprint !== stateFingerprint) {
-          await this.pollSync();
-        }
+        // if (localStateFingerprint !== stateFingerprint) {
+        //   await this.pollSync();
+        // }
       }
     };
 
     // 2. onopen is triggered when the connection is first established
-    eventSource.onopen = () => {
+    this.sse.onopen = () => {
       console.log('SSE connection opened.');
     };
 
     // 3. onerror is triggered if there's a connection issue
-    eventSource.onerror = (err) => {
+    this.sse.onerror = (err) => {
       console.error('SSE error:', err);
       // Depending on the error, the browser may or may not automatically reconnect
     };
@@ -190,9 +208,9 @@ export class CollaborationProvider extends HocuspocusProvider {
   public onMessage(event: MessageEvent) {
     super.onMessage(event);
 
-    console.log('onMessage', event);
-    console.log('isSynced', this.isSynced);
-    console.log('unsyncedChanges', this.unsyncedChanges);
+    // console.log('onMessage', event);
+    // console.log('isSynced', this.isSynced);
+    // console.log('unsyncedChanges', this.unsyncedChanges);
 
     // if (this.hasUnsyncedChanges) {
     //   this.unsyncedChanges = 0;
@@ -216,7 +234,6 @@ export class CollaborationProvider extends HocuspocusProvider {
       if (syncDoc64) {
         const uint8Array = Buffer.from(syncDoc64, 'base64');
         Y.applyUpdate(this.document, uint8Array);
-        this.synced = true;
       }
     } catch (error) {
       console.error('Polling sync failed:', error);

src/frontend/apps/impress/src/features/docs/doc-management/stores/useProviderStore.tsx

@@ -10,6 +10,7 @@ export interface UseCollaborationStore {
   createProvider: (
     providerUrl: string,
     storeId: string,
+    canEdit: boolean,
     initialDoc?: Base64,
   ) => HocuspocusProvider;
   destroyProvider: () => void;
@@ -22,7 +23,7 @@ const defaultValues = {
 
 export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
   ...defaultValues,
-  createProvider: (wsUrl, storeId, initialDoc) => {
+  createProvider: (wsUrl, storeId, canEdit, initialDoc) => {
     const doc = new Y.Doc({
       guid: storeId,
     });
@@ -35,6 +36,7 @@ export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
       url: wsUrl,
       name: storeId,
       document: doc,
+      canEdit,
     });
 
     set({

src/frontend/apps/impress/src/pages/docs/[id]/index.tsx

@@ -64,7 +64,7 @@ const DocPage = ({ id }: DocProps) => {
   const { addTask } = useBroadcastStore();
   const queryClient = useQueryClient();
   const { replace } = useRouter();
-  useCollaboration(doc?.id, doc?.content);
+  useCollaboration(doc?.id, doc?.content, doc?.abilities.partial_update);
 
   useEffect(() => {
     if (doc?.title) {

src/frontend/servers/y-provider/src/libs/PollSync.ts

@@ -3,12 +3,17 @@ import crypto from 'crypto';
 
 import {
   AwarenessUpdate,
+  Debugger,
   Document,
   Hocuspocus,
+  IncomingMessage,
+  MessageReceiver,
+  MessageType,
   OutgoingMessage,
 } from '@hocuspocus/server';
 import { Request, Response } from 'express';
 import { v4 as uuid } from 'uuid';
+import { readSyncMessage, readUpdate } from 'y-protocols/sync.js';
 import * as Y from 'yjs';
 
 import { base64ToYDoc, logger, toBase64 } from '@/utils';
@@ -109,26 +114,47 @@ export class PollSync<T> {
    * Send messages to other clients
    */
   public sendMessages(message64: string) {
-    console.log('Polling Updated 1234');
     const hpDoc = this.getHpDocument();
+    const messageBuffer = Buffer.from(message64, 'base64');
+    const message = new IncomingMessage(messageBuffer);
+    const room = message.readVarString();
 
-    hpDoc.getConnections().forEach((connection) => {
-      connection.handleMessage(Buffer.from(message64, 'base64'));
-    });
+    logger('Polling sendMessages 1010', hpDoc.name);
+
+    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 {
+      hpDoc.getConnections().forEach((connection) => {
+        connection.handleMessage(messageBuffer);
+      });
+    }
+
+    // hpDoc.getConnections().forEach((connection) => {
+    //   connection.handleMessage(messageBuffer);
+    // });
 
     logger('Polling Updated YDoc', hpDoc.name);
   }
 
   public listenMessages(res: Response) {
     const hpDoc = this.getHpDocument();
+    hpDoc.addDirectConnection();
 
     const updateMessagesFn = (
       update: Uint8Array,
       _origin: string,
       updatedDoc: Y.Doc,
       _transaction: Y.Transaction,
     ) => {
-      console.log('Doc Update V2');
+      console.log('Doc Update V2', _transaction.doc.clientID);
 
       res.write(
         `data: ${JSON.stringify({
@@ -166,12 +192,14 @@ export class PollSync<T> {
         this.room,
       ).createAwarenessUpdateMessage(hpDoc.awareness, changedClients);
 
-      console.log('Awareness Update', awarenessMessage);
+      //console.log('Awareness Update', awarenessMessage);
 
+      console.log('Awareness Update V2');
       res.write(
         `data: ${JSON.stringify({
           time: new Date(),
           awareness64: toBase64(awarenessMessage.toUint8Array()),
+          stateFingerprint: this.getStateFingerprint(hpDoc),
         })}\n\n`,
       );
     };
@@ -188,6 +216,7 @@ export class PollSync<T> {
       hpDoc.off('update', updateMessagesFn);
       hpDoc.off('destroy', destroyFn);
       hpDoc.awareness.off('update', updateAwarenessFn);
+      hpDoc.removeDirectConnection();
     });
   }