feat: add broad cast channel provider (#1237)

apps/web/.env.local.template

@@ -12,3 +12,4 @@ NODE_API_SERVER=
 # save workspace to idb
 ENABLE_IDB_PROVIDER=1
 PREFETCH_WORKSPACE=1
+ENABLE_BC_PROVIDER=1

apps/web/package.json

@@ -32,6 +32,7 @@
     "react-helmet-async": "^1.3.0",
     "swr": "^2.0.4",
     "y-indexeddb": "^9.0.9",
+    "y-protocols": "^1.0.5",
     "yjs": "^13.5.47",
     "zod": "^3.20.6"
   },

apps/web/preset.config.mjs

@@ -2,5 +2,8 @@ import 'dotenv/config';
 
 export default {
   enableIndexedDBProvider: Boolean(process.env.ENABLE_IDB_PROVIDER ?? '1'),
+  enableBroadCastChannelProvider: Boolean(
+    process.env.ENABLE_BC_PROVIDER ?? '1'
+  ),
   prefetchWorkspace: Boolean(process.env.PREFETCH_WORKSPACE ?? '1'),
 };

apps/web/src/blocksuite/index.ts

@@ -1,6 +1,10 @@
 import { BlockSuiteWorkspace, Provider } from '../shared';
 import { config } from '../shared/env';
