feat(client): rewrite coordinator socket + StreamClient (parallel rollout)

packages/client/src/__tests__/StreamVideoClient.parity.test.ts

@@ -0,0 +1,77 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import 'dotenv/config';
+import { StreamVideoClient } from '../StreamVideoClient';
+import { generateUUIDv4 } from '../coordinator/connection/utils';
+import { StreamClient } from '@stream-io/node-sdk';
+
+const apiKey = process.env.STREAM_API_KEY;
+const secret = process.env.STREAM_SECRET;
+
+const haveCredentials = Boolean(apiKey && secret);
+const describeIfCreds = haveCredentials ? describe : describe.skip;
+
+describeIfCreds('StreamVideoClient parity (legacy vs. new coordinator)', () => {
+  let serverClient: StreamClient;
+  let tokenProvider: (userId: string) => () => Promise<string>;
+
+  beforeEach(() => {
+    serverClient = new StreamClient(apiKey!, secret!);
+    tokenProvider = (userId: string) => async () =>
+      new Promise<string>((resolve) => {
+        setTimeout(() => {
+          resolve(serverClient.generateUserToken({ user_id: userId }));
+        }, 50);
+      });
+  });
+
+  describe.each([
+    { mode: 'new', useLegacyCoordinator: false },
+    { mode: 'legacy', useLegacyCoordinator: true },
+  ])('mode=$mode', ({ useLegacyCoordinator }) => {
+    let client: StreamVideoClient;
+
+    beforeEach(() => {
+      client = new StreamVideoClient(apiKey!, {
+        browser: true,
+        timeout: 15000,
+        useLegacyCoordinator,
+      });
+    });
+    afterEach(async () => {
+      await client.disconnectUser();
+    });
+
+    it('private endpoint waits for connection then succeeds', async () => {
+      client.connectUser({ id: 'jane' }, tokenProvider('jane'));
+      const response = await client.queryCalls({});
+      expect(response).toBeDefined();
+    });
+
+    it('connection_id is enriched on REST requests', async () => {
+      client.connectUser({ id: 'jane' }, tokenProvider('jane'));
+      const spy = vi.spyOn(client.streamClient.axiosInstance, 'post');
+      await client.call('default', generateUUIDv4()).getOrCreate();
+      const requestConfig: { params: Record<string, unknown> } = spy.mock.calls[
+        spy.mock.calls.length - 1
+      ][2] as {
+        params: Record<string, unknown>;
+      };
+      const params = requestConfig.params;
+      const connectionId = client.streamClient._getConnectionID();
+      expect(connectionId).toBeDefined();
+      expect(params.connection_id).toBe(connectionId);
+    });
+
+    it('private endpoint throws after disconnectUser', async () => {
+      await client.connectUser({ id: 'jane' }, tokenProvider('jane'));
+      await client.disconnectUser();
+      await expect(() => client.queryCalls({})).rejects.toThrowError();
+    });
+
+    it('public endpoint does not require connectUser', async () => {
+      const userId = `guest-${generateUUIDv4()}`;
+      const response = await client.createGuestUser({ user: { id: userId } });
+      expect(response.user.id).toContain(userId);
+    });
+  });
+});

packages/client/src/coordinator/connection/__tests__/ConnectionIdGate.test.ts

@@ -0,0 +1,103 @@
+import { describe, expect, it } from 'vitest';
+import { ConnectionIdGate } from '../internal/ConnectionIdGate';
+
+describe('ConnectionIdGate', () => {
+  it('await() before arm() throws', () => {
+    const gate = new ConnectionIdGate();
+    expect(() => gate.await()).toThrow(/not armed/);
+  });
+
+  it('arm() is idempotent while pending; second await() returns the same promise', () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    const first = gate.await();
+    gate.arm();
+    const second = gate.await();
+    expect(first).toBe(second);
+  });
+
+  it('resolve(id) settles all pending awaiters with id', async () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    const a = gate.await();
+    const b = gate.await();
+    gate.resolve('conn-id');
+    await expect(a).resolves.toBe('conn-id');
+    await expect(b).resolves.toBe('conn-id');
+  });
+
+  it('reject(err) settles all pending awaiters with the error', async () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    const a = gate.await();
+    const err = new Error('boom');
+    gate.reject(err);
+    await expect(a).rejects.toThrow('boom');
+  });
+
+  it('reset() discards the gate; subsequent await() throws until next arm()', async () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    const inflight = gate.await();
+    gate.resolve('conn-id');
+    await inflight; // drain
+    gate.reset();
+    expect(() => gate.await()).toThrow(/not armed/);
+    gate.arm();
+    expect(gate.isPending()).toBe(true);
+  });
+
+  it('resolve()/reject() after settlement is a no-op', async () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    const inflight = gate.await();
+    gate.resolve('first');
+    gate.resolve('second');
+    gate.reject(new Error('ignored'));
+    await expect(inflight).resolves.toBe('first');
+  });
+
+  it('isPending / isSettled reflect state correctly across the lifecycle', () => {
+    const gate = new ConnectionIdGate();
+    expect(gate.isPending()).toBe(false);
+    expect(gate.isSettled()).toBe(false);
+    gate.arm();
+    expect(gate.isPending()).toBe(true);
+    expect(gate.isSettled()).toBe(false);
+    gate.resolve('id');
+    expect(gate.isPending()).toBe(false);
+    expect(gate.isSettled()).toBe(true);
+    gate.reset();
+    expect(gate.isPending()).toBe(false);
+    expect(gate.isSettled()).toBe(false);
+  });
+
+  it('F1: rejects in-flight awaiter BEFORE subsequent arm() rotates state', async () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    const awaiter = gate.await(); // captures P1
+    gate.reject(new Error('boom')); // settles P1
+    gate.arm(); // P1 already settled, so this rotates to fresh P2
+    const newAwaiter = gate.await();
+    await expect(awaiter).rejects.toThrow('boom');
+    expect(awaiter).not.toBe(newAwaiter);
+  });
+
+  it('arm() after settlement rotates to a fresh pending promise', async () => {
+    const gate = new ConnectionIdGate();
+    gate.arm();
+    gate.resolve('id-1');
+    gate.arm();
+    expect(gate.isPending()).toBe(true);
+    const fresh = gate.await();
+    let resolved = false;
+    fresh.then(() => {
+      resolved = true;
+    });
+    // Give microtasks a tick
+    await Promise.resolve();
+    expect(resolved).toBe(false);
+    gate.resolve('id-2');
+    await expect(fresh).resolves.toBe('id-2');
+  });
+});