[IMP] add recording feature

package.json

@@ -30,13 +30,14 @@
     "@jest/globals": "^29.6.2",
     "@rollup/plugin-commonjs": "^25.0.7",
     "@rollup/plugin-node-resolve": "^13.0.4",
-    "@rollup/plugin-typescript": "^10.0.1",
+    "@rollup/plugin-typescript": "^12.1.2",
     "@types/jest": "^29.5.0",
-    "@types/node": "^20.5.0",
+    "@types/node": "^22.13.14",
     "@types/ws": "^8.18.1",
-    "@typescript-eslint/eslint-plugin": "8.32.1",
-    "@typescript-eslint/parser": "8.32.1",
+    "@typescript-eslint/eslint-plugin": "^8.46.3",
+    "@typescript-eslint/parser": "^8.46.3",
     "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",
@@ -49,6 +50,6 @@
     "rollup": "^2.79.1",
     "rollup-plugin-license": "3.2.0",
     "ts-jest": "^29.3.4",
-    "typescript": "~5.4.3"
+    "typescript": "~5.9.3"
   }
 }

rollup.config.js

@@ -35,9 +35,6 @@ export default {
     plugins: [
         typescript({
             tsconfig: "./tsconfig_bundle.json",
-            declaration: false,
-            declarationMap: false,
-            sourceMap: false,
         }),
         resolve({
             browser: true,

src/client.ts

@@ -18,7 +18,14 @@ import {
     SERVER_REQUEST,
     WS_CLOSE_CODE
 } from "#src/shared/enums.ts";
-import type { JSONSerializable, StreamType, BusMessage } from "#src/shared/types";
+import type {
+    JSONSerializable,
+    StreamType,
+    BusMessage,
+    AvailableFeatures,
+    StartupData
+} from "#src/shared/types";
+import type { RequestMessage } from "#src/shared/bus-types";
 import type { TransportConfig, SessionId, SessionInfo } from "#src/models/session";
 
 interface Consumers {
@@ -55,11 +62,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;
@@ -141,6 +150,11 @@ const ACTIVE_STATES = new Set<SfuClientState>([
 export class SfuClient extends EventTarget {
     /** Connection errors encountered */
     public errors: Error[] = [];
+    public availableFeatures: AvailableFeatures = {
+        rtc: false,
+        recording: false
+    };
+    public isRecording: boolean = false;
     /** Current client state */
     private _state: SfuClientState = SfuClientState.DISCONNECTED;
     /** Communication bus */
@@ -256,6 +270,29 @@ export class SfuClient extends EventTarget {
         await Promise.all(proms);
         return stats;
     }
+    async startRecording(): Promise<boolean> {
+        if (this.state !== SfuClientState.CONNECTED) {
+            throw new Error("Cannot start recording when not connected");
+        }
+        return this._bus!.request(
+            {
+                name: CLIENT_REQUEST.START_RECORDING
+            },
+            { batch: true }
+        );
+    }
+
+    async stopRecording(): Promise<boolean> {
+        if (this.state !== SfuClientState.CONNECTED) {
+            throw new Error("Cannot stop recording when not connected");
+        }
+        return this._bus!.request(
+            {
+                name: CLIENT_REQUEST.STOP_RECORDING
+            },
+            { batch: true }
+        );
+    }
 
     /**
      * Updates the server with the info of the session (isTalking, isCameraOn,...) so that it can broadcast it to the
@@ -445,7 +482,14 @@ export class SfuClient extends EventTarget {
              */
             webSocket.addEventListener(
                 "message",
-                () => {
+                (message) => {
+                    if (message.data) {
+                        const { availableFeatures, isRecording } = JSON.parse(
+                            message.data
+                        ) as StartupData;
+                        this.availableFeatures = availableFeatures;
+                        this.isRecording = isRecording;
+                    }
                     resolve(new Bus(webSocket));
                 },
                 { once: true }
@@ -488,10 +532,10 @@ export class SfuClient extends EventTarget {
         });
         transport.on("produce", async ({ kind, rtpParameters, appData }, callback, errback) => {
             try {
-                const result = (await this._bus!.request({
+                const result = await this._bus!.request({
                     name: CLIENT_REQUEST.INIT_PRODUCER,
                     payload: { type: appData.type as StreamType, kind, rtpParameters }
-                })) as { id: string };
+                });
                 callback({ id: result.id });
             } catch (error) {
                 errback(error as Error);
@@ -576,10 +620,17 @@ 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;
         }
     }
 
-    private async _handleRequest({ name, payload }: BusMessage): Promise<JSONSerializable | void> {
+    private async _handleRequest({
+        name,
+        payload
+    }: RequestMessage): Promise<JSONSerializable | void> {
         switch (name) {
             case SERVER_REQUEST.INIT_CONSUMER: {
                 const { id, kind, producerId, rtpParameters, sessionId, type, active } = payload;

src/config.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"
     );
@@ -64,6 +65,11 @@ export const HTTP_INTERFACE: string = process.env.HTTP_INTERFACE || "0.0.0.0";
  */
 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) || testingMode;
+
 /**
  * The number of workers to spawn (up to core limits) to manage RTC servers.
  * 0 < NUM_WORKERS <= os.availableParallelism()
@@ -194,7 +200,26 @@ 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({
+    directory: os.tmpdir() + "/recordings",
+    enabled: RECORDING,
+    maxDuration: 1000 * 60 * 60, // 1 hour, could be a env-var.
+    fileTTL: 1000 * 60 * 60 * 24, // 24 hours
+    fileType: "mp4",
+    videoCodec: "libx264",
+    audioCodec: "aac",
+    audioLimit: 20,
+    cameraLimit: 4, // how many camera can be merged into one recording
+    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
 });
 
 // how many errors can occur before the session is closed, recovery attempts will be made until this limit is reached

src/models/channel.ts

@@ -12,7 +12,9 @@ import {
     type SessionId,
     type SessionInfo
 } from "#src/models/session.ts";
-import { getWorker, type RtcWorker } from "#src/services/rtc.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");
 
@@ -53,6 +55,7 @@ interface ChannelCreateOptions {
     key?: string;
     /** Whether to enable WebRTC functionality */
     useWebRtc?: boolean;
+    recordingAddress?: string | null;
 }
 interface JoinResult {
     /** The channel instance */
@@ -83,6 +86,8 @@ export class Channel extends EventEmitter {
     public readonly key?: Buffer;
     /** mediasoup Router for media routing */
     public readonly router?: Router;
+    /** Manages the recording of this channel, undefined if the feature is disabled */
+    public readonly recorder?: Recorder;
     /** Active sessions in this channel */
     public readonly sessions = new Map<SessionId, Session>();
     /** mediasoup Worker handling this channel */
@@ -102,7 +107,7 @@ export class Channel extends EventEmitter {
         issuer: string,
         options: ChannelCreateOptions = {}
     ): Promise<Channel> {
-        const { key, useWebRtc = true } = options;
+        const { key, useWebRtc = true, recordingAddress } = options;
         const safeIssuer = `${remoteAddress}::${issuer}`;
         const oldChannel = Channel.recordsByIssuer.get(safeIssuer);
         if (oldChannel) {
@@ -112,7 +117,7 @@ export class Channel extends EventEmitter {
         const channelOptions: ChannelCreateOptions & {
             worker?: Worker;
             router?: Router;
-        } = { key };
+        } = { key, recordingAddress: useWebRtc ? recordingAddress : null };
         if (useWebRtc) {
             channelOptions.worker = await getWorker();
             channelOptions.router = await channelOptions.worker.createRouter({
@@ -125,6 +130,8 @@ export class Channel extends EventEmitter {
         logger.info(
             `created channel ${channel.uuid} (${key ? "unique" : "global"} key) for ${safeIssuer}`
         );
+        logger.verbose(`rtc feature: ${Boolean(channel.router)}`);
+        logger.verbose(`recording feature: ${Boolean(channel.recorder)}`);
         const onWorkerDeath = () => {
             logger.warn(`worker died, closing channel ${channel.uuid}`);
             channel.close();
@@ -183,6 +190,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?.on("stateChange", () => this._broadcastState());
         this.key = key ? Buffer.from(key, "base64") : undefined;
         this.uuid = crypto.randomUUID();
         this.name = `${remoteAddress}*${this.uuid.slice(-5)}`;
@@ -295,6 +307,7 @@ export class Channel extends EventEmitter {
      * @fires Channel#close
      */
     close(): void {
+        this.recorder?.stop();
         for (const session of this.sessions.values()) {
             session.off("close", this._onSessionClose);
             session.close({ code: SESSION_CLOSE_CODE.CHANNEL_CLOSED });
@@ -309,6 +322,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/ffmpeg.ts

@@ -0,0 +1,8 @@
+import { EventEmitter } from "node:events";
+
+export class FFMPEG extends EventEmitter {
+
+    constructor() {
+        super();
+    }
+}