feat(electron): shared storage (#7492)

packages/common/infra/src/storage/memento.ts

@@ -20,6 +20,12 @@ export interface Memento {
 export class MemoryMemento implements Memento {
   private readonly data = new Map<string, LiveData<any>>();
 
+  setAll(init: Record<string, any>) {
+    for (const [key, value] of Object.entries(init)) {
+      this.set(key, value);
+    }
+  }
+
   private getLiveData(key: string): LiveData<any> {
     let data$ = this.data.get(key);
     if (!data$) {
@@ -39,7 +45,9 @@ export class MemoryMemento implements Memento {
     this.getLiveData(key).next(value);
   }
   keys(): string[] {
-    return Array.from(this.data.keys());
+    return Array.from(this.data)
+      .filter(([_, v$]) => v$.value !== undefined)
+      .map(([k]) => k);
   }
   clear(): void {
     this.data.clear();

packages/common/infra/src/sync/indexer/impl/indexeddb/index.ts

@@ -1,7 +1,7 @@
 import type { Observable } from 'rxjs';
 import { from, merge, of, Subject, throttleTime } from 'rxjs';
 
-import { exhaustMapWithTrailing } from '../../../../utils/exhaustmap-with-trailing';
+import { exhaustMapWithTrailing } from '../../../../utils/';
 import {
   type AggregateOptions,
   type AggregateResult,

packages/common/infra/src/sync/job/impl/indexeddb/index.ts

@@ -4,7 +4,7 @@ import { merge, Observable, of, throttleTime } from 'rxjs';
 
 import { fromPromise } from '../../../../livedata';
 import { throwIfAborted } from '../../../../utils';
-import { exhaustMapWithTrailing } from '../../../../utils/exhaustmap-with-trailing';
+import { exhaustMapWithTrailing } from '../../../../utils/';
 import type { Job, JobParams, JobQueue } from '../../';
 
 interface IndexDB extends DBSchema {

packages/common/infra/src/utils/index.ts

@@ -1,5 +1,6 @@
 export * from './async-lock';
 export * from './async-queue';
+export * from './exhaustmap-with-trailing';
 export * from './merge-updates';
 export * from './object-pool';
 export * from './stable-hash';

packages/frontend/core/src/modules/index.ts

@@ -12,7 +12,6 @@ import { configurePermissionsModule } from './permissions';
 import { configureWorkspacePropertiesModule } from './properties';
 import { configureQuickSearchModule } from './quicksearch';
 import { configureShareDocsModule } from './share-doc';
-import { configureStorageImpls } from './storage';
 import { configureTagModule } from './tag';
 import { configureTelemetryModule } from './telemetry';
 import { configureWorkbenchModule } from './workbench';
@@ -35,7 +34,3 @@ export function configureCommonModules(framework: Framework) {
   configureDocsSearchModule(framework);
   configureDocLinksModule(framework);
 }
-
-export function configureImpls(framework: Framework) {
-  configureStorageImpls(framework);
-}

packages/frontend/core/src/modules/storage/impls/electron.ts

@@ -0,0 +1,58 @@
+import { sharedStorage } from '@affine/electron-api';
+import type { GlobalCache, GlobalState } from '@toeverything/infra';
+import { Observable } from 'rxjs';
+
+// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+const ensureSharedStorage = sharedStorage!;
+
+export class ElectronGlobalState implements GlobalState {
+  keys(): string[] {
+    return ensureSharedStorage.globalState.keys();
+  }
+  get<T>(key: string): T | undefined {
+    return ensureSharedStorage.globalState.get(key);
+  }
+  watch<T>(key: string) {
+    return new Observable<T | undefined>(subscriber => {
+      const unsubscribe = ensureSharedStorage.globalState.watch<T>(key, i => {
+        subscriber.next(i);
+      });
+      return () => unsubscribe();
+    });
+  }
+  set<T>(key: string, value: T): void {
+    ensureSharedStorage.globalState.set(key, value);
+  }
+  del(key: string): void {
+    ensureSharedStorage.globalState.del(key);
+  }
+  clear(): void {
+    ensureSharedStorage.globalState.clear();
+  }
+}
+
+export class ElectronGlobalCache implements GlobalCache {
+  keys(): string[] {
+    return ensureSharedStorage.globalCache.keys();
+  }
+  get<T>(key: string): T | undefined {
+    return ensureSharedStorage.globalCache.get(key);
+  }
+  watch<T>(key: string) {
+    return new Observable<T | undefined>(subscriber => {
+      const unsubscribe = ensureSharedStorage.globalCache.watch<T>(key, i => {
+        subscriber.next(i);
+      });
+      return () => unsubscribe();
+    });
+  }
+  set<T>(key: string, value: T): void {
+    ensureSharedStorage.globalCache.set(key, value);
+  }
+  del(key: string): void {
+    ensureSharedStorage.globalCache.del(key);
+  }
+  clear(): void {
+    ensureSharedStorage.globalCache.clear();
+  }
+}

packages/frontend/core/src/modules/storage/index.ts

@@ -1,11 +1,17 @@
 import { type Framework, GlobalCache, GlobalState } from '@toeverything/infra';
 
+import { ElectronGlobalCache, ElectronGlobalState } from './impls/electron';
 import {
   LocalStorageGlobalCache,
   LocalStorageGlobalState,
-} from './impls/storage';
+} from './impls/local-storage';
 
-export function configureStorageImpls(framework: Framework) {
+export function configureLocalStorageStateStorageImpls(framework: Framework) {
   framework.impl(GlobalCache, LocalStorageGlobalCache);
   framework.impl(GlobalState, LocalStorageGlobalState);
 }
+
+export function configureElectronStateStorageImpls(framework: Framework) {
+  framework.impl(GlobalCache, ElectronGlobalCache);
+  framework.impl(GlobalState, ElectronGlobalState);
+}

packages/frontend/electron-api/src/index.ts

@@ -10,6 +10,7 @@ import type {
   affine as exposedAffineGlobal,
   appInfo as exposedAppInfo,
 } from '@affine/electron/preload/electron-api';
+import type { sharedStorage as exposedSharedStorage } from '@affine/electron/preload/shared-storage';
 
 type MainHandlers = typeof mainHandlers;
 type HelperHandlers = typeof helperHandlers;
@@ -39,5 +40,8 @@ export const events = (globalThis as any).events as ClientEvents | null;
 export const affine = (globalThis as any).affine as
   | typeof exposedAffineGlobal
   | null;
+export const sharedStorage = (globalThis as any).sharedStorage as
+  | typeof exposedSharedStorage
+  | null;
 
 export type { UpdateMeta } from '@affine/electron/main/updater/event';

packages/frontend/electron/renderer/app.tsx

@@ -5,7 +5,8 @@ import { NotificationCenter } from '@affine/component';
 import { AffineContext } from '@affine/component/context';
 import { GlobalLoading } from '@affine/component/global-loading';
 import { AppFallback } from '@affine/core/components/affine/app-container';
-import { configureCommonModules, configureImpls } from '@affine/core/modules';
+import { configureCommonModules } from '@affine/core/modules';
+import { configureElectronStateStorageImpls } from '@affine/core/modules/storage';
 import {
   configureBrowserWorkspaceFlavours,
   configureSqliteWorkspaceEngineStorageProvider,
@@ -81,7 +82,7 @@ let languageLoadingPromise: Promise<void> | null = null;
 
 const framework = new Framework();
 configureCommonModules(framework);
-configureImpls(framework);
+configureElectronStateStorageImpls(framework);
 configureBrowserWorkspaceFlavours(framework);
 configureSqliteWorkspaceEngineStorageProvider(framework);
 const frameworkProvider = framework.provider();

packages/frontend/electron/src/main/config-storage/handlers.ts

@@ -1,6 +1,9 @@
 import type { NamespaceHandlers } from '../type';
 import { persistentConfig } from './persist';
 
+/**
+ * @deprecated use shared storage
+ */
 export const configStorageHandlers = {
   get: async () => persistentConfig.get(),
   set: async (_, v) => persistentConfig.set(v),

packages/frontend/electron/src/main/config-storage/persist.ts

@@ -10,6 +10,9 @@ import { app } from 'electron';
 const FILENAME = 'config.json';
 const FILEPATH = path.join(app.getPath('userData'), FILENAME);
 
+/**
+ * @deprecated use shared storage
+ */
 export const persistentConfig = new AppConfigStorage({
   config: defaultAppConfig,
   get: () => JSON.parse(fs.readFileSync(FILEPATH, 'utf-8')),

packages/frontend/electron/src/main/events.ts

@@ -2,6 +2,7 @@ import { app, BrowserWindow } from 'electron';
 
 import { applicationMenuEvents } from './application-menu';
 import { logger } from './logger';
+import { sharedStorageEvents } from './shared-storage';
 import { uiEvents } from './ui/events';
 import { updaterEvents } from './updater/event';
 
@@ -9,6 +10,7 @@ export const allEvents = {
   applicationMenu: applicationMenuEvents,
   updater: updaterEvents,
   ui: uiEvents,
+  sharedStorage: sharedStorageEvents,
 };
 
 function getActiveWindows() {

packages/frontend/electron/src/main/handlers.ts

@@ -5,6 +5,7 @@ import { configStorageHandlers } from './config-storage';
 import { exportHandlers } from './export';
 import { findInPageHandlers } from './find-in-page';
 import { getLogFilePath, logger, revealLogFile } from './logger';
+import { sharedStorageHandlers } from './shared-storage';
 import { uiHandlers } from './ui/handlers';
 import { updaterHandlers } from './updater';
 
@@ -26,6 +27,7 @@ export const allHandlers = {
   updater: updaterHandlers,
   configStorage: configStorageHandlers,
   findInPage: findInPageHandlers,
+  sharedStorage: sharedStorageHandlers,
 };
 
 export const registerHandlers = () => {
@@ -34,7 +36,10 @@ export const registerHandlers = () => {
   for (const [namespace, namespaceHandlers] of Object.entries(allHandlers)) {
     for (const [key, handler] of Object.entries(namespaceHandlers)) {
       const chan = `${namespace}:${key}`;
-      ipcMain.handle(chan, async (e, ...args) => {
+      const wrapper = async (
+        e: Electron.IpcMainInvokeEvent,
+        ...args: any[]
+      ) => {
         const start = performance.now();
         try {
           const result = await handler(e, ...args);
@@ -52,6 +57,18 @@ export const registerHandlers = () => {
         } catch (error) {
           logger.error('[ipc]', chan, error);
         }
+      };
+      // for ipcRenderer.invoke
+      ipcMain.handle(chan, wrapper);
+      // for ipcRenderer.sendSync
+      ipcMain.on(chan, (e, ...args) => {
+        wrapper(e, ...args)
+          .then(ret => {
+            e.returnValue = ret;
+          })
+          .catch(() => {
+            // never throw
+          });
       });
     }
   }

packages/frontend/electron/src/main/shared-storage/events.ts

@@ -0,0 +1,25 @@
+import type { MainEventRegister } from '../type';
+import { globalCacheStorage, globalStateStorage } from './storage';
+
+export const sharedStorageEvents = {
+  onGlobalStateChanged: (
+    fn: (state: Record<string, unknown | undefined>) => void
+  ) => {
+    const subscription = globalStateStorage.watchAll().subscribe(updates => {
+      fn(updates);
+    });
+    return () => {
+      subscription.unsubscribe();
+    };
+  },
+  onGlobalCacheChanged: (
+    fn: (state: Record<string, unknown | undefined>) => void
+  ) => {
+    const subscription = globalCacheStorage.watchAll().subscribe(updates => {
+      fn(updates);
+    });
+    return () => {
+      subscription.unsubscribe();
+    };
+  },
+} satisfies Record<string, MainEventRegister>;

packages/frontend/electron/src/main/shared-storage/handlers.ts

@@ -0,0 +1,29 @@
+import type { NamespaceHandlers } from '../type';
+import { globalCacheStorage, globalStateStorage } from './storage';
+
+export const sharedStorageHandlers = {
+  getAllGlobalState: async () => {
+    return globalStateStorage.all();
+  },
+  getAllGlobalCache: async () => {
+    return globalCacheStorage.all();
+  },
+  setGlobalState: async (_, key: string, value: any) => {
+    return globalStateStorage.set(key, value);
+  },
+  delGlobalState: async (_, key: string) => {
+    return globalStateStorage.del(key);
+  },
+  clearGlobalState: async () => {
+    return globalStateStorage.clear();
+  },
+  setGlobalCache: async (_, key: string, value: any) => {
+    return globalCacheStorage.set(key, value);
+  },
+  delGlobalCache: async (_, key: string) => {
+    return globalCacheStorage.del(key);
+  },
+  clearGlobalCache: async () => {
+    return globalCacheStorage.clear();
+  },
+} satisfies NamespaceHandlers;

packages/frontend/electron/src/main/shared-storage/index.ts

@@ -0,0 +1,2 @@
+export { sharedStorageEvents } from './events';
+export { sharedStorageHandlers } from './handlers';

packages/frontend/electron/src/main/shared-storage/json-file.ts

@@ -0,0 +1,139 @@
+import fs from 'node:fs';
+
+import type { Memento } from '@toeverything/infra';
+import {
+  backoffRetry,
+  effect,
+  exhaustMapWithTrailing,
+  fromPromise,
+} from '@toeverything/infra';
+import { debounceTime, EMPTY, mergeMap, Observable, timeout } from 'rxjs';
+
+import { logger } from '../logger';
+
+export class PersistentJSONFileStorage implements Memento {
+  data: Record<string, any> = {};
+  subscriptions: Map<string, Set<(p: any) => void>> = new Map();
+  subscriptionAll: Set<(p: Record<string, any>) => void> = new Set();
+
+  constructor(readonly filepath: string) {
+    try {
+      this.data = JSON.parse(fs.readFileSync(filepath, 'utf-8'));
+    } catch (err) {
+      // ignore ENOENT error
+      if (
+        !(
+          err &&
+          typeof err === 'object' &&
+          'code' in err &&
+          err.code === 'ENOENT'
+        )
+      ) {
+        logger.error('failed to load file', err);
+      }
+    }
+  }
+
+  get<T>(key: string): T | undefined {
+    return this.data[key];
+  }
+  all(): Record<string, any> {
+    return this.data;
+  }
+  watch<T>(key: string): Observable<T | undefined> {
+    const subs = this.subscriptions.get(key) || new Set();
+    this.subscriptions.set(key, subs);
+    return new Observable<T | undefined>(subscriber => {
+      const sub = (p: any) => subscriber.next(p);
+      subs.add(sub);
+      return () => {
+        subs.delete(sub);
+      };
+    });
+  }
+  watchAll(): Observable<Record<string, unknown | undefined>> {
+    return new Observable<Record<string, unknown | undefined>>(subscriber => {
+      const sub = (p: Record<string, unknown | undefined>) =>
+        subscriber.next(p);
+      this.subscriptionAll.add(sub);
+      return () => {
+        this.subscriptionAll.delete(sub);
+      };
+    });
+  }
+  set<T>(key: string, value: T): void {
+    this.data[key] = value;
+    const subs = this.subscriptions.get(key) || new Set();
+    for (const sub of subs) {
+      sub(value);
+    }
+    for (const sub of this.subscriptionAll) {
+      sub({
+        [key]: this.data[key],
+      });
+    }
+    this.save();
+  }
+
+  del(key: string): void {
+    delete this.data[key];
+    const subs = this.subscriptions.get(key) || new Set();
+    for (const sub of subs) {
+      sub(undefined);
+    }
+    for (const sub of this.subscriptionAll) {
+      sub({
+        [key]: undefined,
+      });
+    }
+    this.save();
+  }
+  clear(): void {
+    const oldData = this.data;
+    this.data = {};
+    for (const [_, subs] of this.subscriptions) {
+      for (const sub of subs) {
+        sub(undefined);
+      }
+    }
+    for (const sub of this.subscriptionAll) {
+      sub(
+        Object.fromEntries(
+          Object.entries(oldData).map(([key]) => [key, undefined])
+        )
+      );
+    }
+    this.save();
+  }
+
+  keys(): string[] {
+    return Object.keys(this.data);
+  }
+
+  save = effect(
+    debounceTime(1000),
+    exhaustMapWithTrailing(() => {
+      return fromPromise(async () => {
+        try {
+          await fs.promises.writeFile(
+            this.filepath,
+            JSON.stringify(this.data),
+            'utf-8'
+          );
+        } catch (err) {
+          logger.error(`failed to save file, ${this.filepath}`, err);
+        }
+      }).pipe(
+        timeout(5000),
+        backoffRetry({
+          count: Infinity,
+        }),
+        mergeMap(() => EMPTY)
+      );
+    })
+  );
+
+  dispose() {
+    this.save.unsubscribe();
+  }
+}

packages/frontend/electron/src/main/shared-storage/storage.ts

@@ -0,0 +1,13 @@
+import path from 'node:path';
+
+import { app } from 'electron';
+
+import { PersistentJSONFileStorage } from './json-file';
+
+export const globalStateStorage = new PersistentJSONFileStorage(
+  path.join(app.getPath('userData'), 'global-state.json')
+);
+
+export const globalCacheStorage = new PersistentJSONFileStorage(
+  path.join(app.getPath('userData'), 'global-cache.json')
+);

packages/frontend/electron/src/preload/bootstrap.ts

@@ -1,12 +1,14 @@
 import { contextBridge } from 'electron';
 
 import { affine, appInfo, getElectronAPIs } from './electron-api';
+import { sharedStorage } from './shared-storage';
 
 const { apis, events } = getElectronAPIs();
 
 contextBridge.exposeInMainWorld('appInfo', appInfo);
 contextBridge.exposeInMainWorld('apis', apis);
 contextBridge.exposeInMainWorld('events', events);
+contextBridge.exposeInMainWorld('sharedStorage', sharedStorage);
 
 try {
   contextBridge.exposeInMainWorld('affine', affine);

packages/frontend/electron/src/preload/shared-storage.ts

@@ -0,0 +1,87 @@
+import { MemoryMemento } from '@toeverything/infra';
+import { ipcRenderer } from 'electron';
+
+const initialGlobalState = ipcRenderer.sendSync(
+  'sharedStorage:getAllGlobalState'
+);
+const initialGlobalCache = ipcRenderer.sendSync(
+  'sharedStorage:getAllGlobalCache'
+);
+
+function invokeWithCatch(key: string, ...args: any[]) {
+  ipcRenderer.invoke(key, ...args).catch(err => {
+    console.error(`Failed to invoke ${key}`, err);
+  });
+}
+
+function createSharedStorageApi(
+  init: Record<string, any>,
+  event: string,
+  api: {
+    del: string;
+    clear: string;
+    set: string;
+  }
+) {
+  const memory = new MemoryMemento();
+  memory.setAll(init);
+  ipcRenderer.on(`sharedStorage:${event}`, (_event, updates) => {
+    for (const [key, value] of Object.entries(updates)) {
+      if (value === undefined) {
+        memory.del(key);
+      } else {
+        memory.set(key, value);
+      }
+    }
+  });
+
+  return {
+    del(key: string) {
+      memory.del(key);
+      invokeWithCatch(`sharedStorage:${api.del}`, key);
+    },
+    clear() {
+      memory.clear();
+      invokeWithCatch(`sharedStorage:${api.clear}`);
+    },
+    get<T>(key: string): T | undefined {
+      return memory.get(key);
+    },
+    keys() {
+      return memory.keys();
+    },
+    set(key: string, value: unknown) {
+      memory.set(key, value);
+      invokeWithCatch(`sharedStorage:${api.set}`, key, value);
+    },
+    watch<T>(key: string, cb: (i: T | undefined) => void): () => void {
+      const subscription = memory.watch(key).subscribe(i => cb(i as T));
+      return () => subscription.unsubscribe();
+    },
+  };
+}
+
+export const globalState = createSharedStorageApi(
+  initialGlobalState,
+  'onGlobalStateChanged',
+  {
+    clear: 'clearGlobalState',
+    del: 'delGlobalState',
+    set: 'setGlobalState',
+  }
+);
+
+export const globalCache = createSharedStorageApi(
+  initialGlobalCache,
+  'onGlobalCacheChanged',
+  {
+    clear: 'clearGlobalCache',
+    del: 'delGlobalCache',
+    set: 'setGlobalCache',
+  }
+);
+
+export const sharedStorage = {
+  globalState,
+  globalCache,
+};

packages/frontend/web/src/app.tsx

@@ -5,7 +5,8 @@ import { NotificationCenter } from '@affine/component';
 import { AffineContext } from '@affine/component/context';
 import { GlobalLoading } from '@affine/component/global-loading';
 import { AppFallback } from '@affine/core/components/affine/app-container';
-import { configureCommonModules, configureImpls } from '@affine/core/modules';
+import { configureCommonModules } from '@affine/core/modules';
+import { configureLocalStorageStateStorageImpls } from '@affine/core/modules/storage';
 import {
   configureBrowserWorkspaceFlavours,
   configureIndexedDBWorkspaceEngineStorageProvider,
@@ -69,7 +70,7 @@ let languageLoadingPromise: Promise<void> | null = null;
 
 const framework = new Framework();
 configureCommonModules(framework);
-configureImpls(framework);
+configureLocalStorageStateStorageImpls(framework);
 configureBrowserWorkspaceFlavours(framework);
 configureIndexedDBWorkspaceEngineStorageProvider(framework);
 const frameworkProvider = framework.provider();