feat: remove workspace & cloud sync

packages/data-center/src/datacenter/datacenter.ts

@@ -13,18 +13,18 @@ export class DataCenter {
   private readonly _config;
   private readonly _logger;
 
-  static async init(): Promise<DataCenter> {
-    const dc = new DataCenter();
+  static async init(debug: boolean): Promise<DataCenter> {
+    const dc = new DataCenter(debug);
     dc.addProvider(AffineProvider);
     dc.addProvider(LocalProvider);
 
     return dc;
   }
 
-  private constructor() {
+  private constructor(debug: boolean) {
     this._config = getKVConfigure('sys');
     this._logger = getLogger('dc');
-    this._logger.enabled = true;
+    this._logger.enabled = debug;
   }
 
   private addProvider(provider: typeof BaseProvider) {
@@ -43,7 +43,7 @@ export class DataCenter {
     throw Error(`Provider ${providerId} not found`);
   }
 
-  private async _initWorkspace(id: string, pid: string): Promise<BaseProvider> {
+  private async _getWorkspace(id: string, pid: string): Promise<BaseProvider> {
     this._logger(`Init workspace ${id} with ${pid}`);
 
     const providerId = await this._getProvider(id, pid);
@@ -57,6 +57,7 @@ export class DataCenter {
 
     await provider.init({
       config: getKVConfigure(id),
+      debug: this._logger.enabled,
       logger: this._logger.extend(`${Provider.id}:${id}`),
       workspace,
     });
@@ -66,13 +67,13 @@ export class DataCenter {
     return provider;
   }
 
-  async initWorkspace(
+  async getWorkspace(
     id: string,
     provider = 'local'
   ): Promise<Workspace | null> {
     if (id) {
       if (!this._workspaces.has(id)) {
-        this._workspaces.set(id, this._initWorkspace(id, provider));
+        this._workspaces.set(id, this._getWorkspace(id, provider));
       }
       const workspace = this._workspaces.get(id);
       assert(workspace);
@@ -86,10 +87,24 @@ export class DataCenter {
     return config.set(key, value);
   }
 
-  async getWorkspaceList() {
+  async listWorkspace() {
     const keys = await this._config.keys();
     return keys
       .filter(k => k.startsWith('workspace:'))
       .map(k => k.split(':')[1]);
   }
+
+  async removeWorkspace(id: string) {
+    await this._config.delete(`workspace:${id}:provider`);
+    const provider = await this._workspaces.get(id);
+    if (provider) {
+      this._workspaces.delete(id);
+      await provider.clear();
+    }
+  }
+
+  async clearWorkspaces() {
+    const workspaces = await this.listWorkspace();
+    await Promise.all(workspaces.map(id => this.removeWorkspace(id)));
+  }
 }

packages/data-center/src/datacenter/index.ts

@@ -4,9 +4,9 @@ import { DataCenter } from './datacenter.js';
 const _initializeDataCenter = () => {
   let _dataCenterInstance: Promise<DataCenter>;
 
-  return () => {
+  return (debug = true) => {
     if (!_dataCenterInstance) {
-      _dataCenterInstance = DataCenter.init();
+      _dataCenterInstance = DataCenter.init(debug);
     }
 
     return _dataCenterInstance;

packages/data-center/src/datacenter/provider/affine/index.ts

@@ -5,11 +5,13 @@ import type { InitialParams } from '../index.js';
 import { LocalProvider } from '../local/index.js';
 
 import { downloadWorkspace } from './apis.js';
+import { WebsocketProvider } from './sync.js';
 import { token, Callback } from './token.js';
 
 export class AffineProvider extends LocalProvider {
   static id = 'affine';
   private _onTokenRefresh?: Callback = undefined;
+  private _ws?: WebsocketProvider;
 
   constructor() {
     super();
@@ -64,7 +66,21 @@ export class AffineProvider extends LocalProvider {
       try {
         const updates = await downloadWorkspace(workspace.room);
         if (updates) {
-          applyUpdate(doc, new Uint8Array(updates));
+          await new Promise(resolve => {
+            doc.once('update', resolve);
+            applyUpdate(doc, new Uint8Array(updates));
+          });
+          // TODO: wait util data loaded
+          this._ws = new WebsocketProvider('/', workspace.room, doc);
+          // Wait for ws synchronization to complete, otherwise the data will be modified in reverse, which can be optimized later
+          await new Promise<void>((resolve, reject) => {
+            // TODO: synced will also be triggered on reconnection after losing sync
+            // There needs to be an event mechanism to emit the synchronization state to the upper layer
+            assert(this._ws);
+            this._ws.once('synced', () => resolve());
+            this._ws.once('lost-connection', () => resolve());
+            this._ws.once('connection-error', () => reject());
+          });
         }
       } catch (e) {
         this._logger('Failed to init cloud workspace', e);

packages/data-center/src/datacenter/provider/affine/sync.js

@@ -0,0 +1,508 @@
+/* eslint-disable no-undef */
+/**
+ * @module provider/websocket
+ */
+
+/* eslint-env browser */
+
+// import * as Y from 'yjs'; // eslint-disable-line
+import * as bc from 'lib0/broadcastchannel';
+import * as time from 'lib0/time';
+import * as encoding from 'lib0/encoding';
+import * as decoding from 'lib0/decoding';
+import * as syncProtocol from 'y-protocols/sync';
+import * as authProtocol from 'y-protocols/auth';
+import * as awarenessProtocol from 'y-protocols/awareness';
+import { Observable } from 'lib0/observable';
+import * as math from 'lib0/math';
+import * as url from 'lib0/url';
+
+export const messageSync = 0;
+export const messageQueryAwareness = 3;
+export const messageAwareness = 1;
+export const messageAuth = 2;
+
+/**
+ *                       encoder,          decoder,          provider,          emitSynced, messageType
+ * @type {Array<function(encoding.Encoder, decoding.Decoder, WebsocketProvider, boolean,    number):void>}
+ */
+const messageHandlers = [];
+
+messageHandlers[messageSync] = (
+  encoder,
+  decoder,
+  provider,
+  emitSynced,
+  _messageType
+) => {
+  encoding.writeVarUint(encoder, messageSync);
+  const syncMessageType = syncProtocol.readSyncMessage(
+    decoder,
+    encoder,
+    provider.doc,
+    provider
+  );
+  if (
+    emitSynced &&
+    syncMessageType === syncProtocol.messageYjsSyncStep2 &&
+    !provider.synced
+  ) {
+    provider.synced = true;
+  }
+};
+
+messageHandlers[messageQueryAwareness] = (
+  encoder,
+  _decoder,
+  provider,
+  _emitSynced,
+  _messageType
+) => {
+  encoding.writeVarUint(encoder, messageAwareness);
+  encoding.writeVarUint8Array(
+    encoder,
+    awarenessProtocol.encodeAwarenessUpdate(
+      provider.awareness,
+      Array.from(provider.awareness.getStates().keys())
+    )
+  );
+};
+
+messageHandlers[messageAwareness] = (
+  _encoder,
+  decoder,
+  provider,
+  _emitSynced,
+  _messageType
+) => {
+  awarenessProtocol.applyAwarenessUpdate(
+    provider.awareness,
+    decoding.readVarUint8Array(decoder),
+    provider
+  );
+};
+
+messageHandlers[messageAuth] = (
+  _encoder,
+  decoder,
+  provider,
+  _emitSynced,
+  _messageType
+) => {
+  authProtocol.readAuthMessage(decoder, provider.doc, (_ydoc, reason) =>
+    permissionDeniedHandler(provider, reason)
+  );
+};
+
+// @todo - this should depend on awareness.outdatedTime
+const messageReconnectTimeout = 30000;
+
+/**
+ * @param {WebsocketProvider} provider
+ * @param {string} reason
+ */
+const permissionDeniedHandler = (provider, reason) =>
+  console.warn(`Permission denied to access ${provider.url}.\n${reason}`);
+
+/**
+ * @param {WebsocketProvider} provider
+ * @param {Uint8Array} buf
+ * @param {boolean} emitSynced
+ * @return {encoding.Encoder}
+ */
+const readMessage = (provider, buf, emitSynced) => {
+  const decoder = decoding.createDecoder(buf);
+  const encoder = encoding.createEncoder();
+  const messageType = decoding.readVarUint(decoder);
+  const messageHandler = provider.messageHandlers[messageType];
+  if (/** @type {any} */ (messageHandler)) {
+    messageHandler(encoder, decoder, provider, emitSynced, messageType);
+  } else {
+    console.error('Unable to compute message');
+  }
+  return encoder;
+};
+
+/**
+ * @param {WebsocketProvider} provider
+ */
+const setupWS = provider => {
+  if (provider.shouldConnect && provider.ws === null) {
+    const websocket = new provider._WS(provider.url);
+    websocket.binaryType = 'arraybuffer';
+    provider.ws = websocket;
+    provider.wsconnecting = true;
+    provider.wsconnected = false;
+    provider.synced = false;
+
+    websocket.onmessage = event => {
+      provider.wsLastMessageReceived = time.getUnixTime();
+      const encoder = readMessage(provider, new Uint8Array(event.data), true);
+      if (encoding.length(encoder) > 1) {
+        websocket.send(encoding.toUint8Array(encoder));
+      }
+    };
+    websocket.onerror = event => {
+      provider.emit('connection-error', [event, provider]);
+    };
+    websocket.onclose = event => {
+      provider.emit('connection-close', [event, provider]);
+      provider.ws = null;
+      provider.wsconnecting = false;
+      if (provider.wsconnected) {
+        provider.wsconnected = false;
+        provider.synced = false;
+        // update awareness (all users except local left)
+        awarenessProtocol.removeAwarenessStates(
+          provider.awareness,
+          Array.from(provider.awareness.getStates().keys()).filter(
+            client => client !== provider.doc.clientID
+          ),
+          provider
+        );
+        provider.emit('status', [
+          {
+            status: 'disconnected',
+          },
+        ]);
+      } else {
+        provider.wsUnsuccessfulReconnects++;
+      }
+      // Start with no reconnect timeout and increase timeout by
+      // using exponential backoff starting with 100ms
+      setTimeout(
+        setupWS,
+        math.min(
+          math.pow(2, provider.wsUnsuccessfulReconnects) * 100,
+          provider.maxBackoffTime
+        ),
+        provider
+      );
+    };
+    websocket.onopen = () => {
+      provider.wsLastMessageReceived = time.getUnixTime();
+      provider.wsconnecting = false;
+      provider.wsconnected = true;
+      provider.wsUnsuccessfulReconnects = 0;
+      provider.emit('status', [
+        {
+          status: 'connected',
+        },
+      ]);
+      // always send sync step 1 when connected
+      const encoder = encoding.createEncoder();
+      encoding.writeVarUint(encoder, messageSync);
+      syncProtocol.writeSyncStep1(encoder, provider.doc);
+      websocket.send(encoding.toUint8Array(encoder));
+      // broadcast local awareness state
+      if (provider.awareness.getLocalState() !== null) {
+        const encoderAwarenessState = encoding.createEncoder();
+        encoding.writeVarUint(encoderAwarenessState, messageAwareness);
+        encoding.writeVarUint8Array(
+          encoderAwarenessState,
+          awarenessProtocol.encodeAwarenessUpdate(provider.awareness, [
+            provider.doc.clientID,
+          ])
+        );
+        websocket.send(encoding.toUint8Array(encoderAwarenessState));
+      }
+    };
+
+    provider.emit('status', [
+      {
+        status: 'connecting',
+      },
+    ]);
+  }
+};
+
+/**
+ * @param {WebsocketProvider} provider
+ * @param {ArrayBuffer} buf
+ */
+const broadcastMessage = (provider, buf) => {
+  if (provider.wsconnected) {
+    /** @type {WebSocket} */ (provider.ws).send(buf);
+  }
+  if (provider.bcconnected) {
+    bc.publish(provider.bcChannel, buf, provider);
+  }
+};
+
+/**
+ * Websocket Provider for Yjs. Creates a websocket connection to sync the shared document.
+ * The document name is attached to the provided url. I.e. the following example
+ * creates a websocket connection to http://localhost:1234/my-document-name
+ *
+ * @example
+ *   import * as Y from 'yjs'
+ *   import { WebsocketProvider } from 'y-websocket'
+ *   const doc = new Y.Doc()
+ *   const provider = new WebsocketProvider('http://localhost:1234', 'my-document-name', doc)
+ *
+ * @extends {Observable<string>}
+ */
+export class WebsocketProvider extends Observable {
+  /**
+   * @param {string} serverUrl
+   * @param {string} roomname
+   * @param {Y.Doc} doc
+   * @param {object} [opts]
+   * @param {boolean} [opts.connect]
+   * @param {awarenessProtocol.Awareness} [opts.awareness]
+   * @param {Object<string,string>} [opts.params]
+   * @param {typeof WebSocket} [opts.WebSocketPolyfill] Optionall provide a WebSocket polyfill
+   * @param {number} [opts.resyncInterval] Request server state every `resyncInterval` milliseconds
+   * @param {number} [opts.maxBackoffTime] Maximum amount of time to wait before trying to reconnect (we try to reconnect using exponential backoff)
+   * @param {boolean} [opts.disableBc] Disable cross-tab BroadcastChannel communication
+   */
+  constructor(
+    serverUrl,
+    roomname,
+    doc,
+    {
+      connect = true,
+      awareness = new awarenessProtocol.Awareness(doc),
+      params = {},
+      WebSocketPolyfill = WebSocket,
+      resyncInterval = -1,
+      maxBackoffTime = 2500,
+      disableBc = false,
+    } = {}
+  ) {
+    super();
+    // ensure that url is always ends with /
+    while (serverUrl[serverUrl.length - 1] === '/') {
+      serverUrl = serverUrl.slice(0, serverUrl.length - 1);
+    }
+    const encodedParams = url.encodeQueryParams(params);
+    this.maxBackoffTime = maxBackoffTime;
+    this.bcChannel = serverUrl + '/' + roomname;
+    this.url =
+      serverUrl +
+      '/' +
+      roomname +
+      (encodedParams.length === 0 ? '' : '?' + encodedParams);
+    this.roomname = roomname;
+    this.doc = doc;
+    this._WS = WebSocketPolyfill;
+    this.awareness = awareness;
+    this.wsconnected = false;
+    this.wsconnecting = false;
+    this.bcconnected = false;
+    this.disableBc = disableBc;
+    this.wsUnsuccessfulReconnects = 0;
+    this.messageHandlers = messageHandlers.slice();
+    /**
+     * @type {boolean}
+     */
+    this._synced = false;
+    /**
+     * @type {WebSocket?}
+     */
+    this.ws = null;
+    this.wsLastMessageReceived = 0;
+    /**
+     * Whether to connect to other peers or not
+     * @type {boolean}
+     */
+    this.shouldConnect = connect;
+
+    /**
+     * @type {number}
+     */
+    this._resyncInterval = 0;
+    if (resyncInterval > 0) {
+      this._resyncInterval = /** @type {any} */ (
+        setInterval(() => {
+          if (this.ws && this.ws.readyState === WebSocket.OPEN) {
+            // resend sync step 1
+            const encoder = encoding.createEncoder();
+            encoding.writeVarUint(encoder, messageSync);
+            syncProtocol.writeSyncStep1(encoder, doc);
+            this.ws.send(encoding.toUint8Array(encoder));
+          }
+        }, resyncInterval)
+      );
+    }
+
+    /**
+     * @param {ArrayBuffer} data
+     * @param {any} origin
+     */
+    this._bcSubscriber = (data, origin) => {
+      if (origin !== this) {
+        const encoder = readMessage(this, new Uint8Array(data), false);
+        if (encoding.length(encoder) > 1) {
+          bc.publish(this.bcChannel, encoding.toUint8Array(encoder), this);
+        }
+      }
+    };
+    /**
+     * Listens to Yjs updates and sends them to remote peers (ws and broadcastchannel)
+     * @param {Uint8Array} update
+     * @param {any} origin
+     */
+    this._updateHandler = (update, origin) => {
+      if (origin !== this) {
+        const encoder = encoding.createEncoder();
+        encoding.writeVarUint(encoder, messageSync);
+        syncProtocol.writeUpdate(encoder, update);
+        broadcastMessage(this, encoding.toUint8Array(encoder));
+      }
+    };
+    this.doc.on('update', this._updateHandler);
+    /**
+     * @param {any} changed
+     * @param {any} _origin
+     */
+    this._awarenessUpdateHandler = ({ added, updated, removed }, _origin) => {
+      const changedClients = added.concat(updated).concat(removed);
+      const encoder = encoding.createEncoder();
+      encoding.writeVarUint(encoder, messageAwareness);
+      encoding.writeVarUint8Array(
+        encoder,
+        awarenessProtocol.encodeAwarenessUpdate(awareness, changedClients)
+      );
+      broadcastMessage(this, encoding.toUint8Array(encoder));
+    };
+    this._unloadHandler = () => {
+      awarenessProtocol.removeAwarenessStates(
+        this.awareness,
+        [doc.clientID],
+        'window unload'
+      );
+    };
+    if (typeof window !== 'undefined') {
+      window.addEventListener('unload', this._unloadHandler);
+    } else if (typeof process !== 'undefined') {
+      process.on('exit', this._unloadHandler);
+    }
+    awareness.on('update', this._awarenessUpdateHandler);
+    this._checkInterval = /** @type {any} */ (
+      setInterval(() => {
+        if (
+          this.wsconnected &&
+          messageReconnectTimeout <
+            time.getUnixTime() - this.wsLastMessageReceived
+        ) {
+          // no message received in a long time - not even your own awareness
+          // updates (which are updated every 15 seconds)
+          /** @type {WebSocket} */ (this.ws).close();
+        }
+      }, messageReconnectTimeout / 10)
+    );
+    if (connect) {
+      this.connect();
+    }
+  }
+
+  /**
+   * @type {boolean}
+   */
+  get synced() {
+    return this._synced;
+  }
+
+  set synced(state) {
+    if (this._synced !== state) {
+      this._synced = state;
+      this.emit('synced', [state]);
+      this.emit('sync', [state]);
+    }
+  }
+
+  destroy() {
+    if (this._resyncInterval !== 0) {
+      clearInterval(this._resyncInterval);
+    }
+    clearInterval(this._checkInterval);
+    this.disconnect();
+    if (typeof window !== 'undefined') {
+      window.removeEventListener('unload', this._unloadHandler);
+    } else if (typeof process !== 'undefined') {
+      process.off('exit', this._unloadHandler);
+    }
+    this.awareness.off('update', this._awarenessUpdateHandler);
+    this.doc.off('update', this._updateHandler);
+    super.destroy();
+  }
+
+  connectBc() {
+    if (this.disableBc) {
+      return;
+    }
+    if (!this.bcconnected) {
+      bc.subscribe(this.bcChannel, this._bcSubscriber);
+      this.bcconnected = true;
+    }
+    // send sync step1 to bc
+    // write sync step 1
+    const encoderSync = encoding.createEncoder();
+    encoding.writeVarUint(encoderSync, messageSync);
+    syncProtocol.writeSyncStep1(encoderSync, this.doc);
+    bc.publish(this.bcChannel, encoding.toUint8Array(encoderSync), this);
+    // broadcast local state
+    const encoderState = encoding.createEncoder();
+    encoding.writeVarUint(encoderState, messageSync);
+    syncProtocol.writeSyncStep2(encoderState, this.doc);
+    bc.publish(this.bcChannel, encoding.toUint8Array(encoderState), this);
+    // write queryAwareness
+    const encoderAwarenessQuery = encoding.createEncoder();
+    encoding.writeVarUint(encoderAwarenessQuery, messageQueryAwareness);
+    bc.publish(
+      this.bcChannel,
+      encoding.toUint8Array(encoderAwarenessQuery),
+      this
+    );
+    // broadcast local awareness state
+    const encoderAwarenessState = encoding.createEncoder();
+    encoding.writeVarUint(encoderAwarenessState, messageAwareness);
+    encoding.writeVarUint8Array(
+      encoderAwarenessState,
+      awarenessProtocol.encodeAwarenessUpdate(this.awareness, [
+        this.doc.clientID,
+      ])
+    );
+    bc.publish(
+      this.bcChannel,
+      encoding.toUint8Array(encoderAwarenessState),
+      this
+    );
+  }
+
+  disconnectBc() {
+    // broadcast message with local awareness state set to null (indicating disconnect)
+    const encoder = encoding.createEncoder();
+    encoding.writeVarUint(encoder, messageAwareness);
+    encoding.writeVarUint8Array(
+      encoder,
+      awarenessProtocol.encodeAwarenessUpdate(
+        this.awareness,
+        [this.doc.clientID],
+        new Map()
+      )
+    );
+    broadcastMessage(this, encoding.toUint8Array(encoder));
+    if (this.bcconnected) {
+      bc.unsubscribe(this.bcChannel, this._bcSubscriber);
+      this.bcconnected = false;
+    }
+  }
+
+  disconnect() {
+    this.shouldConnect = false;
+    this.disconnectBc();
+    if (this.ws !== null) {
+      this.ws.close();
+    }
+  }
+
+  connect() {
+    this.shouldConnect = true;
+    if (!this.wsconnected && this.ws === null) {
+      setupWS(this);
+      this.connectBc();
+    }
+  }
+}

packages/data-center/src/datacenter/provider/base.ts

@@ -17,7 +17,12 @@ export class BaseProvider {
     this._config = params.config;
     this._logger = params.logger;
     this._workspace = params.workspace;
-    this._logger.enabled = true;
+    this._logger.enabled = params.debug;
+  }
+
+  async clear() {
+    await this.destroy();
+    await this._config.clear();
   }
 
   async destroy() {

packages/data-center/src/datacenter/provider/index.ts

@@ -7,6 +7,7 @@ export type Logger = ReturnType<typeof getLogger>;
 
 export type InitialParams = {
   config: ConfigStore;
+  debug: boolean;
   logger: Logger;
   workspace: Workspace;
 };

packages/data-center/src/datacenter/provider/local/index.ts

@@ -33,6 +33,12 @@ export class LocalProvider extends BaseProvider {
     this._logger('Local data loaded');
   }
 
+  async clear() {
+    await super.clear();
+    await this._blobs.clear();
+    this._idb?.clearData();
+  }
+
   async destroy(): Promise<void> {
     super.destroy();
     if (this._idb) {

packages/data-center/src/datacenter/store.ts

@@ -1,10 +1,11 @@
-import { createStore, del, get, keys, set } from 'idb-keyval';
+import { createStore, del, get, keys, set, clear } from 'idb-keyval';
 
 export type ConfigStore<T = any> = {
   get: (key: string) => Promise<T | undefined>;
   set: (key: string, value: T) => Promise<void>;
   keys: () => Promise<string[]>;
   delete: (key: string) => Promise<void>;
+  clear: () => Promise<void>;
 };
 
 const initialIndexedDB = <T = any>(database: string): ConfigStore<T> => {
@@ -14,6 +15,7 @@ const initialIndexedDB = <T = any>(database: string): ConfigStore<T> => {
     set: (key: string, value: T) => set(key, value, store),
     keys: () => keys(store),
     delete: (key: string) => del(key, store),
+    clear: () => clear(store),
   };
 };
 
@@ -34,7 +36,16 @@ const scopedIndexedDB = () => {
                 .filter(k => k.startsWith(`${scope}:`))
                 .map(k => k.replace(`${scope}:`, ''))
             ),
-        delete: (key: string) => del(`${scope}:${key}`),
+        delete: (key: string) => idb.delete(`${scope}:${key}`),
+        clear: async () => {
+          await idb
+            .keys()
+            .then(keys =>
+              Promise.all(
+                keys.filter(k => k.startsWith(`${scope}:`)).map(k => del(k))
+              )
+            );
+        },
       };
 
       storeCache.set(scope, store);