From 2a8f18504b5c929f5f44527ff6ece508f65b0615 Mon Sep 17 00:00:00 2001 From: Peng Xiao Date: Tue, 15 Jul 2025 14:45:05 +0800 Subject: [PATCH] fix(core): electron storage sync (#13213) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #### PR Dependency Tree * **PR #13213** 👈 This tree was auto-generated by [Charcoal](https://github.com/danerwilliams/charcoal) ## Summary by CodeRabbit * **New Features** * Added version tracking for global state and cache updates, enabling synchronized updates across multiple windows. * Introduced a unique client identifier to prevent processing self-originated updates. * **Refactor** * Improved event broadcasting for global state and cache changes, ensuring more reliable and efficient update propagation. * **Chores** * Updated internal logic to support structured event formats and revision management for shared storage. --- .../src/main/shared-storage/events.ts | 18 ++----- .../src/main/shared-storage/handlers.ts | 54 ++++++++++++++----- .../electron/src/preload/shared-storage.ts | 28 ++++++++-- 3 files changed, 71 insertions(+), 29 deletions(-) diff --git a/packages/frontend/apps/electron/src/main/shared-storage/events.ts b/packages/frontend/apps/electron/src/main/shared-storage/events.ts index eb0532b87..7676a5212 100644 --- a/packages/frontend/apps/electron/src/main/shared-storage/events.ts +++ b/packages/frontend/apps/electron/src/main/shared-storage/events.ts @@ -1,25 +1,17 @@ import type { MainEventRegister } from '../type'; -import { globalCacheStorage, globalStateStorage } from './storage'; +import { globalCacheUpdates$, globalStateUpdates$ } from './handlers'; export const sharedStorageEvents = { onGlobalStateChanged: ( fn: (state: Record) => void ) => { - const subscription = globalStateStorage.watchAll().subscribe(updates => { - fn(updates); - }); - return () => { - subscription.unsubscribe(); - }; + const subscription = globalStateUpdates$.subscribe(fn); + return () => subscription.unsubscribe(); }, onGlobalCacheChanged: ( fn: (state: Record) => void ) => { - const subscription = globalCacheStorage.watchAll().subscribe(updates => { - fn(updates); - }); - return () => { - subscription.unsubscribe(); - }; + const subscription = globalCacheUpdates$.subscribe(fn); + return () => subscription.unsubscribe(); }, } satisfies Record; diff --git a/packages/frontend/apps/electron/src/main/shared-storage/handlers.ts b/packages/frontend/apps/electron/src/main/shared-storage/handlers.ts index 0c03eb31f..ebc1fdbe3 100644 --- a/packages/frontend/apps/electron/src/main/shared-storage/handlers.ts +++ b/packages/frontend/apps/electron/src/main/shared-storage/handlers.ts @@ -1,6 +1,22 @@ +import { Subject } from 'rxjs'; + import type { NamespaceHandlers } from '../type'; import { globalCacheStorage, globalStateStorage } from './storage'; +// Subjects used by shared-storage/events.ts to broadcast updates to all renderer processes +export const globalStateUpdates$ = new Subject>(); +export const globalCacheUpdates$ = new Subject>(); + +// Revision maps; main generates the next value each time +const globalStateRevisions = new Map(); +const globalCacheRevisions = new Map(); + +function nextRev(revisions: Map, key: string) { + const r = (revisions.get(key) ?? 0) + 1; + revisions.set(key, r); + return r; +} + export const sharedStorageHandlers = { getAllGlobalState: async () => { return globalStateStorage.all(); @@ -8,22 +24,36 @@ export const sharedStorageHandlers = { getAllGlobalCache: async () => { return globalCacheStorage.all(); }, - setGlobalState: async (_, key: string, value: any) => { - return globalStateStorage.set(key, value); + + setGlobalState: async (_e, key: string, value: any, sourceId?: string) => { + const rev = nextRev(globalStateRevisions, key); + globalStateStorage.set(key, value); + globalStateUpdates$.next({ [key]: { v: value, r: rev, s: sourceId } }); }, - delGlobalState: async (_, key: string) => { - return globalStateStorage.del(key); + delGlobalState: async (_e, key: string, sourceId?: string) => { + const rev = nextRev(globalStateRevisions, key); + globalStateStorage.del(key); + globalStateUpdates$.next({ [key]: { v: undefined, r: rev, s: sourceId } }); }, - clearGlobalState: async () => { - return globalStateStorage.clear(); + clearGlobalState: async (_e, sourceId?: string) => { + globalStateRevisions.clear(); + globalStateStorage.clear(); + globalStateUpdates$.next({ '*': { v: undefined, r: 0, s: sourceId } }); }, - setGlobalCache: async (_, key: string, value: any) => { - return globalCacheStorage.set(key, value); + + setGlobalCache: async (_e, key: string, value: any, sourceId?: string) => { + const rev = nextRev(globalCacheRevisions, key); + globalCacheStorage.set(key, value); + globalCacheUpdates$.next({ [key]: { v: value, r: rev, s: sourceId } }); }, - delGlobalCache: async (_, key: string) => { - return globalCacheStorage.del(key); + delGlobalCache: async (_e, key: string, sourceId?: string) => { + const rev = nextRev(globalCacheRevisions, key); + globalCacheStorage.del(key); + globalCacheUpdates$.next({ [key]: { v: undefined, r: rev, s: sourceId } }); }, - clearGlobalCache: async () => { - return globalCacheStorage.clear(); + clearGlobalCache: async (_e, sourceId?: string) => { + globalCacheRevisions.clear(); + globalCacheStorage.clear(); + globalCacheUpdates$.next({ '*': { v: undefined, r: 0, s: sourceId } }); }, } satisfies NamespaceHandlers; diff --git a/packages/frontend/apps/electron/src/preload/shared-storage.ts b/packages/frontend/apps/electron/src/preload/shared-storage.ts index 065781261..0e96daa05 100644 --- a/packages/frontend/apps/electron/src/preload/shared-storage.ts +++ b/packages/frontend/apps/electron/src/preload/shared-storage.ts @@ -6,6 +6,7 @@ import { AFFINE_EVENT_CHANNEL_NAME, } from '../shared/type'; +// Load persisted data from main process synchronously at preload time const initialGlobalState = ipcRenderer.sendSync( AFFINE_API_CHANNEL_NAME, 'sharedStorage:getAllGlobalState' @@ -15,6 +16,9 @@ const initialGlobalCache = ipcRenderer.sendSync( 'sharedStorage:getAllGlobalCache' ); +// Unique id for this renderer instance, used to ignore self-originated broadcasts +const CLIENT_ID: string = Math.random().toString(36).slice(2); + function invokeWithCatch(key: string, ...args: any[]) { ipcRenderer.invoke(AFFINE_API_CHANNEL_NAME, key, ...args).catch(err => { console.error(`Failed to invoke ${key}`, err); @@ -34,7 +38,23 @@ function createSharedStorageApi( memory.setAll(init); ipcRenderer.on(AFFINE_EVENT_CHANNEL_NAME, (_event, channel, updates) => { if (channel === `sharedStorage:${event}`) { - for (const [key, value] of Object.entries(updates)) { + for (const [key, raw] of Object.entries(updates)) { + // support both legacy plain value and new { v, r, s } structure + let value: any; + let source: string | undefined; + + if (raw && typeof raw === 'object' && 'v' in raw) { + value = (raw as any).v; + source = (raw as any).s; + } else { + value = raw; + } + + // Ignore our own broadcasts + if (source && source === CLIENT_ID) { + continue; + } + if (value === undefined) { memory.del(key); } else { @@ -47,11 +67,11 @@ function createSharedStorageApi( return { del(key: string) { memory.del(key); - invokeWithCatch(`sharedStorage:${api.del}`, key); + invokeWithCatch(`sharedStorage:${api.del}`, key, CLIENT_ID); }, clear() { memory.clear(); - invokeWithCatch(`sharedStorage:${api.clear}`); + invokeWithCatch(`sharedStorage:${api.clear}`, CLIENT_ID); }, get(key: string): T | undefined { return memory.get(key); @@ -61,7 +81,7 @@ function createSharedStorageApi( }, set(key: string, value: unknown) { memory.set(key, value); - invokeWithCatch(`sharedStorage:${api.set}`, key, value); + invokeWithCatch(`sharedStorage:${api.set}`, key, value, CLIENT_ID); }, watch(key: string, cb: (i: T | undefined) => void): () => void { const subscription = memory.watch(key).subscribe(i => cb(i as T));