feat: init disk remote source

2026-07-24 05:48:59 +08:00 · 2026-02-27 02:39:53 +08:00
parent 895e774569
commit 6557e5d01d
51 changed files with 6793 additions and 158 deletions
@@ -0,0 +1,169 @@
+import fs from 'node:fs';
+import path from 'node:path';
+
+import type { DiskSyncEvent as NativeDiskSyncEvent } from '@affine/native';
+import { DiskSync } from '@affine/native';
+import type { DocClock, DocUpdate } from '@affine/nbstore';
+import type { DiskSessionOptions, DiskSyncEvent } from '@affine/nbstore/disk';
+
+import { diskSyncSubjects } from './subjects';
+
+interface DiskSyncSubscriber {
+  unsubscribe(): Promise<void | Error> | void | Error;
+}
+
+type NapiMaybe<T> = T | Error;
+
+function unwrapNapiResult<T>(result: NapiMaybe<T>, action: string): T {
+  if (result instanceof Error) {
+    throw new Error(`[disk] ${action} failed: ${result.message}`);
+  }
+  return result;
+}
+
+function normalizeTimestamp(timestamp: unknown): Date | null {
+  const normalized =
+    timestamp instanceof Date ? timestamp : new Date(timestamp as string);
+  if (Number.isNaN(normalized.getTime())) {
+    return null;
+  }
+  return normalized;
+}
+
+function normalizeDiskSyncEvent(
+  event: NativeDiskSyncEvent
+): DiskSyncEvent | null {
+  switch (event.type) {
+    case 'ready':
+      return { type: 'ready' };
+    case 'doc-update': {
+      if (!event.update || !(event.update.bin instanceof Uint8Array)) {
+        return null;
+      }
+      const timestamp = normalizeTimestamp(event.update.timestamp);
+      if (!timestamp) {
+        return null;
+      }
+      return {
+        type: 'doc-update',
+        update: {
+          docId: event.update.docId,
+          bin: event.update.bin,
+          timestamp,
+          editor: event.update.editor,
+        },
+        origin: event.origin,
+      };
+    }
+    case 'doc-delete': {
+      if (typeof event.docId !== 'string') {
+        return null;
+      }
+      const timestamp = normalizeTimestamp(event.timestamp);
+      if (!timestamp) {
+        return null;
+      }
+      return {
+        type: 'doc-delete',
+        docId: event.docId,
+        timestamp,
+      };
+    }
+    case 'error': {
+      if (typeof event.message !== 'string') {
+        return null;
+      }
+      return {
+        type: 'error',
+        message: event.message,
+      };
+    }
+    default:
+      return null;
+  }
+}
+
+type DiskSyncRuntime = InstanceType<typeof DiskSync> & {
+  startSession(
+    sessionId: string,
+    options: DiskSessionOptions
+  ): Promise<NapiMaybe<void>>;
+  stopSession(sessionId: string): Promise<NapiMaybe<void>>;
+  applyLocalUpdate(
+    sessionId: string,
+    update: DocUpdate,
+    origin?: string
+  ): Promise<NapiMaybe<DocClock>>;
+  subscribeEvents(
+    sessionId: string,
+    callback: (err: Error | null, event: NativeDiskSyncEvent) => void
+  ): Promise<NapiMaybe<DiskSyncSubscriber>>;
+};
+
+const diskSync = new DiskSync() as DiskSyncRuntime;
+const subscriptions = new Map<string, () => Promise<void>>();
+
+function e2eLog(options: DiskSessionOptions, line: string) {
+  if (process.env.AFFINE_E2E !== '1') {
+    return;
+  }
+  try {
+    const p = path.join(options.syncFolder, '.disk-e2e.log');
+    fs.appendFileSync(p, `${new Date().toISOString()}\t${line}\n`, 'utf8');
+  } catch {
+    // ignore
+  }
+}
+
+export async function startSession(
+  sessionId: string,
+  options: DiskSessionOptions
+): Promise<void> {
+  e2eLog(
+    options,
+    `startSession\t${sessionId}\tworkspaceId=${options.workspaceId}\tsyncFolder=${options.syncFolder}`
+  );
+  unwrapNapiResult(
+    await diskSync.startSession(sessionId, options),
+    'startSession'
+  );
+
+  if (subscriptions.has(sessionId)) {
+    return;
+  }
+
+  const subscriber = unwrapNapiResult(
+    await diskSync.subscribeEvents(sessionId, (err, event) => {
+      if (err) {
+        return;
+      }
+      const normalizedEvent = normalizeDiskSyncEvent(event);
+      if (!normalizedEvent) {
+        return;
+      }
+      diskSyncSubjects.event$.next({ sessionId, event: normalizedEvent });
+    }),
+    'subscribeEvents'
+  );
+  subscriptions.set(sessionId, async () => {
+    unwrapNapiResult(await subscriber.unsubscribe(), 'unsubscribe');
+  });
+}
+
+export async function stopSession(sessionId: string): Promise<void> {
+  await subscriptions.get(sessionId)?.();
+  subscriptions.delete(sessionId);
+  unwrapNapiResult(await diskSync.stopSession(sessionId), 'stopSession');
+}
+
+export async function applyLocalUpdate(
+  sessionId: string,
+  update: DocUpdate,
+  origin?: string
+): Promise<DocClock> {
+  // syncFolder isn't directly available here; we log per session start only.
+  return unwrapNapiResult(
+    await diskSync.applyLocalUpdate(sessionId, update, origin),
+    'applyLocalUpdate'
+  );
+}
@@ -0,0 +1,20 @@
+import type { MainEventRegister } from '../type';
+import { applyLocalUpdate, startSession, stopSession } from './handlers';
+import { diskSyncSubjects } from './subjects';
+
+export const diskSyncHandlers = {
+  startSession,
+  stopSession,
+  applyLocalUpdate,
+};
+
+export const diskSyncEvents = {
+  onEvent: ((callback: (...args: any[]) => void) => {
+    const subscription = diskSyncSubjects.event$.subscribe(payload => {
+      callback(payload);
+    });
+    return () => {
+      subscription.unsubscribe();
+    };
+  }) satisfies MainEventRegister,
+};
@@ -0,0 +1,11 @@
+import type { DiskSyncEvent } from '@affine/nbstore/disk';
+import { Subject } from 'rxjs';
+
+export interface DiskSyncSessionEvent {
+  sessionId: string;
+  event: DiskSyncEvent;
+}
+
+export const diskSyncSubjects = {
+  event$: new Subject<DiskSyncSessionEvent>(),
+};
@@ -1,4 +1,5 @@
 import { dialogHandlers } from './dialog';
