feat(core): integrate realtime features (#15003)

2026-07-29 08:09:52 +08:00 · 2026-05-20 05:48:03 +08:00
parent 3e42bbf4fa
commit 9f33d37add
108 changed files with 3069 additions and 2729 deletions
@@ -235,11 +235,19 @@ class SocketManager {
      },
    };
  }
+
+  reset() {
+    this.socket.disconnect();
+  }
 }

 const SOCKET_MANAGER_CACHE = new Map<string, SocketManager>();
+function getSocketManagerKey(endpoint: string, isSelfHosted: boolean) {
+  return `${endpoint}:${isSelfHosted ? 'selfhosted' : 'cloud'}`;
+}
+
 function getSocketManager(endpoint: string, isSelfHosted: boolean) {
-  const key = `${endpoint}:${isSelfHosted ? 'selfhosted' : 'cloud'}`;
+  const key = getSocketManagerKey(endpoint, isSelfHosted);
  let manager = SOCKET_MANAGER_CACHE.get(key);
  if (!manager) {
    manager = new SocketManager(endpoint, isSelfHosted);
@@ -252,6 +260,12 @@ export class SocketConnection extends AutoReconnectConnection<{
  socket: Socket;
  disconnect: () => void;
 }> {
+  static resetSharedConnection(endpoint: string, isSelfHosted: boolean) {
+    SOCKET_MANAGER_CACHE.get(
+      getSocketManagerKey(endpoint, isSelfHosted)
+    )?.reset();
+  }
+
  manager = getSocketManager(this.endpoint, this.isSelfHosted);

  constructor(
@@ -41,10 +41,15 @@ class FakeSocket {
  }
 }

+const { resetSharedConnection } = vi.hoisted(() => ({
+  resetSharedConnection: vi.fn(),
+}));
 const socket = new FakeSocket();

 vi.mock('../../impls/cloud/socket', () => ({
  SocketConnection: class {
+    static resetSharedConnection = resetSharedConnection;
+
    readonly inner = { socket };
    status = 'connected';
    readonly maybeConnection = { socket };
@@ -68,6 +73,7 @@ beforeEach(() => {
  socket.nextSubscriptionId = 0;
  socket.connected = true;
  socket.disconnected = false;
+  resetSharedConnection.mockClear();
 });

 test('getRealtimeInputKey is deterministic for realtime subscription inputs', () => {
@@ -124,6 +130,33 @@ test('request rejects server ack error', async () => {
  );
 });

+test('user profile request can bootstrap without authenticated context', async () => {
+  const manager = new RealtimeManager();
+  manager.setContext({
+    endpoint: 'http://server',
+    isSelfHosted: false,
+    authenticated: false,
+  });
+  socket.nextRequestAck = { data: { user: null } };
+
+  await expect(manager.request('user.profile.get', {})).resolves.toEqual({
+    user: null,
+  });
+});
+
+test('non-bootstrap request still requires authenticated context', async () => {
+  const manager = new RealtimeManager();
+  manager.setContext({
+    endpoint: 'http://server',
+    isSelfHosted: false,
+    authenticated: false,
+  });
+
+  await expect(manager.request('notification.count.get', {})).rejects.toThrow(
+    'Realtime is not authenticated'
+  );
+});
+
 test('request rejects when aborted', async () => {
  const manager = new RealtimeManager();
  manager.setContext({
@@ -203,7 +236,7 @@ test('unsubscribe leaves server room and clears status', async () => {
  });
 });

-test('context switch disconnects socket and completes subscriptions', async () => {
+test('context switch disconnects socket and keeps subscriptions for reauth', async () => {
  const manager = new RealtimeManager();
  manager.setContext({
    endpoint: 'http://server',
@@ -223,11 +256,95 @@ test('context switch disconnects socket and completes subscriptions', async () =
  });

  expect(socket.disconnected).toBe(true);
-  expect(completed).toHaveBeenCalled();
+  expect(completed).not.toHaveBeenCalled();
  expect(manager.getStatus()).toMatchObject({
    endpoint: 'http://other-server',
    connected: false,
-    subscriptions: 0,
+    subscriptions: 1,
+  });
+});
+
+test('auth context switch resets shared socket connection', () => {
+  const manager = new RealtimeManager();
+  manager.setContext({
+    endpoint: 'http://server',
+    isSelfHosted: false,
+    authenticated: true,
+  });
+
+  manager.setContext({
+    endpoint: 'http://server',
+    isSelfHosted: false,
+    authenticated: false,
+  });
+
+  expect(resetSharedConnection).toHaveBeenCalledWith('http://server', false);
+});
+
+test('context switch resubscribes existing subscriptions on next connect', async () => {
+  const manager = new RealtimeManager();
+  manager.setContext({
+    endpoint: 'http://server',
+    isSelfHosted: false,
+    authenticated: true,
+  });
+  const received: unknown[] = [];
+  const subscription = manager
+    .subscribe('notification.count.changed', {})
+    .subscribe(event => received.push(event));
+  await vi.waitFor(() => expect(received).toEqual([{ type: 'ready' }]));
+
+  manager.setContext({
+    endpoint: 'http://other-server',
+    isSelfHosted: false,
+    authenticated: true,
+  });
+  await manager.request('notification.count.get', {});
+
+  await vi.waitFor(() =>
+    expect(
+      socket.emitted.filter(item => item.event === 'realtime:subscribe')
+    ).toHaveLength(2)
+  );
+  expect(received).toEqual([{ type: 'ready' }, { type: 'ready' }]);
+  subscription.unsubscribe();
+});
+
+test('unsubscribe uses resubscribed server subscription id', async () => {
+  const manager = new RealtimeManager();
+  manager.setContext({
+    endpoint: 'http://server',
+    isSelfHosted: false,
+    authenticated: true,
+  });
+  const subscription = manager
+    .subscribe('notification.count.changed', {})
+    .subscribe();
+  await vi.waitFor(() => expect(manager.getStatus().subscriptions).toBe(1));
+
+  manager.setContext({
+    endpoint: 'http://other-server',
+    isSelfHosted: false,
+    authenticated: true,
+  });
+  await manager.request('notification.count.get', {});
+  await vi.waitFor(() =>
+    expect(
+      socket.emitted.filter(item => item.event === 'realtime:subscribe')
+    ).toHaveLength(2)
+  );
+
+  subscription.unsubscribe();
+
+  expect(manager.getStatus().subscriptions).toBe(0);
+  expect(socket.emitted.at(-1)).toEqual({
+    event: 'realtime:unsubscribe',
+    payload: {
+      subscriptionId: 'sub-2',
+      topic: 'notification.count.changed',
+      input: {},
+      clientVersion: 'test',
+    },
  });
 });

@@ -37,6 +37,7 @@ export class RealtimeManager {
  private socketConnection?: SocketConnection;
  private socketKey?: string;
  private lastError?: { name: string; message: string };
+  private subscriptionsNeedResubscribe = false;
  private readonly subscriptions = new Map<
    string,
    {
@@ -44,21 +45,32 @@ export class RealtimeManager {
      input: RealtimeTopicInputOf<RealtimeTopicName>;
      inputKey: string;
      subject$: Subject<RealtimeEvent | RealtimeSubscriptionReady>;
+      onResubscribed: (subscriptionId: string) => void;
    }
  >();

  setContext(context: RealtimeContext) {
    const nextContext = { ...context };
+    const previousContext = this.context;
    const changed =
-      !this.context ||
-      this.context.endpoint !== nextContext.endpoint ||
-      this.context.isSelfHosted !== nextContext.isSelfHosted ||
-      this.context.authenticated !== nextContext.authenticated;
+      !previousContext ||
+      previousContext.endpoint !== nextContext.endpoint ||
+      previousContext.isSelfHosted !== nextContext.isSelfHosted ||
+      previousContext.authenticated !== nextContext.authenticated;

    this.context = nextContext;

    if (changed) {
      this.resetConnection();
+      if (
+        previousContext &&
+        previousContext.authenticated !== nextContext.authenticated
+      ) {
+        SocketConnection.resetSharedConnection(
+          previousContext.endpoint,
+          previousContext.isSelfHosted
+        );
+      }
    }
  }

@@ -67,7 +79,7 @@ export class RealtimeManager {
    input: RealtimeRequestInputOf<Op>,
    options?: { timeoutMs?: number; signal?: AbortSignal }
  ): Promise<RealtimeRequestOutputOf<Op>> {
-    const socket = await this.connect();
+    const socket = await this.connect(op === 'user.profile.get');
    const timeoutMs = options?.timeoutMs ?? DEFAULT_REQUEST_TIMEOUT;
    let timeoutId: ReturnType<typeof setTimeout> | undefined;
    let abortHandler: (() => void) | undefined;
@@ -154,6 +166,9 @@ export class RealtimeManager {
            input,
            inputKey: getRealtimeInputKey(input),
            subject$,
+            onResubscribed: nextSubscriptionId => {
+              subscriptionId = nextSubscriptionId;
+            },
          });
          subscriber.next({
            type: 'ready',
@@ -167,7 +182,7 @@ export class RealtimeManager {
              }
            },
            error: error => subscriber.error(error),
-            complete: () => subscriber.complete(),
+            complete: () => {},
          });
        } catch (error) {
          this.lastError = normalizeError(error);
@@ -209,8 +224,11 @@ export class RealtimeManager {
    };
  }

-  private async connect() {
-    if (!this.context?.endpoint || !this.context.authenticated) {
+  private async connect(allowUnauthenticated = false) {
+    if (
+      !this.context?.endpoint ||
+      (!this.context.authenticated && !allowUnauthenticated)
+    ) {
      const error = new Error('Realtime is not authenticated');
      error.name = 'RealtimeUnauthenticated';
      throw error;
@@ -232,6 +250,9 @@ export class RealtimeManager {
    this.socketConnection.inner.socket.on('realtime:event', this.handleEvent);
    this.socketConnection.inner.socket.off('connect', this.handleReconnect);
    this.socketConnection.inner.socket.on('connect', this.handleReconnect);
+    if (this.subscriptionsNeedResubscribe && this.context.authenticated) {
+      await this.resubscribeAll();
+    }
    return this.socketConnection.inner.socket;
  }

@@ -272,6 +293,7 @@ export class RealtimeManager {

        this.subscriptions.delete(subscriptionId);
        this.subscriptions.set(ack.data.subscriptionId, subscription);
+        subscription.onResubscribed(ack.data.subscriptionId);
        subscription.subject$.next({
          type: 'ready',
        });
@@ -281,6 +303,7 @@ export class RealtimeManager {
        subscription.subject$.error(error);
      }
    }
+    this.subscriptionsNeedResubscribe = false;
  }

  private resetConnection() {
@@ -297,9 +320,6 @@ export class RealtimeManager {
    }
    this.socketConnection = undefined;
    this.socketKey = undefined;
-    for (const subscription of this.subscriptions.values()) {
-      subscription.subject$.complete();
-    }
-    this.subscriptions.clear();
+    this.subscriptionsNeedResubscribe = this.subscriptions.size > 0;
  }
 }