wip

src/excalidraw-backend/server.ts

@@ -20,6 +20,7 @@ import {
   defaultSaveTimeout,
   DISCONNECT,
   DISCONNECTING,
+  ExcalidrawElement,
   IDLE_STATE,
   INIT_ROOM,
   JOIN_ROOM,
@@ -31,6 +32,7 @@ import {
   SERVER_SAVE_REQUEST,
   SERVER_SIDE_ROOM_DELETED,
   SERVER_VOLATILE_BROADCAST,
+  SocketEventData,
   SocketIoServer,
   SocketIoSocket,
 } from './types';
@@ -55,11 +57,19 @@ import { CREATE_ROOM, DELETE_ROOM } from './adapters/adapter.event.names';
 import { APP_ID } from '../app.id';
 import { arrayRandomElement, isAbortError } from '../util';
 import { ConfigType } from '../config';
+import { tryDecodeIncoming } from './utils/decode.incoming';
+import { ServerBroadcastPayload } from './types/events';
+import { reconcileElements } from './utils/reconcile';
+import { detectChanges } from './utils/detect.changes';
 
 type SaveMessageOpts = { timeout: number };
 type RoomTrackers = Map<string, AbortController>;
 type SocketTrackers = Map<string, AbortController>;
 
+type MasterSnapshot = {
+  elements: ExcalidrawElement[];
+} & Record<string, unknown>;
+
 @Injectable()
 export class Server {
   private readonly wsServer: SocketIoServer;
@@ -74,6 +84,8 @@ export class Server {
   private readonly saveConsecutiveFailedAttempts: number;
   private readonly collaboratorInactivityMs: number;
 
+  private snapshots: Map<string, MasterSnapshot> = new Map();
+
   constructor(
     @Inject(WINSTON_MODULE_NEST_PROVIDER) private logger: LoggerService,
     private readonly utilService: UtilService,
@@ -210,6 +222,9 @@ export class Server {
       });
 
       socket.on(JOIN_ROOM, async (roomID) => {
+        if (!this.snapshots.has(roomID)) {
+          this.snapshots.set(roomID, { elements: [] });
+        }
         // this logic could be provided by an entitlement (license) service
         await authorizeWithRoomAndJoinHandler(
           roomID,
@@ -228,6 +243,44 @@ export class Server {
               this.utilService.contentModified(socket.data.userInfo.id, roomId),
             );
             this.resetCollaboratorInactivityTrackerForSocket(socket);
+            let eventData: SocketEventData<ServerBroadcastPayload> | undefined;
+            try {
+              eventData = tryDecodeIncoming<ServerBroadcastPayload>(data);
+            } catch (e) {
+              this.logger.error({
+                message: e?.message ?? JSON.stringify(e),
+              });
+            }
+
+            if (!eventData) {
+              return;
+            }
+
+            const snapshot = this.snapshots.get(roomID);
+
+            if (!snapshot) {
+              return;
+            }
+
+            if (eventData.type === 'sync-check') {
+              console.log(
+                'sync-check',
+                JSON.stringify(
+                  detectChanges(snapshot.elements, eventData.payload.elements),
+                  null,
+                  2,
+                ),
+              );
+            }
+
+            const a = reconcileElements(
+              snapshot.elements,
+              eventData.payload.elements,
+            );
+            // console.log(
+            //   JSON.stringify(detectChanges(snapshot.elements, a), null, 2),
+            // );
+            snapshot.elements = a;
           });
           socket.on(SCENE_INIT, (roomID: string, data: ArrayBuffer) => {
             socket.broadcast.to(roomID).emit(CLIENT_BROADCAST, data);

src/excalidraw-backend/types/client.broadcast.payload.type.ts → src/excalidraw-backend/types/broadcast.payload.type.ts

@@ -1,4 +1,4 @@
-export enum ClientBroadcastPayloadType {
+export enum BroadcastPayloadType {
   SCENE_INIT = 'SCENE_INIT',
   SCENE_UPDATE = 'SCENE_UPDATE',
 }

src/excalidraw-backend/types/events/index.ts

@@ -1 +1,2 @@
 export * from './idle.state';
+export * from './server.broadcast.payload';

src/excalidraw-backend/types/events/server.broadcast.payload.ts

@@ -0,0 +1,7 @@
+import { ExcalidrawElement } from '../excalidraw.element';
+import { ExcalidrawFile } from '../excalidraw.file';
+
+export type ServerBroadcastPayload = {
+  elements: readonly ExcalidrawElement[];
+  files: readonly ExcalidrawFile[];
+};

src/excalidraw-backend/types/excalidraw.element.ts

@@ -0,0 +1,6 @@
+export type ExcalidrawElement = {
+  id: string;
+  index: number;
+  version: number;
+  versionNonce: number;
+};

src/excalidraw-backend/types/excalidraw.file.ts

@@ -0,0 +1 @@
+export type ExcalidrawFile = Record<string, unknown>;

src/excalidraw-backend/types/index.ts

@@ -1,4 +1,4 @@
-export * from './client.broadcast.payload.type';
+export * from './broadcast.payload.type';
 export * from './collaboration.mode.reasons';
 
 export * from './defaults';
@@ -14,3 +14,5 @@ export * from './socket.io.socket';
 
 export * from './user.info.for.room';
 export * from './user.idle.state';
+
+export * from './excalidraw.element';

src/excalidraw-backend/types/socket.event.data.ts

@@ -1,4 +1,4 @@
-type BasePayload = Record<string, unknown>;
+export type BasePayload = Record<string, unknown>;
 
 export type SocketEventData<TPayload extends BasePayload = BasePayload> = {
   type: string;

src/excalidraw-backend/utils/array.to.map.ts

@@ -0,0 +1,15 @@
+/**
+ * Transforms array of objects containing `id` attribute,
+ * or array of ids (strings), into a Map, keyd by `id`.
+ */
+export const arrayToMap = <T extends { id: string } | string>(
+  items: readonly T[] | Map<string, T>,
+) => {
+  if (items instanceof Map) {
+    return items;
+  }
+  return items.reduce((acc: Map<string, T>, element) => {
+    acc.set(typeof element === 'string' ? element : element.id, element);
+    return acc;
+  }, new Map());
+};

src/excalidraw-backend/utils/decode.incoming.ts

@@ -0,0 +1,25 @@
+import { BasePayload, SocketEventData } from '../types';
+
+/**
+ * Tries to decode incoming binary data.
+ * Throws an exception otherwise.
+ * @param {ArrayBuffer}data The incoming binary data
+ * @returns {SocketEventData} Returns the decoded data in form of event data
+ * @throws {TypeError} Throws an error if the data cannot be decoded
+ * @throws {SyntaxError} Thrown if the data to parse is not valid JSON.
+ */
+export const tryDecodeIncoming = <TPayload extends BasePayload>(
+  data: ArrayBuffer,
+): SocketEventData<TPayload> | never => {
+  const strEventData = tryDecodeBinary(data);
+
+  return JSON.parse(strEventData) as SocketEventData<TPayload>;
+};
+/**
+ *
+ * @throws {TypeError} Throws an error if the data cannot be decoded
+ */
+export const tryDecodeBinary = (data: ArrayBuffer): string => {
+  const decoder = new TextDecoder('utf-8');
+  return decoder.decode(data);
+};

src/excalidraw-backend/utils/detect.changes.ts

@@ -0,0 +1,66 @@
+interface Item {
+  id: string;
+  [key: string]: any;
+}
+
+function arraysEqual(arr1: any[], arr2: any[]): boolean {
+  if (arr1.length !== arr2.length) return false;
+  for (let i = 0; i < arr1.length; i++) {
+    if (arr1[i] !== arr2[i]) return false;
+  }
+  return true;
+}
+
+export function detectChanges(
+  oldArray: readonly Item[],
+  newArray: readonly Item[],
+) {
+  const changes = {
+    added: [] as Item[],
+    removed: [] as Item[],
+    updated: [] as {
+      id: string;
+      changes: { before: Partial<Item>; after: Partial<Item> };
+    }[],
+  };
+
+  const oldMap = new Map<string, Item>();
+  const newMap = new Map<string, Item>();
+
+  oldArray.forEach((item) => oldMap.set(item.id, item));
+  newArray.forEach((item) => newMap.set(item.id, item));
+
+  // Detect added and updated items
+  newMap.forEach((newItem, id) => {
+    const oldItem = oldMap.get(id);
+    if (!oldItem) {
+      changes.added.push(newItem);
+    } else {
+      const before: Partial<Item> = {};
+      const after: Partial<Item> = {};
+      for (const key in newItem) {
+        if (Array.isArray(newItem[key]) && Array.isArray(oldItem[key])) {
+          if (!arraysEqual(newItem[key], oldItem[key])) {
+            before[key] = oldItem[key];
+            after[key] = newItem[key];
+          }
+        } else if (newItem[key] !== oldItem[key]) {
+          before[key] = oldItem[key];
+          after[key] = newItem[key];
+        }
+      }
+      if (Object.keys(before).length > 0) {
+        changes.updated.push({ id, changes: { before, after } });
+      }
+    }
+  });
+
+  // Detect removed items
+  oldMap.forEach((oldItem, id) => {
+    if (!newMap.has(id)) {
+      changes.removed.push(oldItem);
+    }
+  });
+
+  return changes;
+}

src/excalidraw-backend/utils/handlers.ts

@@ -15,8 +15,9 @@ import {
 } from '../types';
 import { minCollaboratorsInRoom } from '../types';
 import { SocketEventData } from '../types';
-import { IdleStatePayload } from '../types/events';
+import { IdleStatePayload, ServerBroadcastPayload } from '../types/events';
 import { closeConnection } from './util';
+import { tryDecodeBinary, tryDecodeIncoming } from './decode.incoming';
 
 const fetchSocketsSafe = async (
   wsServer: SocketIoServer,
@@ -157,17 +158,13 @@ export const idleStateEventHandler = (
 ) => {
   socket.broadcast.to(roomID).emit(IDLE_STATE, data);
 
-  const decoder = new TextDecoder('utf-8');
-  const strEventData = decoder.decode(data);
   try {
-    const eventData = JSON.parse(
-      strEventData,
-    ) as SocketEventData<IdleStatePayload>;
+    const eventData = tryDecodeIncoming<IdleStatePayload>(data);
     socket.data.state = eventData.payload.userState;
   } catch (e) {
     logger.error({
       message: e?.message ?? JSON.stringify(e),
-      data: strEventData,
+      data: e instanceof SyntaxError ? tryDecodeBinary(data) : undefined,
     });
   }
 };

src/excalidraw-backend/utils/reconcile.ts

@@ -0,0 +1,94 @@
+import { ExcalidrawElement } from '../types';
+import { arrayToMap } from './array.to.map';
+
+const shouldDiscardRemoteElement = (
+  local: ExcalidrawElement | undefined,
+  remote: ExcalidrawElement,
+): boolean => {
+  return !!(
+    local &&
+    // local element is newer
+    (local.version > remote.version ||
+      // resolve conflicting edits deterministically by taking the one with
+      // the lowest versionNonce
+      (local.version === remote.version &&
+        local.versionNonce < remote.versionNonce))
+  );
+};
+
+/*const validateIndicesThrottled = throttle(
+  (
+    orderedElements: readonly OrderedExcalidrawElement[],
+    localElements: readonly OrderedExcalidrawElement[],
+    remoteElements: readonly RemoteExcalidrawElement[],
+  ) => {
+    if (
+      import.meta.env.DEV ||
+      import.meta.env.MODE === ENV.TEST ||
+      window?.DEBUG_FRACTIONAL_INDICES
+    ) {
+      // create new instances due to the mutation
+      const elements = syncInvalidIndices(
+        orderedElements.map((x) => ({ ...x })),
+      );
+
+      validateFractionalIndices(elements, {
+        // throw in dev & test only, to remain functional on `DEBUG_FRACTIONAL_INDICES`
+        shouldThrow: import.meta.env.DEV || import.meta.env.MODE === ENV.TEST,
+        includeBoundTextValidation: true,
+        reconciliationContext: {
+          localElements,
+          remoteElements,
+        },
+      });
+    }
+  },
+  1000 * 60,
+  { leading: true, trailing: false },
+);*/
+
+export const reconcileElements = (
+  localElements: readonly ExcalidrawElement[],
+  remoteElements: readonly ExcalidrawElement[],
+): ExcalidrawElement[] => {
+  const localElementsMap = arrayToMap(localElements);
+  const reconciledElements: ExcalidrawElement[] = [];
+  const added = new Set<string>();
+
+  // process remote elements
+  for (const remoteElement of remoteElements) {
+    if (!added.has(remoteElement.id)) {
+      const localElement = localElementsMap.get(remoteElement.id);
+      const discardRemoteElement = shouldDiscardRemoteElement(
+        localElement,
+        remoteElement,
+      );
+
+      if (localElement && discardRemoteElement) {
+        reconciledElements.push(localElement);
+        added.add(localElement.id);
+      } else {
+        reconciledElements.push(remoteElement);
+        added.add(remoteElement.id);
+      }
+    }
+  }
+
+  // process remaining local elements
+  for (const localElement of localElements) {
+    if (!added.has(localElement.id)) {
+      reconciledElements.push(localElement);
+      added.add(localElement.id);
+    }
+  }
+
+  // const orderedElements = orderByFractionalIndex(reconciledElements);
+
+  // validateIndicesThrottled(orderedElements, localElements, remoteElements);
+
+  // de-duplicate indices
+  // syncInvalidIndices(orderedElements);
+
+  // return orderedElements as ReconciledExcalidrawElement[];
+  return reconciledElements;
+};