Implement Voice Broadcast recording (#9307)

* Implement VoiceBroadcastRecording

* Implement PR feedback

* Add voice broadcast recording stores

* Refactor startNewVoiceBroadcastRecording

* Refactor VoiceBroadcastRecordingsStore to VoiceBroadcastRecording

* Rename VoiceBroadcastRecording to VoiceBroadcastRecorder

* Return remaining chunk on stop

* Extract createVoiceMessageContent

* Implement recording

* Replace dev value with config

* Fix clientInformation-test

* Refactor VoiceBroadcastRecording

* Fix VoiceBroadcastRecording types

* Re-order getter

* Mark voice_broadcast config as optional

* Merge voice-broadcast modules

* Remove underscore props

* Add Optional types

* Add return types everywhere

* Remove test casts

* Add magic comments

* Trigger CI

* Switch VoiceBroadcastRecorder to TypedEventEmitter

* Trigger CI

* Add voice broadcast chunk event content

Co-authored-by: Travis Ralston <travisr@matrix.org>

src/IConfigOptions.ts

@@ -181,6 +181,11 @@ export interface IConfigOptions {
 
     sync_timeline_limit?: number;
     dangerously_allow_unsafe_and_insecure_passwords?: boolean; // developer option
+
+    voice_broadcast?: {
+        // length per voice chunk in seconds
+        chunk_length?: number;
+    };
 }
 
 export interface ISsoRedirectOptions {

src/SdkConfig.ts

@@ -46,6 +46,9 @@ export const DEFAULTS: IConfigOptions = {
         logo: require("../res/img/element-desktop-logo.svg").default,
         url: "https://element.io/get-started",
     },
+    voice_broadcast: {
+        chunk_length: 60 * 1000, // one minute
+    },
 };
 
 export default class SdkConfig {

src/audio/VoiceRecording.ts

@@ -203,9 +203,7 @@ export class VoiceRecording extends EventEmitter implements IDestroyable {
         // In testing, recorder time and worker time lag by about 400ms, which is roughly the
         // time needed to encode a sample/frame.
         //
-        // Ref for recorderSeconds: https://github.com/chris-rudmin/opus-recorder#instance-fields
-        const recorderSeconds = this.recorder.encodedSamplePosition / 48000;
-        const secondsLeft = TARGET_MAX_LENGTH - recorderSeconds;
+        const secondsLeft = TARGET_MAX_LENGTH - this.recorderSeconds;
         if (secondsLeft < 0) { // go over to make sure we definitely capture that last frame
             // noinspection JSIgnoredPromiseFromCall - we aren't concerned with it overlapping
             this.stop();
@@ -217,6 +215,13 @@ export class VoiceRecording extends EventEmitter implements IDestroyable {
         }
     };
 
+    /**
+     * {@link https://github.com/chris-rudmin/opus-recorder#instance-fields ref for recorderSeconds}
+    */
+    public get recorderSeconds() {
+        return this.recorder.encodedSamplePosition / 48000;
+    }
+
     public async start(): Promise<void> {
         if (this.recording) {
             throw new Error("Recording already in progress");

src/voice-broadcast/audio/VoiceBroadcastRecorder.ts

@@ -0,0 +1,141 @@
+/*
+Copyright 2022 The Matrix.org Foundation C.I.C.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { Optional } from "matrix-events-sdk";
+import { TypedEventEmitter } from "matrix-js-sdk/src/models/typed-event-emitter";
+
+import { VoiceRecording } from "../../audio/VoiceRecording";
+import SdkConfig, { DEFAULTS } from "../../SdkConfig";
+import { concat } from "../../utils/arrays";
+import { IDestroyable } from "../../utils/IDestroyable";
+
+export enum VoiceBroadcastRecorderEvent {
+    ChunkRecorded = "chunk_recorded",
+}
+
+interface EventMap {
+    [VoiceBroadcastRecorderEvent.ChunkRecorded]: (chunk: ChunkRecordedPayload) => void;
+}
+
+export interface ChunkRecordedPayload {
+    buffer: Uint8Array;
+    length: number;
+}
+
+/**
+ * This class provides the function to seamlessly record fixed length chunks.
+ * Subscribe with on(VoiceBroadcastRecordingEvents.ChunkRecorded, (payload: ChunkRecordedPayload) => {})
+ * to retrieve chunks while recording.
+ */
+export class VoiceBroadcastRecorder
+    extends TypedEventEmitter<VoiceBroadcastRecorderEvent, EventMap>
+    implements IDestroyable {
+    private headers = new Uint8Array(0);
+    private chunkBuffer = new Uint8Array(0);
+    private previousChunkEndTimePosition = 0;
+    private pagesFromRecorderCount = 0;
+
+    public constructor(
+        private voiceRecording: VoiceRecording,
+        public readonly targetChunkLength: number,
+    ) {
+        super();
+        this.voiceRecording.onDataAvailable = this.onDataAvailable;
+    }
+
+    public async start(): Promise<void> {
+        return this.voiceRecording.start();
+    }
+
+    /**
+     * Stops the recording and returns the remaining chunk (if any).
+     */
+    public async stop(): Promise<Optional<ChunkRecordedPayload>> {
+        await this.voiceRecording.stop();
+        return this.extractChunk();
+    }
+
+    public get contentType(): string {
+        return this.voiceRecording.contentType;
+    }
+
+    private get chunkLength(): number {
+        return this.voiceRecording.recorderSeconds - this.previousChunkEndTimePosition;
+    }
+
+    private onDataAvailable = (data: ArrayBuffer): void => {
+        const dataArray = new Uint8Array(data);
+        this.pagesFromRecorderCount++;
+
+        if (this.pagesFromRecorderCount <= 2) {
+            // first two pages contain the headers
+            this.headers = concat(this.headers, dataArray);
+            return;
+        }
+
+        this.handleData(dataArray);
+    };
+
+    private handleData(data: Uint8Array): void {
+        this.chunkBuffer = concat(this.chunkBuffer, data);
+        this.emitChunkIfTargetLengthReached();
+    }
+
+    private emitChunkIfTargetLengthReached(): void {
+        if (this.chunkLength >= this.targetChunkLength) {
+            this.emitAndResetChunk();
+        }
+    }
+
+    /**
+     * Extracts the current chunk and resets the buffer.
+     */
+    private extractChunk(): Optional<ChunkRecordedPayload> {
+        if (this.chunkBuffer.length === 0) {
+            return null;
+        }
+
+        const currentRecorderTime = this.voiceRecording.recorderSeconds;
+        const payload: ChunkRecordedPayload = {
+            buffer: concat(this.headers, this.chunkBuffer),
+            length: this.chunkLength,
+        };
+        this.chunkBuffer = new Uint8Array(0);
+        this.previousChunkEndTimePosition = currentRecorderTime;
+        return payload;
+    }
+
+    private emitAndResetChunk(): void {
+        if (this.chunkBuffer.length === 0) {
+            return;
+        }
+
+        this.emit(
+            VoiceBroadcastRecorderEvent.ChunkRecorded,
+            this.extractChunk(),
+        );
+    }
+
+    public destroy(): void {
+        this.removeAllListeners();
+        this.voiceRecording.destroy();
+    }
+}
+
+export const createVoiceBroadcastRecorder = (): VoiceBroadcastRecorder => {
+    const targetChunkLength = SdkConfig.get("voice_broadcast")?.chunk_length || DEFAULTS.voice_broadcast!.chunk_length;
+    return new VoiceBroadcastRecorder(new VoiceRecording(), targetChunkLength);
+};

src/voice-broadcast/components/VoiceBroadcastBody.tsx

@@ -31,7 +31,7 @@ export const VoiceBroadcastBody: React.FC<IBodyProps> = ({ mxEvent }) => {
     const client = MatrixClientPeg.get();
     const room = client.getRoom(mxEvent.getRoomId());
     const recording = VoiceBroadcastRecordingsStore.instance().getByInfoEvent(mxEvent, client);
-    const [recordingState, setRecordingState] = useState(recording.state);
+    const [recordingState, setRecordingState] = useState(recording.getState());
 
     useTypedEventEmitter(
         recording,

src/voice-broadcast/components/index.ts

@@ -1,19 +0,0 @@
-/*
-Copyright 2022 The Matrix.org Foundation C.I.C.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-export * from "./atoms/LiveBadge";
-export * from "./molecules/VoiceBroadcastRecordingBody";
-export * from "./VoiceBroadcastBody";

src/voice-broadcast/index.ts

@@ -21,10 +21,14 @@ limitations under the License.
 
 import { RelationType } from "matrix-js-sdk/src/matrix";
 
-export * from "./components";
-export * from "./models";
-export * from "./utils";
-export * from "./stores";
+export * from "./audio/VoiceBroadcastRecorder";
+export * from "./components/VoiceBroadcastBody";
+export * from "./components/atoms/LiveBadge";
+export * from "./components/molecules/VoiceBroadcastRecordingBody";
+export * from "./models/VoiceBroadcastRecording";
+export * from "./stores/VoiceBroadcastRecordingsStore";
+export * from "./utils/shouldDisplayAsVoiceBroadcastTile";
+export * from "./utils/startNewVoiceBroadcastRecording";
 
 export const VoiceBroadcastInfoEventType = "io.element.voice_broadcast_info";
 

src/voice-broadcast/models/VoiceBroadcastRecording.ts

@@ -14,10 +14,22 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import { MatrixClient, MatrixEvent, RelationType } from "matrix-js-sdk/src/matrix";
+import { logger } from "matrix-js-sdk/src/logger";
+import { IAbortablePromise, MatrixClient, MatrixEvent, RelationType } from "matrix-js-sdk/src/matrix";
 import { TypedEventEmitter } from "matrix-js-sdk/src/models/typed-event-emitter";
 
-import { VoiceBroadcastInfoEventType, VoiceBroadcastInfoState } from "..";
+import {
+    ChunkRecordedPayload,
+    createVoiceBroadcastRecorder,
+    VoiceBroadcastInfoEventType,
+    VoiceBroadcastInfoState,
+    VoiceBroadcastRecorder,
+    VoiceBroadcastRecorderEvent,
+} from "..";
+import { uploadFile } from "../../ContentMessages";
+import { IEncryptedFile } from "../../customisations/models/IMediaEventContent";
+import { createVoiceMessageContent } from "../../utils/createVoiceMessageContent";
+import { IDestroyable } from "../../utils/IDestroyable";
 
 export enum VoiceBroadcastRecordingEvent {
     StateChanged = "liveness_changed",
@@ -27,8 +39,12 @@ interface EventMap {
     [VoiceBroadcastRecordingEvent.StateChanged]: (state: VoiceBroadcastInfoState) => void;
 }
 
-export class VoiceBroadcastRecording extends TypedEventEmitter<VoiceBroadcastRecordingEvent, EventMap> {
-    private _state: VoiceBroadcastInfoState;
+export class VoiceBroadcastRecording
+    extends TypedEventEmitter<VoiceBroadcastRecordingEvent, EventMap>
+    implements IDestroyable {
+    private state: VoiceBroadcastInfoState;
+    private recorder: VoiceBroadcastRecorder;
+    private sequence = 1;
 
     public constructor(
         public readonly infoEvent: MatrixEvent,
@@ -43,21 +59,89 @@ export class VoiceBroadcastRecording extends TypedEventEmitter<VoiceBroadcastRec
             VoiceBroadcastInfoEventType,
         );
         const relatedEvents = relations?.getRelations();
-        this._state = !relatedEvents?.find((event: MatrixEvent) => {
+        this.state = !relatedEvents?.find((event: MatrixEvent) => {
             return event.getContent()?.state === VoiceBroadcastInfoState.Stopped;
         }) ? VoiceBroadcastInfoState.Started : VoiceBroadcastInfoState.Stopped;
 
         // TODO Michael W: add listening for updates
     }
 
+    public async start(): Promise<void> {
+        return this.getRecorder().start();
+    }
+
+    public async stop(): Promise<void> {
+        this.setState(VoiceBroadcastInfoState.Stopped);
+        await this.stopRecorder();
+        await this.sendStoppedStateEvent();
+    }
+
+    public getState(): VoiceBroadcastInfoState {
+        return this.state;
+    }
+
+    private getRecorder(): VoiceBroadcastRecorder {
+        if (!this.recorder) {
+            this.recorder = createVoiceBroadcastRecorder();
+            this.recorder.on(VoiceBroadcastRecorderEvent.ChunkRecorded, this.onChunkRecorded);
+        }
+
+        return this.recorder;
+    }
+
+    public destroy(): void {
+        if (this.recorder) {
+            this.recorder.off(VoiceBroadcastRecorderEvent.ChunkRecorded, this.onChunkRecorded);
+            this.recorder.stop();
+        }
+
+        this.removeAllListeners();
+    }
+
     private setState(state: VoiceBroadcastInfoState): void {
-        this._state = state;
+        this.state = state;
         this.emit(VoiceBroadcastRecordingEvent.StateChanged, this.state);
     }
 
-    public async stop() {
-        this.setState(VoiceBroadcastInfoState.Stopped);
-        // TODO Michael W: add error handling
+    private onChunkRecorded = async (chunk: ChunkRecordedPayload): Promise<void> => {
+        const { url, file } = await this.uploadFile(chunk);
+        await this.sendVoiceMessage(chunk, url, file);
+    };
+
+    private uploadFile(chunk: ChunkRecordedPayload): IAbortablePromise<{ url?: string, file?: IEncryptedFile }> {
+        return uploadFile(
+            this.client,
+            this.infoEvent.getRoomId(),
+            new Blob(
+                [chunk.buffer],
+                {
+                    type: this.getRecorder().contentType,
+                },
+            ),
+        );
+    }
+
+    private async sendVoiceMessage(chunk: ChunkRecordedPayload, url: string, file: IEncryptedFile): Promise<void> {
+        const content = createVoiceMessageContent(
+            url,
+            this.getRecorder().contentType,
+            Math.round(chunk.length * 1000),
+            chunk.buffer.length,
+            file,
+        );
+        content["m.relates_to"] = {
+            rel_type: RelationType.Reference,
+            event_id: this.infoEvent.getId(),
+        };
+        content["io.element.voice_broadcast_chunk"] = {
+            sequence: this.sequence++,
+        };
+
+        await this.client.sendMessage(this.infoEvent.getRoomId(), content);
+    }
+
+    private async sendStoppedStateEvent(): Promise<void> {
+        // TODO Michael W: add error handling for state event
         await this.client.sendStateEvent(
             this.infoEvent.getRoomId(),
             VoiceBroadcastInfoEventType,
@@ -72,7 +156,18 @@ export class VoiceBroadcastRecording extends TypedEventEmitter<VoiceBroadcastRec
         );
     }
 
-    public get state(): VoiceBroadcastInfoState {
-        return this._state;
+    private async stopRecorder(): Promise<void> {
+        if (!this.recorder) {
+            return;
+        }
+
+        try {
+            const lastChunk = await this.recorder.stop();
+            if (lastChunk) {
+                await this.onChunkRecorded(lastChunk);
+            }
+        } catch (err) {
+            logger.warn("error stopping voice broadcast recorder", err);
+        }
     }
 }

src/voice-broadcast/models/index.ts

@@ -1,17 +0,0 @@
-/*
-Copyright 2022 The Matrix.org Foundation C.I.C.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-export * from "./VoiceBroadcastRecording";

src/voice-broadcast/stores/VoiceBroadcastRecordingsStore.ts

@@ -31,7 +31,7 @@ interface EventMap {
  * This store provides access to the current and specific Voice Broadcast recordings.
  */
 export class VoiceBroadcastRecordingsStore extends TypedEventEmitter<VoiceBroadcastRecordingsStoreEvent, EventMap> {
-    private _current: VoiceBroadcastRecording | null;
+    private current: VoiceBroadcastRecording | null;
     private recordings = new Map<string, VoiceBroadcastRecording>();
 
     public constructor() {
@@ -39,15 +39,15 @@ export class VoiceBroadcastRecordingsStore extends TypedEventEmitter<VoiceBroadc
     }
 
     public setCurrent(current: VoiceBroadcastRecording): void {
-        if (this._current === current) return;
+        if (this.current === current) return;
 
-        this._current = current;
+        this.current = current;
         this.recordings.set(current.infoEvent.getId(), current);
         this.emit(VoiceBroadcastRecordingsStoreEvent.CurrentChanged, current);
     }
 
-    public get current(): VoiceBroadcastRecording {
-        return this._current;
+    public getCurrent(): VoiceBroadcastRecording {
+        return this.current;
     }
 
     public getByInfoEvent(infoEvent: MatrixEvent, client: MatrixClient): VoiceBroadcastRecording {
@@ -60,12 +60,12 @@ export class VoiceBroadcastRecordingsStore extends TypedEventEmitter<VoiceBroadc
         return this.recordings.get(infoEventId);
     }
 
-    public static readonly _instance = new VoiceBroadcastRecordingsStore();
+    private static readonly cachedInstance = new VoiceBroadcastRecordingsStore();
 
     /**
      * TODO Michael W: replace when https://github.com/matrix-org/matrix-react-sdk/pull/9293 has been merged
      */
-    public static instance() {
-        return VoiceBroadcastRecordingsStore._instance;
+    public static instance(): VoiceBroadcastRecordingsStore {
+        return VoiceBroadcastRecordingsStore.cachedInstance;
     }
 }

src/voice-broadcast/stores/index.ts

@@ -1,17 +0,0 @@
-/*
-Copyright 2022 The Matrix.org Foundation C.I.C.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-export * from "./VoiceBroadcastRecordingsStore";

src/voice-broadcast/utils/index.ts

@@ -1,18 +0,0 @@
-/*
-Copyright 2022 The Matrix.org Foundation C.I.C.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-export * from "./shouldDisplayAsVoiceBroadcastTile";
-export * from "./startNewVoiceBroadcastRecording";

src/voice-broadcast/utils/startNewVoiceBroadcastRecording.ts

@@ -53,6 +53,7 @@ export const startNewVoiceBroadcastRecording = async (
                 client,
             );
             recordingsStore.setCurrent(recording);
+            recording.start();
             resolve(recording);
         }
     };