[MMB-200] Implement cursor offsets rather than intervals

src/CursorBatching.ts

@@ -4,10 +4,10 @@ import { CURSOR_UPDATE } from './CursorConstants.js';
 import type { CursorUpdate } from './types.js';
 import type { CursorsOptions } from './types.js';
 
-type OutgoingBuffer = Pick<CursorUpdate, 'position' | 'data'>[];
+type OutgoingBuffer = { cursor: Pick<CursorUpdate, 'position' | 'data'>; offset: number };
 
 export default class CursorBatching {
-  outgoingBuffers: OutgoingBuffer = [];
+  outgoingBuffer: OutgoingBuffer[] = [];
 
   batchTime: number;
 
@@ -20,15 +20,31 @@ export default class CursorBatching {
   // Set to `true` if there is more than one user listening to cursors
   shouldSend: boolean = false;
 
+  // Used for tracking offsets in the buffer
+  bufferStartTimestamp: number = 0;
+
   constructor(readonly outboundBatchInterval: CursorsOptions['outboundBatchInterval']) {
     this.batchTime = outboundBatchInterval;
   }
 
   pushCursorPosition(channel: Types.RealtimeChannelPromise, cursor: Pick<CursorUpdate, 'position' | 'data'>) {
     // Ignore the cursor update if there is no one listening
     if (!this.shouldSend) return;
+
+    const timestamp = new Date().getTime();
+
+    let offset: number;
+    // First update in the buffer is always 0
+    if (this.outgoingBuffer.length === 0) {
+      offset = 0;
+      this.bufferStartTimestamp = timestamp;
+    } else {
+      // Add the offset compared to the first update in the buffer
+      offset = timestamp - this.bufferStartTimestamp;
+    }
+
     this.hasMovement = true;
-    this.pushToBuffer(cursor);
+    this.pushToBuffer({ cursor, offset });
     this.publishFromBuffer(channel, CURSOR_UPDATE);
   }
 
@@ -40,8 +56,8 @@ export default class CursorBatching {
     this.batchTime = batchTime;
   }
 
-  private pushToBuffer(value: Pick<CursorUpdate, 'position' | 'data'>) {
-    this.outgoingBuffers.push(value);
+  private pushToBuffer(value: OutgoingBuffer) {
+    this.outgoingBuffer.push(value);
   }
 
   private async publishFromBuffer(channel: Types.RealtimeChannelPromise, eventName: string) {
@@ -57,10 +73,10 @@ export default class CursorBatching {
       return;
     }
     // Must be copied here to avoid a race condition where the buffer is cleared before the publish happens
-    const bufferCopy = [...this.outgoingBuffers];
+    const bufferCopy = [...this.outgoingBuffer];
     channel.publish(eventName, bufferCopy);
     setTimeout(() => this.batchToChannel(channel, eventName), this.batchTime);
-    this.outgoingBuffers = [];
+    this.outgoingBuffer = [];
     this.hasMovement = false;
     this.isRunning = true;
   }

src/CursorDispensing.ts

@@ -1,47 +1,26 @@
-import { clamp } from './utilities/math.js';
-
 import { type CursorUpdate } from './types.js';
 import { type RealtimeMessage } from './utilities/types.js';
 
 export default class CursorDispensing {
-  private buffer: Record<string, CursorUpdate[]> = {};
-  private handlerRunning: boolean = false;
-  private timerIds: ReturnType<typeof setTimeout>[] = [];
-
-  constructor(private emitCursorUpdate: (update: CursorUpdate) => void, private getCurrentBatchTime: () => number) {}
-
-  emitFromBatch(batchDispenseInterval: number) {
-    if (!this.bufferHaveData()) {
-      this.handlerRunning = false;
-      return;
-    }
-
-    this.handlerRunning = true;
+  private buffer: Record<string, { cursor: CursorUpdate; offset: number }[]> = {};
 
-    const processBuffer = () => {
-      for (let connectionId in this.buffer) {
-        const buffer = this.buffer[connectionId];
-        const update = buffer.shift();
+  constructor(private emitCursorUpdate: (update: CursorUpdate) => void) {}
 
-        if (!update) continue;
-        this.emitCursorUpdate(update);
-      }
+  setEmitCursorUpdate(update: CursorUpdate) {
+    this.emitCursorUpdate(update);
+  }
 
-      if (this.bufferHaveData()) {
-        this.emitFromBatch(this.calculateDispenseInterval());
-      } else {
-        this.handlerRunning = false;
-      }
+  emitFromBatch() {
+    for (let connectionId in this.buffer) {
+      const buffer = this.buffer[connectionId];
+      const update = buffer.shift();
 
-      this.timerIds.shift();
-    };
+      if (!update) continue;
+      setTimeout(() => this.setEmitCursorUpdate(update.cursor), update.offset);
+    }
 
-    if (typeof document !== 'undefined' && document.visibilityState === 'hidden') {
-      this.timerIds.forEach((id) => clearTimeout(id));
-      this.timerIds = [];
-      processBuffer();
-    } else {
-      this.timerIds.push(setTimeout(processBuffer, batchDispenseInterval));
+    if (this.bufferHaveData()) {
+      this.emitFromBatch();
     }
   }
 
@@ -53,33 +32,29 @@ export default class CursorDispensing {
     );
   }
 
-  calculateDispenseInterval(): number {
-    const bufferLengths = Object.entries(this.buffer).map(([, v]) => v.length);
-    const highest = bufferLengths.sort()[bufferLengths.length - 1];
-    const finalOutboundBatchInterval = this.getCurrentBatchTime();
-    return Math.floor(clamp(finalOutboundBatchInterval / highest, 1, 1000 / 15));
-  }
-
   processBatch(message: RealtimeMessage) {
-    const updates: CursorUpdate[] = message.data || [];
+    const updates: { cursor: CursorUpdate; offset: number }[] = message.data || [];
 
-    updates.forEach((update) => {
+    updates.forEach((update: { cursor: CursorUpdate; offset: number }) => {
       const enhancedMsg = {
-        clientId: message.clientId,
-        connectionId: message.connectionId,
-        position: update.position,
-        data: update.data,
+        cursor: {
+          clientId: message.clientId,
+          connectionId: message.connectionId,
+          position: update.cursor.position,
+          data: update.cursor.data,
+        },
+        offset: update.offset,
       };
 
-      if (this.buffer[enhancedMsg.connectionId]) {
-        this.buffer[enhancedMsg.connectionId].push(enhancedMsg);
+      if (this.buffer[enhancedMsg.cursor.connectionId]) {
+        this.buffer[enhancedMsg.cursor.connectionId].push(enhancedMsg);
       } else {
-        this.buffer[enhancedMsg.connectionId] = [enhancedMsg];
+        this.buffer[enhancedMsg.cursor.connectionId] = [enhancedMsg];
       }
     });
 
-    if (!this.handlerRunning && this.bufferHaveData()) {
-      this.emitFromBatch(this.calculateDispenseInterval());
+    if (this.bufferHaveData()) {
+      this.emitFromBatch();
     }
   }
 }

src/Cursors.test.ts

@@ -67,7 +67,10 @@ describe('Cursors', () => {
     it<CursorsTestContext>('emits a cursorsUpdate event', ({ space, dispensing, batching, fakeMessageStub }) => {
       const fakeMessage = {
         ...fakeMessageStub,
-        data: [{ position: { x: 1, y: 1 } }, { position: { x: 1, y: 2 }, data: { color: 'red' } }],
+        data: [
+          { cursor: { position: { x: 1, y: 1 } } },
+          { cursor: { position: { x: 1, y: 2 }, data: { color: 'red' } } },
+        ],
       };
 
       const spy = vitest.fn();
@@ -151,16 +154,19 @@ describe('Cursors', () => {
 
       it<CursorsTestContext>('creates an outgoingBuffer for a new cursor movement', ({ batching, channel }) => {
         batching.pushCursorPosition(channel, { position: { x: 1, y: 1 }, data: {} });
-        expect(batching.outgoingBuffers).toEqual([{ position: { x: 1, y: 1 }, data: {} }]);
+        expect(batching.outgoingBuffer).toEqual([{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }]);
       });
 
       it<CursorsTestContext>('adds cursor data to an existing buffer', ({ batching, channel }) => {
+        vi.useFakeTimers();
         batching.pushCursorPosition(channel, { position: { x: 1, y: 1 }, data: {} });
-        expect(batching.outgoingBuffers).toEqual([{ position: { x: 1, y: 1 }, data: {} }]);
+        expect(batching.outgoingBuffer).toEqual([{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }]);
+
+        vi.advanceTimersByTime(10);
         batching.pushCursorPosition(channel, { position: { x: 2, y: 2 }, data: {} });
-        expect(batching.outgoingBuffers).toEqual([
-          { position: { x: 1, y: 1 }, data: {} },
-          { position: { x: 2, y: 2 }, data: {} },
+        expect(batching.outgoingBuffer).toEqual([
+          { cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 },
+          { cursor: { position: { x: 2, y: 2 }, data: {} }, offset: 10 },
         ]);
       });
 
@@ -195,17 +201,19 @@ describe('Cursors', () => {
 
       it<CursorsTestContext>('should publish the cursor buffer', async ({ batching, channel }) => {
         batching.hasMovement = true;
-        batching.outgoingBuffers = [{ position: { x: 1, y: 1 }, data: {} }];
+        batching.outgoingBuffer = [{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }];
         const spy = vi.spyOn(channel, 'publish');
         await batching['batchToChannel'](channel, CURSOR_UPDATE);
-        expect(spy).toHaveBeenCalledWith(CURSOR_UPDATE, [{ position: { x: 1, y: 1 }, data: {} }]);
+        expect(spy).toHaveBeenCalledWith(CURSOR_UPDATE, [
+          { cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 },
+        ]);
       });
 
       it<CursorsTestContext>('should clear the buffer', async ({ batching, channel }) => {
         batching.hasMovement = true;
-        batching.outgoingBuffers = [{ position: { x: 1, y: 1 }, data: {} }];
+        batching.outgoingBuffer = [{ cursor: { position: { x: 1, y: 1 }, data: {} }, offset: 0 }];
         await batching['batchToChannel'](channel, CURSOR_UPDATE);
-        expect(batching.outgoingBuffers).toEqual([]);
+        expect(batching.outgoingBuffer).toEqual([]);
       });
 
       it<CursorsTestContext>('should set hasMovements to false', async ({ batching, channel }) => {
@@ -233,28 +241,12 @@ describe('Cursors', () => {
         expect(spy).not.toHaveBeenCalled();
       });
 
-      it<CursorsTestContext>('does not call emitFromBatch if the loop is already running', async ({
-        dispensing,
-        fakeMessageStub,
-      }) => {
-        const spy = vi.spyOn(dispensing, 'emitFromBatch');
-
-        const fakeMessage = {
-          ...fakeMessageStub,
-          data: [{ position: { x: 1, y: 1 } }],
-        };
-
-        dispensing['handlerRunning'] = true;
-        dispensing.processBatch(fakeMessage);
-        expect(spy).not.toHaveBeenCalled();
-      });
-
       it<CursorsTestContext>('call emitFromBatch if there are updates', async ({ dispensing, fakeMessageStub }) => {
         const spy = vi.spyOn(dispensing, 'emitFromBatch');
 
         const fakeMessage = {
           ...fakeMessageStub,
-          data: [{ position: { x: 1, y: 1 } }],
+          data: [{ cursor: { position: { x: 1, y: 1 } } }],
         };
 
         dispensing.processBatch(fakeMessage);
@@ -268,70 +260,70 @@ describe('Cursors', () => {
         const fakeMessage = {
           ...fakeMessageStub,
           data: [
-            { position: { x: 1, y: 1 } },
-            { position: { x: 2, y: 3 }, data: { color: 'blue' } },
-            { position: { x: 5, y: 4 } },
+            { cursor: { position: { x: 1, y: 1 } }, offset: 10 },
+            { cursor: { position: { x: 2, y: 3 }, data: { color: 'blue' } }, offset: 20 },
+            { cursor: { position: { x: 5, y: 4 } }, offset: 30 },
           ],
         };
+        vi.useFakeTimers();
 
+        const spy = vi.spyOn(dispensing, 'setEmitCursorUpdate');
         dispensing.processBatch(fakeMessage);
-        expect(dispensing['buffer']).toEqual({
-          connectionId: [
-            {
-              position: { x: 1, y: 1 },
-              data: undefined,
-              clientId: 'clientId',
-              connectionId: 'connectionId',
-            },
-            {
-              position: { x: 2, y: 3 },
-              data: { color: 'blue' },
-              clientId: 'clientId',
-              connectionId: 'connectionId',
-            },
-            {
-              position: { x: 5, y: 4 },
-              data: undefined,
-              clientId: 'clientId',
-              connectionId: 'connectionId',
-            },
-          ],
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledWith({
+          position: { x: 1, y: 1 },
+          data: undefined,
+          clientId: 'clientId',
+          connectionId: 'connectionId',
+        });
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledWith({
+          position: { x: 2, y: 3 },
+          data: { color: 'blue' },
+          clientId: 'clientId',
+          connectionId: 'connectionId',
         });
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledWith({
+          position: { x: 5, y: 4 },
+          data: undefined,
+          clientId: 'clientId',
+          connectionId: 'connectionId',
+        });
+        expect(spy).toHaveBeenCalledTimes(3);
       });
 
-      it<CursorsTestContext>('runs until the batch is empty', async ({ dispensing, batching, fakeMessageStub }) => {
+      it<CursorsTestContext>('runs until the batch is empty', async ({ dispensing, fakeMessageStub }) => {
         vi.useFakeTimers();
 
         const fakeMessage = {
           ...fakeMessageStub,
           data: [
-            { position: { x: 1, y: 1 } },
-            { position: { x: 2, y: 3 }, data: { color: 'blue' } },
-            { position: { x: 5, y: 4 } },
+            { cursor: { position: { x: 1, y: 1 } }, offset: 10 },
+            { cursor: { position: { x: 2, y: 3 }, data: { color: 'blue' } }, offset: 20 },
+            { cursor: { position: { x: 5, y: 4 } }, offset: 30 },
           ],
         };
 
-        expect(dispensing['handlerRunning']).toBe(false);
+        const spy = vi.spyOn(dispensing, 'setEmitCursorUpdate');
+
         expect(dispensing.bufferHaveData()).toBe(false);
 
         dispensing.processBatch(fakeMessage);
-        expect(dispensing['buffer']['connectionId']).toHaveLength(3);
-        expect(dispensing['handlerRunning']).toBe(true);
-        expect(dispensing.bufferHaveData()).toBe(true);
-        vi.advanceTimersByTime(batching.batchTime / 2);
-
-        expect(dispensing['buffer']['connectionId']).toHaveLength(2);
-        expect(dispensing['handlerRunning']).toBe(true);
-        expect(dispensing.bufferHaveData()).toBe(true);
-
-        vi.advanceTimersByTime(batching.batchTime / 2);
-        expect(dispensing['buffer']['connectionId']).toHaveLength(1);
-        expect(dispensing['handlerRunning']).toBe(true);
-        expect(dispensing.bufferHaveData()).toBe(true);
-
-        vi.advanceTimersByTime(batching.batchTime);
-        expect(dispensing['buffer']['connectionId']).toHaveLength(0);
-        expect(dispensing['handlerRunning']).toBe(false);
+        expect(spy).toHaveBeenCalledTimes(0);
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledTimes(1);
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledTimes(2);
+
+        vi.advanceTimersByTime(10);
+        expect(spy).toHaveBeenCalledTimes(3);
+
         expect(dispensing.bufferHaveData()).toBe(false);
 
         vi.useRealTimers();
@@ -498,6 +490,7 @@ describe('Cursors', () => {
             x: 2,
             y: 3,
           },
+          offset: 0,
         },
       };