+import { diskSyncEvents, diskSyncHandlers } from './disk-sync';
 import { dbEventsV1, dbHandlersV1, nbstoreHandlers } from './nbstore';
 import { provideExposed } from './provide';
 import { workspaceEvents, workspaceHandlers } from './workspace';
@@ -6,12 +7,14 @@ import { workspaceEvents, workspaceHandlers } from './workspace';
 export const handlers = {
  db: dbHandlersV1,
  nbstore: nbstoreHandlers,
+  diskSync: diskSyncHandlers,
  workspace: workspaceHandlers,
  dialog: dialogHandlers,
 };

 export const events = {
  db: dbEventsV1,
+  diskSync: diskSyncEvents,
  workspace: workspaceEvents,
 };

@@ -82,30 +82,33 @@ function createSharedStorageApi(
    }
  });

-  const initPromise = (async () => {
-    try {
-      memory.setAll(init);
-      const latest = await ipcRenderer.invoke(
-        AFFINE_API_CHANNEL_NAME,
-        event === 'onGlobalStateChanged'
-          ? 'sharedStorage:getAllGlobalState'
-          : 'sharedStorage:getAllGlobalCache'
-      );
-      if (latest && typeof latest === 'object') {
-        memory.setAll(latest);
-      }
-    } catch (err) {
-      console.error('Failed to load initial shared storage', err);
-    } finally {
-      loaded = true;
-      while (updateQueue.length) {
-        const updates = updateQueue.shift();
-        if (updates) {
-          applyUpdates(updates);
-        }
+  // Load initial state synchronously so consumers can read values during early
+  // bootstrap without awaiting `ready`. This prevents feature config races
+  // (e.g. folder sync remote options) on first app load.
+  try {
+    memory.setAll(init);
+    const latest = ipcRenderer.sendSync(
+      AFFINE_API_CHANNEL_NAME,
+      event === 'onGlobalStateChanged'
+        ? 'sharedStorage:getAllGlobalState'
+        : 'sharedStorage:getAllGlobalCache'
+    );
+    if (latest && typeof latest === 'object') {
+      memory.setAll(latest);
+    }
+  } catch (err) {
+    console.error('Failed to load initial shared storage (sync)', err);
+  } finally {
+    loaded = true;
+    while (updateQueue.length) {
+      const updates = updateQueue.shift();
+      if (updates) {
+        applyUpdates(updates);
      }
    }
-  })();
+  }
+
+  const initPromise = Promise.resolve();

  return {
    ready: initPromise,
@@ -0,0 +1,118 @@
+import type { DiskSyncEvent } from '@affine/nbstore/disk';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const diskSyncMocks = vi.hoisted(() => {
+  return {
+    startSession: vi.fn(async () => {}),
+    stopSession: vi.fn(async () => {}),
+    applyLocalUpdate: vi.fn(async () => ({
+      docId: 'doc-1',
+      timestamp: new Date('2026-01-06T00:00:00.000Z'),
+    })),
+    subscribeEvents: vi.fn(
+      (
+        _sessionId: string,
+        _callback: (err: Error | null, event: DiskSyncEvent) => void
+      ) => {
+        return Promise.resolve({
+          unsubscribe: () => {},
+        });
+      }
+    ),
+  };
+});
+
+vi.mock('@affine/native', () => {
+  class DiskSyncMock {
+    subscribeEvents(
+      sessionId: string,
+      callback: (err: Error | null, event: DiskSyncEvent) => void
+    ) {
+      return diskSyncMocks.subscribeEvents(sessionId, callback);
+    }
+
+    startSession(
+      sessionId: string,
+      options: { workspaceId: string; syncFolder: string }
+    ) {
+      return diskSyncMocks.startSession(sessionId, options);
+    }
+
+    stopSession(sessionId: string) {
+      return diskSyncMocks.stopSession(sessionId);
+    }
+
+    applyLocalUpdate(
+      sessionId: string,
+      update: { docId: string; bin: Uint8Array },
+      origin?: string
+    ) {
+      return diskSyncMocks.applyLocalUpdate(sessionId, update, origin);
+    }
+  }
+
+  return { DiskSync: DiskSyncMock };
+});
+
+import {
+  applyLocalUpdate,
+  startSession,
+  stopSession,
+} from '../../src/helper/disk-sync/handlers';
+import { diskSyncSubjects } from '../../src/helper/disk-sync/subjects';
+
+describe('disk helper handlers', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it('forwards subscribeEvents payload and unsubscribes on stop', async () => {
+    const unsubscribe = vi.fn();
+    diskSyncMocks.subscribeEvents.mockImplementation(
+      (
+        _sessionId: string,
+        callback: (err: Error | null, event: DiskSyncEvent) => void
+      ) => {
+        callback(null, {
+          type: 'ready',
+        } as DiskSyncEvent);
+        return Promise.resolve({
+          unsubscribe,
+        });
+      }
+    );
+
+    const seen: string[] = [];
+    const subscription = diskSyncSubjects.event$.subscribe(payload => {
+      seen.push(payload.event.type);
+    });
+
+    await startSession('session-subscribe', {
+      workspaceId: 'workspace-subscribe',
+      syncFolder: '/tmp/disk-sync',
+    });
+
+    expect(seen).toContain('ready');
+    expect(diskSyncMocks.subscribeEvents).toHaveBeenCalledWith(
+      'session-subscribe',
+      expect.any(Function)
+    );
+
+    await stopSession('session-subscribe');
+    expect(unsubscribe).toHaveBeenCalledTimes(1);
+    subscription.unsubscribe();
+  });
+
+  it('throws when native applyLocalUpdate returns Error payload', async () => {
+    diskSyncMocks.applyLocalUpdate.mockResolvedValueOnce(
+      new Error('invalid_binary')
+    );
+
+    await expect(
+      applyLocalUpdate('session-subscribe', {
+        docId: 'doc-failed',
+        bin: new Uint8Array([1, 2, 3]),
+      })
+    ).rejects.toThrow('[disk] applyLocalUpdate failed: invalid_binary');
+  });
+});