feat(rn): add end-to-end encryption support

packages/client/index.ts

@@ -26,6 +26,7 @@ export * from './src/helpers/sound-detector';
 export * from './src/helpers/loopback';
 export * from './src/helpers/MediaStreamRecorder';
 export * from './src/helpers/participantUtils';
+export * from './src/helpers/TypedEventEmitter';
 export * from './src/rtc/e2ee/E2EEManager';
 export * from './src/rtc/e2ee/EncryptionManager';
 export * as Browsers from './src/helpers/browsers';

packages/client/src/Call.ts

@@ -790,6 +790,11 @@ export class Call {
       await this.dynascaleManager?.dispose();
 
       this.state.setCallingState(CallingState.LEFT);
+      // `ringingSubject` is cleared further down, so this still reads true for
+      // a call that was ringing.
+      if (this.ringing) {
+        globalThis.streamRNVideoSDK?.ringingCallLifecycle?.onLeave(this);
+      }
       this.state.setParticipants([]);
       this.state.dispose();
 
@@ -1133,6 +1138,10 @@ export class Call {
   /**
    * Will start to watch for call related WebSocket events and initiate a call session with the server.
    *
+   * One instance is one call flow: discard it after leaving, cancelling, or a
+   * failed join, and create a fresh one for a later flow. Reconnection inside a
+   * live call is handled here and needs no new instance.
+   *
    * @returns a promise which resolves once the call join-flow has finished.
    */
   join = singleFlight(
@@ -1152,6 +1161,8 @@ export class Call {
     } = {}): Promise<void> => {
       const callingState = this.state.callingState;
 
+      // Ahead of the failure boundary below on purpose: a duplicate join on a
+      // live call is refused without tearing that call down.
       if ([CallingState.JOINED, CallingState.JOINING].includes(callingState)) {
         throw new Error(`Illegal State: call.join() shall be called only once`);
       }
@@ -1171,43 +1182,63 @@ export class Call {
         this.ringingSubject.next(true);
       }
 
+      // A ringing call is joined by the SDK rather than by app code, so React
+      // Native prepares it from in here - there is no earlier point at which the
+      // app holds the call. Read after `data.ring` above, which is what makes an
+      // outgoing call ringing in the first place.
+      const ringingLifecycle = this.ringing
+        ? globalThis.streamRNVideoSDK?.ringingCallLifecycle
+        : undefined;
       const callingX = globalThis.streamRNVideoSDK?.callingX;
-      if (callingX) {
-        // for Android/iOS, we need to start the call in the callingx library as soon as possible
-        await callingX.joinCall(this, this.clientState.calls);
+      try {
+        if (ringingLifecycle) {
+          await ringingLifecycle.beforeJoin(this);
+          if (supersededByLeave()) {
+            this.logger.debug('Join superseded by leave; not joining');
+            return;
+          }
+        }
+
+        if (callingX) {
+          // for Android/iOS, we need to start the call in the callingx library as soon as possible
+          await callingX.joinCall(
+            this,
+            this.clientState.calls,
+            supersededByLeave,
+          );
+          if (supersededByLeave()) {
+            this.logger.debug('Join superseded by leave; not setting up');
+            return;
+          }
+        }
+
+        await this.setup();
         if (supersededByLeave()) {
-          this.logger.debug('Join superseded by leave; not setting up');
+          this.logger.debug('Join superseded by leave; not registering');
           return;
         }
-      }
-
-      await this.setup();
-      if (supersededByLeave()) {
-        this.logger.debug('Join superseded by leave; not registering');
-        return;
-      }
 
-      this.clientEventReporter.registerCall(this.cid, {
-        callType: this.type,
-        callId: this.id,
-        getCallSessionId: () => this.state.session?.id ?? '',
-        getSfuId: () => this.credentials?.server.edge_name ?? '',
-      });
+        this.clientEventReporter.registerCall(this.cid, {
+          callType: this.type,
+          callId: this.id,
+          getCallSessionId: () => this.state.session?.id ?? '',
+          getSfuId: () => this.credentials?.server.edge_name ?? '',
+        });
 
-      this.joinResponseTimeout = joinResponseTimeout;
-      this.rpcRequestTimeout = rpcRequestTimeout;
-      // we will count the number of join failures per SFU.
-      // once the number of failures reaches 2, we will piggyback on the `migrating_from`
-      // field to force the coordinator to provide us another SFU
-      const sfuJoinFailures = new Map<string, number>();
-      const joinData: JoinCallData = data;
-      maxJoinRetries = Math.max(maxJoinRetries, 1);
-      try {
+        this.joinResponseTimeout = joinResponseTimeout;
+        this.rpcRequestTimeout = rpcRequestTimeout;
+        // we will count the number of join failures per SFU.
+        // once the number of failures reaches 2, we will piggyback on the `migrating_from`
+        // field to force the coordinator to provide us another SFU
+        const sfuJoinFailures = new Map<string, number>();
+        const joinData: JoinCallData = data;
+        maxJoinRetries = Math.max(maxJoinRetries, 1);
         await this.clientEventReporter.withJoinLifecycle(
           this.cid,
           { joinReason: 'first-attempt', joinSource },
           async () => {
             for (let attempt = 0; attempt < maxJoinRetries; attempt++) {
+              // Also covers a leave that lands during the backoff below.
               if (supersededByLeave()) {
                 this.logger.debug('Join superseded by leave; not attempting');
                 return;
@@ -1263,6 +1294,9 @@ export class Call {
         );
       } catch (error) {
         callingX?.endCall(this, 'error');
+        // Ends the failed ringing flow and releases what its setup installed.
+        // Never rejects, so `error` is what the caller sees.
+        await ringingLifecycle?.onJoinFailed(this);
         throw error;
       }
     },
@@ -2601,10 +2635,11 @@ export class Call {
    * Must be called before {@link join} so the RTCPeerConnection can be
    * configured for E2EE.
    *
-   * The manager is kept across {@link leave} so a rejoin of this same instance
-   * stays encrypted: do not dispose it while this call may be joined again.
-   * A disposed manager throws from `encrypt`/`decrypt` rather than silently
-   * publishing nothing, so re-attach a fresh one instead of reusing it.
+   * The application owns the manager's lifetime - the SDK never disposes it, and
+   * closing the peer connections does not release it. Discard the manager
+   * together with this call instance once its flow has ended, and create a fresh
+   * pair for a later flow: a disposed manager throws from `encrypt`/`decrypt`
+   * rather than silently publishing nothing.
    *
    * @param e2ee - Any `E2EEManager`. Use `EncryptionManager.create()` for the
    *         built-in AES-GCM scheme, or pass your own implementation.

packages/client/src/__tests__/Call.ringingLifecycle.test.ts

@@ -0,0 +1,242 @@
+/**
+ * @vitest-environment happy-dom
+ */
+
+import '../rtc/__tests__/mocks/webrtc.mocks';
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { Call } from '../Call';
+import { CallingState } from '../store';
+import { StreamClient } from '../coordinator/connection/client';
+import { ClientEventReporter } from '../reporting';
+import { generateUUIDv4 } from '../coordinator/connection/utils';
+import { ClientState } from '../store';
+
+/**
+ * The core's whole part in the React Native ringing lifecycle: await RN's
+ * preparation inside the join, tell RN when a ringing join failed terminally or
+ * the call ended, and stop a join that a `leave()` overtook. What preparation
+ * and release actually do is RN's, and is tested there.
+ */
+
+const createCall = (ringing: boolean) => {
+  const streamClient = new StreamClient('abc');
+  return new Call({
+    type: 'test',
+    id: generateUUIDv4(),
+    ringing,
+    streamClient,
+    clientEventReporter: new ClientEventReporter({
+      streamClient,
+      enabled: false,
+    }),
+    clientState: new ClientState(),
+  });
+};
+
+const install = (overrides: Record<string, unknown> = {}) => {
+  const beforeJoin = vi.fn(() => Promise.resolve());
+  const onJoinFailed = vi.fn(() => Promise.resolve());
+  const onLeave = vi.fn();
+  const callingX = { joinCall: vi.fn(), endCall: vi.fn() };
+  globalThis.streamRNVideoSDK = {
+    ringingCallLifecycle: { beforeJoin, onJoinFailed, onLeave },
+    callingX,
+    // `leave()` reaches this unconditionally; the globals object is all-or-nothing
+    callManager: { setup: vi.fn(), start: vi.fn(), stop: vi.fn() },
+    ...overrides,
+  } as any;
+  return { beforeJoin, onJoinFailed, onLeave, callingX };
+};
+
+describe('ringing call lifecycle integration', () => {
+  beforeEach(() => {
+    globalThis.streamRNVideoSDK = undefined;
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue({
+      headers: { get: () => 'AMS1-P2' },
+    } as Response);
+  });
+
+  afterEach(() => {
+    globalThis.streamRNVideoSDK = undefined;
+    vi.restoreAllMocks();
+  });
+
+  it('prepares a ringing call before it joins', async () => {
+    const { beforeJoin, callingX } = install();
+    const call = createCall(true);
+    const doJoin = vi.spyOn(call as any, 'doJoin').mockResolvedValue(undefined);
+
+    await call.join();
+
+    expect(beforeJoin).toHaveBeenCalledTimes(1);
+    expect(beforeJoin).toHaveBeenCalledBefore(callingX.joinCall);
+    expect(doJoin).toHaveBeenCalledTimes(1);
+  });
+
+  it('leaves ordinary calls to join themselves', async () => {
+    const { beforeJoin, onJoinFailed } = install();
+    const call = createCall(false);
+    vi.spyOn(call as any, 'doJoin').mockResolvedValue(undefined);
+
+    await call.join();
+
+    expect(beforeJoin).not.toHaveBeenCalled();
+    expect(onJoinFailed).not.toHaveBeenCalled();
+  });
+
+  it('does not join when preparation fails, and reports the failure', async () => {
+    const { beforeJoin, onJoinFailed } = install();
+    beforeJoin.mockRejectedValue(new Error('no key'));
+    const call = createCall(true);
+    const doJoin = vi.spyOn(call as any, 'doJoin').mockResolvedValue(undefined);
+
+    await expect(call.join()).rejects.toThrow('no key');
+
+    expect(doJoin).not.toHaveBeenCalled();
+    expect(onJoinFailed).toHaveBeenCalledWith(call);
+  });
+
+  it('reports a terminal join failure, keeping the original error', async () => {
+    const { onJoinFailed } = install();
+    const call = createCall(true);
+    vi.spyOn(call as any, 'doJoin').mockRejectedValue(new Error('sfu down'));
+
+    await expect(call.join({ maxJoinRetries: 1 })).rejects.toThrow('sfu down');
+
+    expect(onJoinFailed).toHaveBeenCalledWith(call);
+  });
+
+  it('refuses a duplicate join on a live call without tearing it down', async () => {
+    const { beforeJoin, onJoinFailed, onLeave } = install();
+    const call = createCall(true);
+    vi.spyOn(call as any, 'doJoin').mockImplementation(async () => {
+      call.state.setCallingState(CallingState.JOINED);
+    });
+
+    await call.join();
+    await expect(call.join()).rejects.toThrow('Illegal State');
+
+    expect(beforeJoin).toHaveBeenCalledTimes(1);
+    // the guard sits ahead of the failure boundary, so nothing is released
+    expect(onJoinFailed).not.toHaveBeenCalled();
+    expect(onLeave).not.toHaveBeenCalled();
+  });
+
+  it('tells the owner when a ringing call ends', async () => {
+    const { onLeave } = install();
+    const call = createCall(true);
+
+    await call.leave();
+
+    expect(onLeave).toHaveBeenCalledWith(call);
+    expect(call.state.callingState).toBe(CallingState.LEFT);
+  });
+
+  it('does not tell the owner about an ordinary call', async () => {
+    const { onLeave } = install();
+
+    await createCall(false).leave();
+
+    expect(onLeave).not.toHaveBeenCalled();
+  });
+
+  it('keeps an app-owned manager across leave, as setE2EEManager promises', async () => {
+    install();
+    const call = createCall(true);
+    const manager = { encrypt: vi.fn(), decrypt: vi.fn() } as any;
+    call.setE2EEManager(manager);
+
+    await call.leave();
+
+    expect(call.e2eeManager).toBe(manager);
+  });
+
+  it('abandons a join that a leave overtook during native registration', async () => {
+    let finishRegistration: () => void = () => {};
+    const { callingX, onLeave, onJoinFailed } = install();
+    callingX.joinCall.mockImplementation(
+      () => new Promise<void>((resolve) => (finishRegistration = resolve)),
+    );
+    const call = createCall(true);
+    const doJoin = vi.spyOn(call as any, 'doJoin').mockResolvedValue(undefined);
+
+    const joining = call.join();
+    await vi.waitFor(() => expect(callingX.joinCall).toHaveBeenCalled());
+    await call.leave({ reject: false });
+    finishRegistration();
+
+    // A superseding leave is not a join failure: the join settles quietly rather
+    // than rejecting, and the error path stays shut.
+    await expect(joining).resolves.toBeUndefined();
+    expect(onJoinFailed).not.toHaveBeenCalled();
+    expect(callingX.endCall).not.toHaveBeenCalledWith(call, 'error');
+    // Release still happens - `leave()` itself fires it, which is why the join
+    // does not need to.
+    expect(onLeave).toHaveBeenCalledTimes(1);
+    // `doJoin` captures the generation itself, so this check has to be here
+    expect(doJoin).not.toHaveBeenCalled();
+  });
+
+  it('does not resume the retry loop after a leave during backoff', async () => {
+    vi.useFakeTimers();
+    try {
+      install();
+      const call = createCall(true);
+      const doJoin = vi
+        .spyOn(call as any, 'doJoin')
+        .mockRejectedValue(new Error('recoverable'));
+
+      const joining = call.join().catch(() => {});
+      await vi.waitFor(() => expect(doJoin).toHaveBeenCalledTimes(1));
+      await call.leave({ reject: false });
+      await vi.advanceTimersByTimeAsync(30_000);
+      await joining;
+
+      expect(doJoin).toHaveBeenCalledTimes(1);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('prepares an outgoing call that this join makes ringing', async () => {
+    const { beforeJoin } = install();
+    // a fresh instance, not ringing until its first join says so
+    const call = createCall(false);
+    vi.spyOn(call as any, 'doJoin').mockResolvedValue(undefined);
+
+    await call.join({ ring: true });
+
+    // reading `call.ringing` before applying `options.ring` would skip
+    // preparation entirely on this path
+    expect(beforeJoin).toHaveBeenCalledTimes(1);
+  });
+
+  it('stops a join that a leave overtook during preparation', async () => {
+    let releaseHook: () => void = () => {};
+    const { beforeJoin, onLeave, onJoinFailed } = install();
+    beforeJoin.mockImplementation(
+      () => new Promise<void>((resolve) => (releaseHook = resolve)),
+    );
+    const call = createCall(true);
+    const doJoin = vi.spyOn(call as any, 'doJoin').mockResolvedValue(undefined);
+
+    const joining = call.join();
+    await vi.waitFor(() => expect(beforeJoin).toHaveBeenCalled());
+    await call.leave({ reject: false });
+    releaseHook();
+
+    await expect(joining).resolves.toBeUndefined();
+    expect(onJoinFailed).not.toHaveBeenCalled();
+    // Whatever `beforeJoin` installed is released exactly once, by the `leave()`
+    // that overtook the join rather than by the join itself.
+    expect(onLeave).toHaveBeenCalledTimes(1);
+    expect(doJoin).not.toHaveBeenCalled();
+    expect(call.state.callingState).toBe(CallingState.LEFT);
+  });
+
+  it('is inert when no React Native globals are registered', async () => {
+    const call = createCall(true);
+    await expect(call.leave()).resolves.not.toThrow();
+  });
+});