Merge branch 'feat/cloud-sync-saika' of github.com:toeverything/AFFiNE into feat/datacenter-dev

packages/data-center/src/datacenter.ts

@@ -122,8 +122,8 @@ export class DataCenter {
    * @param {string} workspaceId workspace id
    */
   private _getBlocksuiteWorkspace(workspaceId: string) {
-    const workspaceInfo = this._workspaceUnitCollection.find(workspaceId);
+    // const workspaceInfo = this._workspaceUnitCollection.find(workspaceId);
-    assert(workspaceInfo, 'Workspace not found');
+    // assert(workspaceInfo, 'Workspace not found');
     return (
       // this._workspaceInstances.get(workspaceId) ||
       createBlocksuiteWorkspace(workspaceId)
@@ -176,22 +176,21 @@ export class DataCenter {
   }
 
   public async loadPublicWorkspace(workspaceId: string) {
-    const workspaceUnit = this._workspaceUnitCollection.find(workspaceId);
+    // FIXME: hard code for public workspace
-    assert(workspaceUnit, 'Workspace not found');
+    const provider = this.providerMap.get('affine');
-    const provider = this.providerMap.get(workspaceUnit.provider);
     assert(provider);
     const blocksuiteWorkspace = this._getBlocksuiteWorkspace(workspaceId);
     await provider.loadPublicWorkspace(blocksuiteWorkspace);
 
     const workspaceUnitForPublic = new WorkspaceUnit({
-      id: workspaceUnit.id,
+      id: workspaceId,
-      name: workspaceUnit.name,
+      name: blocksuiteWorkspace.meta.name,
-      avatar: workspaceUnit.avatar,
+      avatar: blocksuiteWorkspace.meta.avatar,
-      owner: workspaceUnit.owner,
+      owner: undefined,
-      published: workspaceUnit.published,
+      published: true,
-      provider: workspaceUnit.provider,
+      provider: 'affine',
-      memberCount: workspaceUnit.memberCount,
+      memberCount: 1,
-      syncMode: workspaceUnit.syncMode,
+      syncMode: 'core',
     });
 
     workspaceUnitForPublic.setBlocksuiteWorkspace(blocksuiteWorkspace);

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

@@ -11,12 +11,13 @@ import { storage } from './storage.js';
 import assert from 'assert';
 import { WebsocketProvider } from './sync.js';
 // import { IndexedDBProvider } from '../local/indexeddb';
-import { getApis } from './apis/index.js';
+import { getApis, Workspace } 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';
+import { SyncMode } from '../../workspace-unit';
 
 export interface AffineProviderConstructorParams
   extends ProviderConstructorParams {
@@ -33,21 +34,12 @@ export class AffineProvider extends BaseProvider {
   private _onTokenRefresh?: Callback = undefined;
   private _wsMap: Map<string, WebsocketProvider> = new Map();
   private _apis: Apis;
-  private _channel: WebsocketClient;
+  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() {
@@ -76,14 +68,59 @@ export class AffineProvider extends BaseProvider {
     } else {
       storage.setItem('token', this._apis.token.refresh);
     }
+
+    if (token.isLogin) {
+      this._connectChannel();
+    }
   }
 
   private _connectChannel() {
-    this._channel.connect();
+    if (!this._channel) {
+      this._channel = new WebsocketClient(
+        `${window.location.protocol === 'https:' ? 'wss' : 'ws'}://${
+          window.location.host
+        }/api/global/sync/`,
+        this._logger,
+        {
+          params: {
+            token: this._apis.token.refresh,
+          },
+        }
+      );
+    }
+    this._channel.on('message', this.handlerAffineListMessage);
+  }
 
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  private handlerAffineListMessage({
-    this._channel.on('message', (message: any) => {
+    ws_details,
-      console.log('message', message);
+    metadata,
+  }: {
+    ws_list: Workspace[];
+    ws_details: Record<string, WorkspaceDetail>;
+    metadata: Record<string, { avatar: string; name: string }>;
+  }) {
+    Object.entries(ws_details).forEach(([id, detail]) => {
+      const { name, avatar } = metadata[id];
+      assert(name);
+      const workspace = {
+        name: name,
+        avatar,
+        owner: {
+          name: detail.owner.name,
+          id: detail.owner.id,
+          email: detail.owner.email,
+          avatar: detail.owner.avatar_url,
+        },
+        published: detail.public,
+        memberCount: detail.member_count,
+        provider: 'affine',
+        syncMode: 'core' as SyncMode,
+      };
+      if (this._workspaces.get(id)) {
+        this._workspaces.update(id, workspace);
+      } else {
+        this._workspaces.add({ id, ...workspace });
+      }
     });
   }
 
@@ -104,10 +141,13 @@ export class AffineProvider extends BaseProvider {
     return ws;
   }
 
-  private async _applyCloudUpdates(blocksuiteWorkspace: BlocksuiteWorkspace) {
+  private async _applyCloudUpdates(
+    blocksuiteWorkspace: BlocksuiteWorkspace,
+    published = false
+  ) {
     const { doc, room: workspaceId } = blocksuiteWorkspace;
     assert(workspaceId, 'Blocksuite Workspace without room(workspaceId).');
-    const updates = await this._apis.downloadWorkspace(workspaceId);
+    const updates = await this._apis.downloadWorkspace(workspaceId, published);
     if (updates && updates.byteLength) {
       await new Promise(resolve => {
         doc.once('update', resolve);
@@ -117,7 +157,7 @@ export class AffineProvider extends BaseProvider {
   }
 
   override async loadPublicWorkspace(blocksuiteWorkspace: BlocksuiteWorkspace) {
-    await this._applyCloudUpdates(blocksuiteWorkspace);
+    await this._applyCloudUpdates(blocksuiteWorkspace, true);
     return blocksuiteWorkspace;
   }
 
@@ -235,7 +275,7 @@ export class AffineProvider extends BaseProvider {
       }
     }
     const user = await this._apis.signInWithGoogle?.();
-    if (!this._channel.connected) {
+    if (!this._channel?.connected) {
       this._connectChannel();
     }
     if (!user) {
@@ -365,13 +405,13 @@ export class AffineProvider extends BaseProvider {
     workspace_id: string,
     email: string
   ): Promise<User | null> {
-    const user = await this._apis.getUserByEmail({ workspace_id, email });
+    const users = await this._apis.getUserByEmail({ workspace_id, email });
-    return user
+    return users?.length
       ? {
-          id: user.id,
+          id: users[0].id,
-          name: user.name,
+          name: users[0].name,
-          avatar: user.avatar_url,
+          avatar: users[0].avatar_url,
-          email: user.email,
+          email: users[0].email,
         }
       : null;
   }
@@ -396,7 +436,7 @@ export class AffineProvider extends BaseProvider {
 
   public override async logout(): Promise<void> {
     token.clear();
-    this._channel.disconnect();
+    this._channel?.disconnect();
     this._wsMap.forEach(ws => ws.disconnect());
     storage.removeItem('token');
   }

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

@@ -15,7 +15,7 @@ export interface User {
 
 export async function getUserByEmail(
   params: GetUserByEmailParams
-): Promise<User | null> {
+): Promise<User[] | null> {
   const searchParams = new URLSearchParams({ ...params });
-  return client.get('api/user', { searchParams }).json<User | null>();
+  return client.get('api/user', { searchParams }).json<User[] | null>();
 }

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

@@ -172,7 +172,11 @@ export async function leaveWorkspace({ id }: LeaveWorkspaceParams) {
 }
 
 export async function downloadWorkspace(
-  workspaceId: string
+  workspaceId: string,
+  published = false
 ): Promise<ArrayBuffer> {
+  if (published) {
+    return bareClient.get(`api/workspace/${workspaceId}/doc`).arrayBuffer();
+  }
   return client.get(`api/workspace/${workspaceId}/doc`).arrayBuffer();
 }

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

@@ -1,6 +1,7 @@
 import * as websocket from 'lib0/websocket';
 import { Logger } from 'src/types';
 import { token } from './apis/token';
+import * as url from 'lib0/url';
 
 const RECONNECT_INTERVAL_TIME = 5000;
 const MAX_RECONNECT_TIMES = 50;
@@ -10,11 +11,21 @@ export class WebsocketClient extends websocket.WebsocketClient {
   private _reconnectInterval: number | null = null;
   private _logger: Logger;
   constructor(
-    url: string,
+    serverUrl: string,
     logger: Logger,
-    options?: { binaryType: 'arraybuffer' | 'blob' | null }
+    options?: ConstructorParameters<typeof websocket.WebsocketClient>[1] & {
+      params: Record<string, string>;
+    }
   ) {
-    super(url, options);
+    const params = options?.params || {};
+    // ensure that url is always ends with /
+    while (serverUrl[serverUrl.length - 1] === '/') {
+      serverUrl = serverUrl.slice(0, serverUrl.length - 1);
+    }
+    const encodedParams = url.encodeQueryParams(params);
+    const newUrl =
+      serverUrl + '/' + (encodedParams.length === 0 ? '' : '?' + encodedParams);
+    super(newUrl, options);
     this._logger = logger;
     this._setupChannel();
   }

packages/data-center/src/workspace-unit-collection.ts

@@ -61,7 +61,8 @@ export class WorkspaceUnitCollection {
 
     const add = (workspace: WorkspaceUnitCtorParams) => {
       if (this._workspaceUnitMap.has(workspace.id)) {
-        throw new Error(`Duplicate workspace id.`);
+        // FIXME: multiple add same workspace
+        return;
       }
 
       const workspaceUnit = new WorkspaceUnit(workspace);