[WIP] recording: state broadcasting, updated tests

Author: ThanhDodeurOdoo

package-lock.json

@@ -37,6 +37,7 @@
     "@typescript-eslint/eslint-plugin": "8.32.1",
     "@typescript-eslint/parser": "8.32.1",
     "eslint": "8.57.1",
+    "eslint-config-prettier": "^10.1.8",
     "eslint-plugin-import": "^2.25.3",
     "eslint-plugin-jest": "28.11.0",
     "eslint-plugin-node": "^11.1.0",

package.json

@@ -18,7 +18,13 @@ import {
     SERVER_REQUEST,
     WS_CLOSE_CODE
 } from "#src/shared/enums.ts";
-import type { JSONSerializable, StreamType, BusMessage, AvailableFeatures } from "#src/shared/types";
+import type {
+    JSONSerializable,
+    StreamType,
+    BusMessage,
+    AvailableFeatures,
+    StartupData
+} from "#src/shared/types";
 import type { TransportConfig, SessionId, SessionInfo } from "#src/models/session";
 
 interface Consumers {
@@ -55,11 +61,13 @@ export enum CLIENT_UPDATE {
     /** A session has left the channel */
     DISCONNECT = "disconnect",
     /** Session info has changed */
-    INFO_CHANGE = "info_change"
+    INFO_CHANGE = "info_change",
+    CHANNEL_INFO_CHANGE = "channel_info_change"
 }
 type ClientUpdatePayload =
     | { senderId: SessionId; message: JSONSerializable }
     | { sessionId: SessionId }
+    | { isRecording: boolean }
     | Record<SessionId, SessionInfo>
     | {
           type: StreamType;
@@ -142,9 +150,10 @@ export class SfuClient extends EventTarget {
     /** Connection errors encountered */
     public errors: Error[] = [];
     public availableFeatures: AvailableFeatures = {
-        "rtc": false,
-        "recording": false,
+        rtc: false,
+        recording: false
     };
+    public isRecording: boolean = false;
     /** Current client state */
     private _state: SfuClientState = SfuClientState.DISCONNECTED;
     /** Communication bus */
@@ -266,7 +275,7 @@ export class SfuClient extends EventTarget {
         }
         return this._bus?.request(
             {
-                name: CLIENT_REQUEST.START_RECORDING,
+                name: CLIENT_REQUEST.START_RECORDING
             },
             { batch: true }
         );
@@ -278,7 +287,7 @@ export class SfuClient extends EventTarget {
         }
         return this._bus?.request(
             {
-                name: CLIENT_REQUEST.STOP_RECORDING,
+                name: CLIENT_REQUEST.STOP_RECORDING
             },
             { batch: true }
         );
@@ -473,7 +482,13 @@ export class SfuClient extends EventTarget {
             webSocket.addEventListener(
                 "message",
                 (message) => {
-                    this.availableFeatures = JSON.parse(message.data) as AvailableFeatures;
+                    if (message.data) {
+                        const { availableFeatures, isRecording } = JSON.parse(
+                            message.data
+                        ) as StartupData;
+                        this.availableFeatures = availableFeatures;
+                        this.isRecording = isRecording;
+                    }
                     resolve(new Bus(webSocket));
                 },
                 { once: true }
@@ -604,6 +619,10 @@ export class SfuClient extends EventTarget {
             case SERVER_MESSAGE.INFO_CHANGE:
                 this._updateClient(CLIENT_UPDATE.INFO_CHANGE, payload);
                 break;
+            case SERVER_MESSAGE.CHANNEL_INFO_CHANGE:
+                this.isRecording = payload.isRecording;
+                this._updateClient(CLIENT_UPDATE.CHANNEL_INFO_CHANGE, payload);
+                break;
         }
     }
 

src/client.ts

@@ -11,6 +11,7 @@ import type { ProducerOptions } from "mediasoup-client/lib/Producer";
 const FALSY_INPUT = new Set(["disable", "false", "none", "no", "0"]);
 type LogLevel = "none" | "error" | "warn" | "info" | "debug" | "verbose";
 type WorkerLogLevel = "none" | "error" | "warn" | "debug";
+const testingMode = Boolean(process.env.JEST_WORKER_ID);
 
 // ------------------------------------------------------------
 // ------------------   ENV VARIABLES   -----------------------
@@ -22,7 +23,7 @@ type WorkerLogLevel = "none" | "error" | "warn" | "debug";
  * e.g: AUTH_KEY=u6bsUQEWrHdKIuYplirRnbBmLbrKV5PxKG7DtA71mng=
  */
 export const AUTH_KEY: string = process.env.AUTH_KEY!;
-if (!AUTH_KEY && !process.env.JEST_WORKER_ID) {
+if (!AUTH_KEY && !testingMode) {
     throw new Error(
         "AUTH_KEY env variable is required, it is not possible to authenticate requests without it"
     );
@@ -34,7 +35,7 @@ if (!AUTH_KEY && !process.env.JEST_WORKER_ID) {
  * e.g: PUBLIC_IP=190.165.1.70
  */
 export const PUBLIC_IP: string = process.env.PUBLIC_IP!;
-if (!PUBLIC_IP && !process.env.JEST_WORKER_ID) {
+if (!PUBLIC_IP && !testingMode) {
     throw new Error(
         "PUBLIC_IP env variable is required, clients cannot establish webRTC connections without it"
     );
@@ -67,7 +68,7 @@ export const PORT: number = Number(process.env.PORT) || 8070;
 /**
  * Whether the recording feature is enabled, false by default.
  */
-export const RECORDING: boolean = Boolean(process.env.RECORDING);
+export const RECORDING: boolean = Boolean(process.env.RECORDING) || testingMode;
 
 /**
  * The number of workers to spawn (up to core limits) to manage RTC servers.
@@ -199,7 +200,7 @@ export const timeouts: TimeoutConfig = Object.freeze({
     // how long before a channel is closed after the last session leaves
     channel: 60 * 60_000,
     // how long to wait to gather messages before sending through the bus
-    busBatch: process.env.JEST_WORKER_ID ? 10 : 300
+    busBatch: testingMode ? 10 : 300
 });
 
 export const recording = Object.freeze({
@@ -212,16 +213,15 @@ export const recording = Object.freeze({
     audioCodec: "aac",
     audioLimit: 20,
     cameraLimit: 4, // how many camera can be merged into one recording
-    screenLimit: 1,
+    screenLimit: 1
 });
 
 // TODO: This should probably be env variable, and at least documented so that deployment can open these ports.
 export const dynamicPorts = Object.freeze({
     min: 50000,
-    max: 59999,
+    max: 59999
 });
 
-
 // how many errors can occur before the session is closed, recovery attempts will be made until this limit is reached
 export const maxSessionErrors: number = 6;
 

src/config.ts

@@ -14,6 +14,7 @@ import {
 } from "#src/models/session.ts";
 import { Recorder } from "#src/models/recorder.ts";
 import { getWorker, type RtcWorker } from "#src/services/resources.ts";
+import { SERVER_MESSAGE } from "#src/shared/enums.ts";
 
 const logger = new Logger("CHANNEL");
 
@@ -187,7 +188,11 @@ export class Channel extends EventEmitter {
         const now = new Date();
         this.createDate = now.toISOString();
         this.remoteAddress = remoteAddress;
-        this.recorder = config.recording.enabled && options.recordingAddress ? new Recorder(this, options.recordingAddress) : undefined;
+        this.recorder =
+            config.recording.enabled && options.recordingAddress
+                ? new Recorder(this, options.recordingAddress)
+                : undefined;
+        this.recorder?.on("stateChange", () => this._broadcastState());
         this.key = key ? Buffer.from(key, "base64") : undefined;
         this.uuid = crypto.randomUUID();
         this.name = `${remoteAddress}*${this.uuid.slice(-5)}`;
@@ -315,6 +320,28 @@ export class Channel extends EventEmitter {
         this.emit("close", this.uuid);
     }
 
+    /**
+     * Broadcast the state of this channel to all its participants
+     */
+    private _broadcastState() {
+        for (const session of this.sessions.values()) {
+            // TODO maybe the following should be on session and some can be made in common with the startupData getter.
+            if (!session.bus) {
+                logger.warn(`tried to broadcast state to session ${session.id}, but had no Bus`);
+                continue;
+            }
+            session.bus.send(
+                {
+                    name: SERVER_MESSAGE.CHANNEL_INFO_CHANGE,
+                    payload: {
+                        isRecording: Boolean(this.recorder?.isRecording)
+                    }
+                },
+                { batch: true }
+            );
+        }
+    }
+
     /**
      * @param event - Close event with session ID
      * @fires Channel#sessionLeave

src/models/channel.ts

@@ -1,22 +1,22 @@
 import { EventEmitter } from "node:events";
-import type { Channel } from "./channel";
-import { getFolder } from "#src/services/resources.ts";
-import type { Folder } from "#src/services/resources.ts";
+import { getFolder, type Folder } from "#src/services/resources.ts";
 import { Logger } from "#src/utils/utils.ts";
 
+import type { Channel } from "./channel";
+
 export enum RECORDER_STATE {
     STARTED = "started",
-    STOPPED = "stopped",
+    STOPPED = "stopped"
 }
 const logger = new Logger("RECORDER");
 
 export class Recorder extends EventEmitter {
     channel: Channel;
-    state: RECORDER_STATE = RECORDER_STATE.STOPPED;
     folder: Folder | undefined;
     ffmpeg = null;
     /** Path to which the final recording will be uploaded to */
     recordingAddress: string;
+    private _state: RECORDER_STATE = RECORDER_STATE.STOPPED;
 
     constructor(channel: Channel, recordingAddress: string) {
         super();
@@ -26,10 +26,10 @@ export class Recorder extends EventEmitter {
 
     async start() {
         if (this.state === RECORDER_STATE.STOPPED) {
-            this.folder  = getFolder();
+            this.folder = getFolder();
             this.state = RECORDER_STATE.STARTED;
-             logger.trace("TO IMPLEMENT");
-             // TODO ffmpeg instance creation for recording to folder.path with proper name, start, build timestamps object
+            logger.trace("TO IMPLEMENT");
+            // TODO ffmpeg instance creation for recording to folder.path with proper name, start, build timestamps object
         }
         this._record();
         return { state: this.state };
@@ -38,7 +38,11 @@ export class Recorder extends EventEmitter {
     async stop() {
         if (this.state === RECORDER_STATE.STARTED) {
             logger.trace("TO IMPLEMENT");
-            await this.folder!.seal("test-name");
+            try {
+                await this.folder!.seal("test-name");
+            } catch {
+                logger.verbose("failed to save the recording"); // TODO maybe warn and give more info
+            }
             this.folder = undefined;
             // TODO ffmpeg instance stop, cleanup,
             // only resolve promise and switch state when completely ready to start a new recording.
@@ -47,11 +51,24 @@ export class Recorder extends EventEmitter {
         return { state: this.state };
     }
 
+    get isRecording(): boolean {
+        return this.state === RECORDER_STATE.STARTED;
+    }
+
+    get state(): RECORDER_STATE {
+        return this._state;
+    }
+
+    set state(state: RECORDER_STATE) {
+        this._state = state;
+        this.emit("stateChange", state);
+    }
+
     /**
      * @param video whether we want to record videos or not (will always record audio)
      */
     _record(video: boolean = false) {
-        console.trace(`TO IMPLEMENT: recording channel ${this.channel.name}, video: ${video}`);
+        logger.trace(`TO IMPLEMENT: recording channel ${this.channel.name}, video: ${video}`);
         // iterate all producers on all sessions of the channel, create a ffmpeg for each,
         // save them on a map by session id+type.
         // check if recording for that session id+type is already in progress

src/models/recorder.ts

@@ -1,14 +1,14 @@
 import { EventEmitter } from "node:events";
 
 import type {
-    IceParameters,
-    IceCandidate,
-    DtlsParameters,
-    SctpParameters,
     Consumer,
+    DtlsParameters,
+    IceCandidate,
+    IceParameters,
     Producer,
-    WebRtcTransport,
-    RtpCapabilities
+    RtpCapabilities,
+    SctpParameters,
+    WebRtcTransport
 } from "mediasoup/node/lib/types";
 
 import * as config from "#src/config.ts";
@@ -20,9 +20,10 @@ import {
     SERVER_REQUEST,
     STREAM_TYPE
 } from "#src/shared/enums.ts";
-import type {JSONSerializable, StreamType, BusMessage, AvailableFeatures } from "#src/shared/types";
+import type { BusMessage, JSONSerializable, StartupData, StreamType } from "#src/shared/types";
 import type { Bus } from "#src/shared/bus.ts";
 import type { Channel } from "#src/models/channel.ts";
+import { RECORDER_STATE } from "#src/models/recorder.ts";
 
 export type SessionId = number | string;
 export type SessionInfo = {
@@ -168,11 +169,16 @@ export class Session extends EventEmitter {
         this.setMaxListeners(config.CHANNEL_SIZE * 2);
     }
 
-    get availableFeatures(): AvailableFeatures {
+    get startupData(): StartupData {
         return {
-            "rtc": Boolean(this._channel.router),
-            "recording": Boolean(this._channel.router && this._channel.recorder && this.permissions.recording)
-        }
+            availableFeatures: {
+                rtc: Boolean(this._channel.router),
+                recording: Boolean(
+                    this._channel.router && this._channel.recorder && this.permissions.recording
+                )
+            },
+            isRecording: this._channel.recorder?.state === RECORDER_STATE.STARTED
+        };
     }
 
     get name(): string {

src/models/session.ts

@@ -131,7 +131,7 @@ function connect(webSocket: WebSocket, credentials: Credentials): Session {
     const bus = new Bus(webSocket, { batchDelay: config.timeouts.busBatch });
     const { session } = Channel.join(channel.uuid, session_id);
     session.updatePermissions(permissions);
-    webSocket.send(JSON.stringify(session.availableFeatures)); // client can start using ws after this message.
+    webSocket.send(JSON.stringify(session.startupData)); // client can start using ws after this message.
     session.once("close", ({ code }: { code: string }) => {
         let wsCloseCode = WS_CLOSE_CODE.CLEAN;
         switch (code) {

src/services/ws.ts

@@ -24,7 +24,9 @@ export enum SERVER_MESSAGE {
     /** Signals the clients that one of the session in their channel has left */
     SESSION_LEAVE = "SESSION_LEAVE",
     /** Signals the clients that the info (talking, mute,...) of one of the session in their channel has changed */
-    INFO_CHANGE = "S_INFO_CHANGE"
+    INFO_CHANGE = "S_INFO_CHANGE",
+    /** Signals the clients that the info of the channel (isRecording,...) has changed */
+    CHANNEL_INFO_CHANGE = "C_INFO_CHANGE"
 }
 
 export enum CLIENT_REQUEST {