-import { createIndexedDBProvider, createWebSocketProvider } from './providers';
+import {
+  createBroadCastChannelProvider,
+  createIndexedDBProvider,
+  createWebSocketProvider,
+} from './providers';
 
 export const createAffineProviders = (
   blockSuiteWorkspace: BlockSuiteWorkspace
@@ -8,6 +12,8 @@ export const createAffineProviders = (
   return (
     [
       createWebSocketProvider(blockSuiteWorkspace),
+      config.enableBroadCastChannelProvider &&
+        createBroadCastChannelProvider(blockSuiteWorkspace),
       config.enableIndexedDBProvider &&
         createIndexedDBProvider(blockSuiteWorkspace),
     ] as any[]
@@ -19,6 +25,8 @@ export const createLocalProviders = (
 ): Provider[] => {
   return (
     [
+      config.enableBroadCastChannelProvider &&
+        createBroadCastChannelProvider(blockSuiteWorkspace),
       config.enableIndexedDBProvider &&
         createIndexedDBProvider(blockSuiteWorkspace),
     ] as any[]

apps/web/src/blocksuite/providers/broad-cast-channel/index.ts

@@ -0,0 +1,91 @@
+import { assertExists } from '@blocksuite/store';
+import {
+  applyAwarenessUpdate,
+  Awareness,
+  encodeAwarenessUpdate,
+} from 'y-protocols/awareness';
+
+import { BlockSuiteWorkspace, BroadCastChannelProvider } from '../../../shared';
+import {
+  BroadcastChannelMessageEvent,
+  getClients,
+  TypedBroadcastChannel,
+} from './type';
+
+export const createBroadCastChannelProvider = (
+  blockSuiteWorkspace: BlockSuiteWorkspace
+): BroadCastChannelProvider => {
+  const Y = BlockSuiteWorkspace.Y;
+  const doc = blockSuiteWorkspace.doc;
+  const awareness = blockSuiteWorkspace.awarenessStore
+    .awareness as unknown as Awareness;
+  let broadcastChannel: TypedBroadcastChannel | null = null;
+  const handleBroadcastChannelMessage = (
+    event: BroadcastChannelMessageEvent
+  ) => {
+    const [eventName] = event.data;
+    switch (eventName) {
+      case 'doc:diff': {
+        const [, diff, clientId] = event.data;
+        const updateV2 = Y.encodeStateAsUpdateV2(doc, diff);
+        broadcastChannel!.postMessage(['doc:update', updateV2, clientId]);
+        break;
+      }
+      case 'doc:update': {
+        const [, updateV2, clientId] = event.data;
+        Y.applyUpdateV2(doc, updateV2, clientId);
+        break;
+      }
+      case 'awareness:query': {
+        const [, clientId] = event.data;
+        const clients = getClients(awareness);
+        const update = encodeAwarenessUpdate(awareness, clients);
+        broadcastChannel!.postMessage(['awareness:update', update, clientId]);
+        break;
+      }
+      case 'awareness:update': {
+        const [, update, clientId] = event.data;
+        applyAwarenessUpdate(awareness, update, clientId);
+        break;
+      }
+    }
+  };
+  return {
+    flavour: 'broadcast-channel',
+    connect: () => {
+      assertExists(blockSuiteWorkspace.room);
+      broadcastChannel = Object.assign(
+        new BroadcastChannel(blockSuiteWorkspace.room),
+        {
+          onmessage: handleBroadcastChannelMessage,
+        }
+      );
+      const docDiff = Y.encodeStateVector(doc);
+      broadcastChannel.postMessage(['doc:diff', docDiff, awareness.clientID]);
+      const docUpdateV2 = Y.encodeStateAsUpdateV2(doc);
+      broadcastChannel.postMessage(['doc:update', docUpdateV2]);
+      broadcastChannel.postMessage(['awareness:query', awareness.clientID]);
+      const awarenessUpdate = encodeAwarenessUpdate(awareness, [
+        awareness.clientID,
+      ]);
+      broadcastChannel.postMessage(['awareness:update', awarenessUpdate]);
+      const handleDocUpdate = (updateV1: Uint8Array, origin: any) => {
+        if (origin !== awareness.clientID) {
+          // not self update, ignore
+          return;
+        }
+        const updateV2 = Y.convertUpdateFormatV1ToV2(updateV1);
+        broadcastChannel?.postMessage(['doc:update', updateV2]);
+      };
+      doc.on('update', handleDocUpdate);
+    },
+    disconnect: () => {
+      assertExists(broadcastChannel);
+      broadcastChannel.close();
+    },
+    cleanup: () => {
+      assertExists(broadcastChannel);
+      broadcastChannel.close();
+    },
+  };
+};

apps/web/src/blocksuite/providers/broad-cast-channel/type.ts

@@ -0,0 +1,81 @@
+import { Awareness as YAwareness } from 'y-protocols/awareness';
+
+export type ClientId = YAwareness['clientID'];
+// eslint-disable-next-line @typescript-eslint/ban-types
+export type DefaultClientData = {};
+
+type EventHandler = (...args: any[]) => void;
+export type DefaultEvents = {
+  [eventName: string]: EventHandler;
+};
+
+type EventNameWithScope<
+  Scope extends string,
+  Type extends string = string
+> = `${Scope}:${Type}`;
+
+type DataScope = 'data';
+type RoomScope = 'room';
+
+type YDocScope = 'doc';
+type AwarenessScope = 'awareness';
+type ObservableScope = YDocScope | AwarenessScope;
+type ObservableEventName = EventNameWithScope<ObservableScope>;
+
+type ValidEventScope = DataScope | RoomScope | ObservableScope;
+
+type ValidateEvents<
+  Events extends DefaultEvents & {
+    [EventName in keyof Events]: EventName extends EventNameWithScope<
+      infer EventScope
+    >
+      ? EventScope extends ValidEventScope
+        ? Events[EventName]
+        : never
+      : Events[EventName];
+  }
+> = Events;
+
+export type DefaultServerToClientEvents<
+  ClientData extends DefaultClientData = DefaultClientData
+> = ValidateEvents<{
+  ['data:update']: (data: ClientData) => void;
+  ['doc:diff']: (diff: ArrayBuffer) => void;
+  ['doc:update']: (update: ArrayBuffer) => void;
+  ['awareness:update']: (update: ArrayBuffer) => void;
+}>;
+
+export type ServerToClientEvents<
+  ClientData extends DefaultClientData = DefaultClientData
+> = DefaultServerToClientEvents<ClientData>;
+
+export type DefaultClientToServerEvents = ValidateEvents<{
+  ['room:close']: () => void;
+  ['doc:diff']: (diff: Uint8Array) => void;
+  ['doc:update']: (update: Uint8Array, callback?: () => void) => void;
+  ['awareness:update']: (update: Uint8Array) => void;
+}>;
+
+export type ClientToServerEvents = DefaultClientToServerEvents;
+
+type ClientToServerEventNames = keyof ClientToServerEvents;
+
+export type BroadcastChannelMessageData<
+  EventName extends ClientToServerEventNames = ClientToServerEventNames
+> =
+  | (EventName extends ObservableEventName
+      ? [eventName: EventName, payload: Uint8Array, clientId?: ClientId]
+      : never)
+  | [eventName: `${AwarenessScope}:query`, clientId: ClientId];
+
+export type BroadcastChannelMessageEvent =
+  MessageEvent<BroadcastChannelMessageData>;
+
+export interface TypedBroadcastChannel extends BroadcastChannel {
+  onmessage: ((event: BroadcastChannelMessageEvent) => void) | null;
+  postMessage: (message: BroadcastChannelMessageData) => void;
+}
+
+export const getClients = (awareness: YAwareness): ClientId[] => [
+  ...awareness.getStates().keys(),
+];

apps/web/src/blocksuite/providers/index.ts

@@ -8,8 +8,9 @@ import {
   LocalIndexedDBProvider,
 } from '../../shared';
 import { apis } from '../../shared/apis';
+import { createBroadCastChannelProvider } from './broad-cast-channel';
 
-export const createWebSocketProvider = (
+const createWebSocketProvider = (
   blockSuiteWorkspace: BlockSuiteWorkspace
 ): AffineWebSocketProvider => {
   let webSocketProvider: WebsocketProvider | null = null;
@@ -31,6 +32,8 @@ export const createWebSocketProvider = (
           params: { token: apis.auth.refresh },
           // @ts-expect-error ignore the type
           awareness: blockSuiteWorkspace.awarenessStore.awareness,
+          // we maintain broadcast channel by ourselves
+          disableBc: true,
         }
       );
       console.log('connect', webSocketProvider.roomname);
@@ -44,7 +47,7 @@ export const createWebSocketProvider = (
   };
 };
 
-export const createIndexedDBProvider = (
+const createIndexedDBProvider = (
   blockSuiteWorkspace: BlockSuiteWorkspace
 ): LocalIndexedDBProvider => {
   let indexdbProvider: IndexeddbPersistence | null = null;
@@ -69,3 +72,9 @@ export const createIndexedDBProvider = (
     },
   };
 };
+
+export {
+  createBroadCastChannelProvider,
+  createIndexedDBProvider,
+  createWebSocketProvider,
+};

apps/web/src/shared/index.ts

@@ -91,6 +91,10 @@ export type BaseProvider = {
   cleanup: () => void;
 };
 
+export interface BroadCastChannelProvider extends BaseProvider {
+  flavour: 'broadcast-channel';
+}
+
 export interface LocalIndexedDBProvider extends BaseProvider {
   flavour: 'local-indexeddb';
 }
@@ -99,7 +103,10 @@ export interface AffineWebSocketProvider extends BaseProvider {
   flavour: 'affine-websocket';
 }
 
-export type Provider = LocalIndexedDBProvider | AffineWebSocketProvider;
+export type Provider =
+  | LocalIndexedDBProvider
+  | AffineWebSocketProvider
+  | BroadCastChannelProvider;
 
 export type AffineRemoteWorkspace =
   | AffineRemoteSyncedWorkspace

apps/web/src/types/index.ts

@@ -8,6 +8,7 @@ export const publicRuntimeConfigSchema = z.object({
   serverAPI: z.string(),
   editorVersion: z.string(),
   enableIndexedDBProvider: z.boolean(),
+  enableBroadCastChannelProvider: z.boolean(),
   prefetchWorkspace: z.boolean(),
 });
 

pnpm-lock.yaml

@@ -182,6 +182,7 @@ importers:
       swr: ^2.0.4
       typescript: ^4.9.5
       y-indexeddb: ^9.0.9
+      y-protocols: ^1.0.5
       yjs: ^13.5.47
       zod: ^3.20.6
     dependencies:
@@ -208,6 +209,7 @@ importers:
       react-helmet-async: 1.3.0_biqbaboplfbrettd7655fr4n2y
       swr: 2.0.4_react@18.2.0
       y-indexeddb: 9.0.9_yjs@13.5.47
+      y-protocols: 1.0.5
       yjs: 13.5.47
       zod: 3.20.6
     devDependencies:

scripts/vitest/next-config-mock.ts

@@ -1,9 +1,14 @@
 export default function getConfig() {
   return {
     publicRuntimeConfig: {
-      serverAPI: 'http://localhost:3000/api',
+      PROJECT_NAME: 'AFFiNE Mock',
-      enableIndexedDBProvider: true,
+      BUILD_DATE: '2021-09-01T00:00:00.000Z',
+      gitVersion: 'UNKNOWN',
+      hash: 'UNKNOWN',
       editorVersion: 'UNKNOWN',
+      serverAPI: 'http://localhost:3000/api',
+      enableBroadCastChannelProvider: true,
+      enableIndexedDBProvider: true,
       prefetchWorkspace: false,
     },
   };