feat: add channel

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

@@ -7,33 +7,47 @@ import type {
 import type { User } from '../../types';
 import { Workspace as BlocksuiteWorkspace } from '@blocksuite/store';
 import { BlockSchema } from '@blocksuite/blocks/models';
-import { applyUpdate, encodeStateAsUpdate } from 'yjs';
 import { storage } from './storage.js';
 import assert from 'assert';
 import { WebsocketProvider } from './sync.js';
 // import { IndexedDBProvider } from '../local/indexeddb';
-import { getApis, Member } from './apis/index.js';
+import { getApis } from './apis/index.js';
 import type { Apis, WorkspaceDetail, Callback } from './apis';
 import { setDefaultAvatar } from '../utils.js';
 import { MessageCode } from '../../message';
 import { token } from './apis/token.js';
+import { WebsocketClient } from './channel';
 
 export interface AffineProviderConstructorParams
   extends ProviderConstructorParams {
   apis?: Apis;
 }
 
+const {
+  Y: { applyUpdate, encodeStateAsUpdate },
+} = BlocksuiteWorkspace;
+
 export class AffineProvider extends BaseProvider {
   public id = 'affine';
   private _workspacesCache: Map<string, BlocksuiteWorkspace> = new Map();
   private _onTokenRefresh?: Callback = undefined;
   private _wsMap: Map<string, WebsocketProvider> = new Map();
   private _apis: Apis;
+  private _channel: WebsocketClient;
   // private _idbMap: Map<string, IndexedDBProvider> = new Map();
 
   constructor({ apis, ...params }: AffineProviderConstructorParams) {
     super(params);
     this._apis = apis || getApis();
+    this._channel = new WebsocketClient(
+      `${window.location.protocol === 'https:' ? 'wss' : 'ws'}://${
+        window.location.host
+      }/global/sync/`,
+      this._logger
+    );
+    if (token.isLogin) {
+      this._connectChannel();
+    }
   }
 
   override async init() {
@@ -64,6 +78,15 @@ export class AffineProvider extends BaseProvider {
     }
   }
 
+  private _connectChannel() {
+    this._channel.connect();
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    this._channel.on('message', (message: any) => {
+      console.log('message', message);
+    });
+  }
+
   private _getWebsocketProvider(workspace: BlocksuiteWorkspace) {
     const { doc, room } = workspace;
     assert(room);
@@ -206,6 +229,9 @@ export class AffineProvider extends BaseProvider {
       }
     }
     const user = await this._apis.signInWithGoogle?.();
+    if (!this._channel.connected) {
+      this._connectChannel();
+    }
     if (!user) {
       this._messageCenter.send(MessageCode.loginError);
     }
@@ -363,6 +389,8 @@ export class AffineProvider extends BaseProvider {
 
   public override async logout(): Promise<void> {
     token.clear();
+    this._channel.disconnect();
+    this._wsMap.forEach(ws => ws.disconnect());
     storage.removeItem('token');
   }
 

packages/data-center/src/provider/affine/channel.ts

@@ -0,0 +1,53 @@
+import websocket from 'lib0/websocket';
+import { Logger } from 'src/types';
+import { token } from './apis/token';
+
+const RECONNECT_INTERVAL_TIME = 5000;
+const MAX_RECONNECT_TIMES = 50;
+
+export class WebsocketClient extends websocket.WebsocketClient {
+  public shouldReconnect = false;
+  private _reconnectInterval: number | null = null;
+  private _logger: Logger;
+  constructor(
+    url: string,
+    logger: Logger,
+    options?: { binaryType: 'arraybuffer' | 'blob' | null }
+  ) {
+    super(url, options);
+    this._logger = logger;
+    this._setupChannel();
+  }
+
+  private _setupChannel() {
+    this.on('connect', () => {
+      this._logger('Affine channel connected');
+      this.shouldReconnect = true;
+      if (this._reconnectInterval) {
+        window.clearInterval(this._reconnectInterval);
+      }
+    });
+
+    this.on('disconnect', ({ error }: { error: Error }) => {
+      if (error) {
+        let times = 0;
+        // Try reconnect if connect error has occurred
+        this._reconnectInterval = window.setInterval(() => {
+          if (this.shouldReconnect && token.isLogin && !this.connected) {
+            try {
+              this.connect();
+              this._logger(`try reconnect channel ${++times} times`);
+              if (times > MAX_RECONNECT_TIMES) {
+                this._logger('reconnect failed, max reconnect times reached');
+                this._reconnectInterval &&
+                  window.clearInterval(this._reconnectInterval);
+              }
+            } catch (e) {
+              this._logger('reconnect failed', e);
+            }
+          }
+        }, RECONNECT_INTERVAL_TIME);
+      }
+    });
+  }
+}

packages/data-center/src/provider/local/indexeddb/indexeddb.ts

@@ -1,14 +1,19 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import * as idb from 'lib0/indexeddb.js';
 import { Observable } from 'lib0/observable.js';
-import type { Doc } from 'yjs';
-import { applyUpdate, encodeStateAsUpdate, transact } from 'yjs';
+import { Workspace as BlocksuiteWorkspace } from '@blocksuite/store';
 
 const customStoreName = 'custom';
 const updatesStoreName = 'updates';
 
 const PREFERRED_TRIM_SIZE = 500;
 
+const {
+  Y: { applyUpdate, transact, encodeStateAsUpdate },
+} = BlocksuiteWorkspace;
+
+type Doc = Parameters<typeof transact>[0];
+
 const fetchUpdates = async (provider: IndexedDBProvider) => {
   const [updatesStore] = idb.transact(provider.db as IDBDatabase, [
     updatesStoreName,