refactor: workspace manager (#5060)

packages/frontend/workspace/src/utils/__tests__/async-queue.spec.ts

@@ -0,0 +1,45 @@
+import { describe, expect, test, vi } from 'vitest';
+
+import { AsyncQueue } from '../async-queue';
+
+describe('async-queue', () => {
+  test('push & pop', async () => {
+    const queue = new AsyncQueue();
+    queue.push(1, 2, 3);
+    expect(queue.length).toBe(3);
+    expect(await queue.next()).toBe(1);
+    expect(await queue.next()).toBe(2);
+    expect(await queue.next()).toBe(3);
+    expect(queue.length).toBe(0);
+  });
+
+  test('await', async () => {
+    const queue = new AsyncQueue<number>();
+    queue.push(1, 2);
+    expect(await queue.next()).toBe(1);
+    expect(await queue.next()).toBe(2);
+
+    let v = -1;
+
+    // setup 2 pop tasks
+    queue.next().then(next => {
+      v = next;
+    });
+    queue.next().then(next => {
+      v = next;
+    });
+
+    // Wait for 100ms
+    await new Promise(resolve => setTimeout(resolve, 100));
+    // v should not be changed
+    expect(v).toBe(-1);
+
+    // push 3, should trigger the first pop task
+    queue.push(3);
+    await vi.waitFor(() => v === 3);
+
+    // push 4, should trigger the second pop task
+    queue.push(4);
+    await vi.waitFor(() => v === 4);
+  });
+});

packages/frontend/workspace/src/utils/__tests__/buffer-to-blob.spec.ts

@@ -0,0 +1,13 @@
+import { Buffer } from 'node:buffer';
+
+import { describe, expect, test } from 'vitest';
+
+import { isSvgBuffer } from '../buffer-to-blob';
+
+describe('isSvgBuffer', () => {
+  test('basic', async () => {
+    expect(isSvgBuffer(Buffer.from('<svg></svg>'))).toBe(true);
+    expect(isSvgBuffer(Buffer.from(' \n\r\t<svg></svg>'))).toBe(true);
+    expect(isSvgBuffer(Buffer.from('<123>'))).toBe(false);
+  });
+});

packages/frontend/workspace/src/utils/__tests__/throw-if-aborted.spec.ts

@@ -0,0 +1,13 @@
+import { describe, expect, test } from 'vitest';
+
+import { throwIfAborted } from '../throw-if-aborted';
+
+describe('throw-if-aborted', () => {
+  test('basic', async () => {
+    const abortController = new AbortController();
+    const abortSignal = abortController.signal;
+    expect(throwIfAborted(abortSignal)).toBe(true);
+    abortController.abort('TEST_ABORT');
+    expect(() => throwIfAborted(abortSignal)).toThrowError('TEST_ABORT');
+  });
+});

packages/frontend/workspace/src/utils/affine-io.ts

@@ -0,0 +1,15 @@
+import { Manager } from 'socket.io-client';
+
+let ioManager: Manager | null = null;
+
+// use lazy initialization socket.io io manager
+export function getIoManager(): Manager {
+  if (ioManager) {
+    return ioManager;
+  }
+  ioManager = new Manager(runtimeConfig.serverUrlPrefix + '/', {
+    autoConnect: false,
+    transports: ['websocket'],
+  });
+  return ioManager;
+}

packages/frontend/workspace/src/utils/async-queue.ts

@@ -0,0 +1,101 @@
+export class AsyncQueue<T> {
+  private _queue: T[];
+
+  private _resolveUpdate: (() => void) | null = null;
+  private _waitForUpdate: Promise<void> | null = null;
+
+  constructor(init: T[] = []) {
+    this._queue = init;
+  }
+
+  get length() {
+    return this._queue.length;
+  }
+
+  async next(
+    abort?: AbortSignal,
+    dequeue: (arr: T[]) => T | undefined = a => a.shift()
+  ): Promise<T> {
+    const update = dequeue(this._queue);
+    if (update) {
+      return update;
+    } else {
+      if (!this._waitForUpdate) {
+        this._waitForUpdate = new Promise(resolve => {
+          this._resolveUpdate = resolve;
+        });
+      }
+
+      await Promise.race([
+        this._waitForUpdate,
+        new Promise((_, reject) => {
+          if (abort?.aborted) {
+            reject(abort?.reason);
+          }
+          abort?.addEventListener('abort', () => {
+            reject(abort.reason);
+          });
+        }),
+      ]);
+
+      return this.next(abort, dequeue);
+    }
+  }
+
+  push(...updates: T[]) {
+    this._queue.push(...updates);
+    if (this._resolveUpdate) {
+      const resolve = this._resolveUpdate;
+      this._resolveUpdate = null;
+      this._waitForUpdate = null;
+      resolve();
+    }
+  }
+
+  remove(predicate: (update: T) => boolean) {
+    const index = this._queue.findIndex(predicate);
+    if (index !== -1) {
+      this._queue.splice(index, 1);
+    }
+  }
+
+  find(predicate: (update: T) => boolean) {
+    return this._queue.find(predicate);
+  }
+
+  clear() {
+    this._queue = [];
+  }
+}
+
+export class PriorityAsyncQueue<
+  T extends { id: string },
+> extends AsyncQueue<T> {
+  constructor(
+    init: T[] = [],
+    public readonly priorityTarget: SharedPriorityTarget = new SharedPriorityTarget()
+  ) {
+    super(init);
+  }
+
+  override next(abort?: AbortSignal | undefined): Promise<T> {
+    return super.next(abort, arr => {
+      if (this.priorityTarget.priorityRule !== null) {
+        const index = arr.findIndex(
+          update => this.priorityTarget.priorityRule?.(update.id)
+        );
+        if (index !== -1) {
+          return arr.splice(index, 1)[0];
+        }
+      }
+      return arr.shift();
+    });
+  }
+}
+
+/**
+ * Shared priority target can be shared by multiple queues.
+ */
+export class SharedPriorityTarget {
+  public priorityRule: ((id: string) => boolean) | null = null;
+}

packages/frontend/workspace/src/utils/base64.ts

@@ -0,0 +1,28 @@
+export function uint8ArrayToBase64(array: Uint8Array): Promise<string> {
+  return new Promise<string>(resolve => {
+    // Create a blob from the Uint8Array
+    const blob = new Blob([array]);
+
+    const reader = new FileReader();
+    reader.onload = function () {
+      const dataUrl = reader.result as string | null;
+      if (!dataUrl) {
+        resolve('');
+        return;
+      }
+      // The result includes the `data:` URL prefix and the MIME type. We only want the Base64 data
+      const base64 = dataUrl.split(',')[1];
+      resolve(base64);
+    };
+
+    reader.readAsDataURL(blob);
+  });
+}
+
+export function base64ToUint8Array(base64: string) {
+  const binaryString = atob(base64);
+  const binaryArray = binaryString.split('').map(function (char) {
+    return char.charCodeAt(0);
+  });
+  return new Uint8Array(binaryArray);
+}

packages/frontend/workspace/src/utils/buffer-to-blob.ts

@@ -0,0 +1,50 @@
+import isSvg from 'is-svg';
+
+function fastCheckIsNotSvg(buffer: Uint8Array) {
+  // check first non-whitespace character is not '<svg'
+  for (let i = 0; i < buffer.length; i++) {
+    const ch = buffer[i];
+
+    // skip whitespace
+    if (
+      ch === 0x20 /* \s */ ||
+      ch === 0x09 /* \t */ ||
+      ch === 0x0b /* \v */ ||
+      ch === 0x0c /* \f */ ||
+      ch === 0x0a /* \n */ ||
+      ch === 0x0d /* \r */ ||
+      ch === 0xa0
+    ) {
+      continue;
+    }
+
+    return !(
+      buffer[i] === /* '<' */ 0x3c &&
+      buffer[i + 1] === /* 's' */ 0x73 &&
+      buffer[i + 2] === /* 'v' */ 0x76 &&
+      buffer[i + 3] === /* 'g' */ 0x67
+    );
+  }
+
+  return true;
+}
+
+// this has a overhead of converting to string for testing if it is svg.
+export function isSvgBuffer(buffer: Uint8Array) {
+  if (fastCheckIsNotSvg(buffer)) {
+    return false;
+  }
+  const decoder = new TextDecoder('utf-8');
+  const str = decoder.decode(buffer);
+  return isSvg(str);
+}
+
+export function bufferToBlob(buffer: Uint8Array | ArrayBuffer) {
+  const isSVG = isSvgBuffer(
+    buffer instanceof ArrayBuffer ? new Uint8Array(buffer) : buffer
+  );
+  // for svg blob, we need to explicitly set the type to image/svg+xml
+  return isSVG
+    ? new Blob([buffer], { type: 'image/svg+xml' })
+    : new Blob([buffer]);
+}

packages/frontend/workspace/src/utils/merge-updates.ts

@@ -0,0 +1,17 @@
+import { applyUpdate, Doc, encodeStateAsUpdate } from 'yjs';
+
+export function mergeUpdates(updates: Uint8Array[]) {
+  if (updates.length === 0) {
+    return new Uint8Array();
+  }
+  if (updates.length === 1) {
+    return updates[0];
+  }
+  const doc = new Doc();
+  doc.transact(() => {
+    updates.forEach(update => {
+      applyUpdate(doc, update);
+    });
+  });
+  return encodeStateAsUpdate(doc);
+}

packages/frontend/workspace/src/utils/throw-if-aborted.ts

@@ -0,0 +1,9 @@
+// because AbortSignal.throwIfAborted is not available in abortcontroller-polyfill
+export function throwIfAborted(abort?: AbortSignal) {
+  if (abort?.aborted) {
+    throw new Error(abort.reason);
+  }
+  return true;
+}
+
+export const MANUALLY_STOP = 'manually-stop';