ThreadNet-Web/src/voice-broadcast/models/VoiceBroadcastPlayback.ts

/*
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 {
    EventType,
    MatrixClient,
    MatrixEvent,
    MsgType,
    RelationType,
} from "matrix-js-sdk/src/matrix";
import { TypedEventEmitter } from "matrix-js-sdk/src/models/typed-event-emitter";

import { Playback, PlaybackState } from "../../audio/Playback";
import { PlaybackManager } from "../../audio/PlaybackManager";
import { UPDATE_EVENT } from "../../stores/AsyncStore";
import { MediaEventHelper } from "../../utils/MediaEventHelper";
import { IDestroyable } from "../../utils/IDestroyable";
import { VoiceBroadcastChunkEventType, VoiceBroadcastInfoEventType, VoiceBroadcastInfoState } from "..";
import { RelationsHelper, RelationsHelperEvent } from "../../events/RelationsHelper";
import { getReferenceRelationsForEvent } from "../../events";

export enum VoiceBroadcastPlaybackState {
    Paused,
    Playing,
    Stopped,
    Buffering,
}

export enum VoiceBroadcastPlaybackEvent {
    LengthChanged = "length_changed",
    StateChanged = "state_changed",
    InfoStateChanged = "info_state_changed",
}

interface EventMap {
    [VoiceBroadcastPlaybackEvent.LengthChanged]: (length: number) => void;
    [VoiceBroadcastPlaybackEvent.StateChanged]: (
        state: VoiceBroadcastPlaybackState,
        playback: VoiceBroadcastPlayback
    ) => void;
    [VoiceBroadcastPlaybackEvent.InfoStateChanged]: (state: VoiceBroadcastInfoState) => void;
}

export class VoiceBroadcastPlayback
    extends TypedEventEmitter<VoiceBroadcastPlaybackEvent, EventMap>
    implements IDestroyable {
    private state = VoiceBroadcastPlaybackState.Stopped;
    private infoState: VoiceBroadcastInfoState;
    private chunkEvents = new Map<string, MatrixEvent>();
    private queue: Playback[] = [];
    private currentlyPlaying: Playback;
    private lastInfoEvent: MatrixEvent;
    private chunkRelationHelper: RelationsHelper;
    private infoRelationHelper: RelationsHelper;

    public constructor(
        public readonly infoEvent: MatrixEvent,
        private client: MatrixClient,
    ) {
        super();
        this.addInfoEvent(this.infoEvent);
        this.setUpRelationsHelper();
    }

    private setUpRelationsHelper(): void {
        this.infoRelationHelper = new RelationsHelper(
            this.infoEvent,
            RelationType.Reference,
            VoiceBroadcastInfoEventType,
            this.client,
        );
        this.infoRelationHelper.on(RelationsHelperEvent.Add, this.addInfoEvent);
        this.infoRelationHelper.emitCurrent();

        this.chunkRelationHelper = new RelationsHelper(
            this.infoEvent,
            RelationType.Reference,
            EventType.RoomMessage,
            this.client,
        );
        this.chunkRelationHelper.on(RelationsHelperEvent.Add, this.addChunkEvent);
        this.chunkRelationHelper.emitCurrent();
    }

    private addChunkEvent = async (event: MatrixEvent): Promise<boolean> => {
        const eventId = event.getId();

        if (!eventId
            || eventId.startsWith("~!") // don't add local events
            || event.getContent()?.msgtype !== MsgType.Audio // don't add non-audio event
            || this.chunkEvents.has(eventId)) {
            return false;
        }

        this.chunkEvents.set(eventId, event);

        if (this.getState() !== VoiceBroadcastPlaybackState.Stopped) {
            await this.enqueueChunk(event);
        }

        if (this.getState() === VoiceBroadcastPlaybackState.Buffering) {
            await this.start();
        }

        return true;
    };

    private addInfoEvent = (event: MatrixEvent): void => {
        if (this.lastInfoEvent && this.lastInfoEvent.getTs() >= event.getTs()) {
            // Only handle newer events
            return;
        }

        const state = event.getContent()?.state;

        if (!Object.values(VoiceBroadcastInfoState).includes(state)) {
            // Do not handle unknown voice broadcast states
            return;
        }

        this.lastInfoEvent = event;
        this.setInfoState(state);
    };

    private async loadChunks(): Promise<void> {
        const relations = getReferenceRelationsForEvent(this.infoEvent, EventType.RoomMessage, this.client);
        const chunkEvents = relations?.getRelations();

        if (!chunkEvents) {
            return;
        }

        for (const chunkEvent of chunkEvents) {
            await this.enqueueChunk(chunkEvent);
        }
    }

    private async enqueueChunk(chunkEvent: MatrixEvent) {
        const sequenceNumber = parseInt(chunkEvent.getContent()?.[VoiceBroadcastChunkEventType]?.sequence, 10);
        if (isNaN(sequenceNumber) || sequenceNumber < 1) return;

        const helper = new MediaEventHelper(chunkEvent);
        const blob = await helper.sourceBlob.value;
        const buffer = await blob.arrayBuffer();
        const playback = PlaybackManager.instance.createPlaybackInstance(buffer);
        await playback.prepare();
        playback.clockInfo.populatePlaceholdersFrom(chunkEvent);
        this.queue[sequenceNumber - 1] = playback; // -1 because the sequence number starts at 1
        playback.on(UPDATE_EVENT, (state) => this.onPlaybackStateChange(playback, state));
    }

    private async onPlaybackStateChange(playback: Playback, newState: PlaybackState) {
        if (newState !== PlaybackState.Stopped) {
            return;
        }

        await this.playNext(playback);
    }

    private async playNext(current: Playback): Promise<void> {
        const next = this.queue[this.queue.indexOf(current) + 1];

        if (next) {
            this.setState(VoiceBroadcastPlaybackState.Playing);
            this.currentlyPlaying = next;
            await next.play();
            return;
        }

        if (this.getInfoState() === VoiceBroadcastInfoState.Stopped) {
            this.setState(VoiceBroadcastPlaybackState.Stopped);
        } else {
            // No more chunks available, although the broadcast is not finished → enter buffering state.
            this.setState(VoiceBroadcastPlaybackState.Buffering);
        }
    }

    public async start(): Promise<void> {
        if (this.queue.length === 0) {
            await this.loadChunks();
        }

        const toPlayIndex = this.getInfoState() === VoiceBroadcastInfoState.Stopped
            ? 0 // start at the beginning for an ended voice broadcast
            : this.queue.length - 1; // start at the current chunk for an ongoing voice broadcast

        if (this.queue[toPlayIndex]) {
            this.setState(VoiceBroadcastPlaybackState.Playing);
            this.currentlyPlaying = this.queue[toPlayIndex];
            await this.currentlyPlaying.play();
            return;
        }

        this.setState(VoiceBroadcastPlaybackState.Buffering);
    }

    public get length(): number {
        return this.chunkEvents.size;
    }

    public stop(): void {
        this.setState(VoiceBroadcastPlaybackState.Stopped);

        if (this.currentlyPlaying) {
            this.currentlyPlaying.stop();
        }
    }

    public pause(): void {
        // stopped voice broadcasts cannot be paused
        if (this.getState() === VoiceBroadcastPlaybackState.Stopped) return;

        this.setState(VoiceBroadcastPlaybackState.Paused);
        if (!this.currentlyPlaying) return;
        this.currentlyPlaying.pause();
    }

    public resume(): void {
        if (!this.currentlyPlaying) {
            // no playback to resume, start from the beginning
            this.start();
            return;
        }

        this.setState(VoiceBroadcastPlaybackState.Playing);
        this.currentlyPlaying.play();
    }

    /**
     * Toggles the playback:
     * stopped → playing
     * playing → paused
     * paused → playing
     */
    public async toggle() {
        if (this.state === VoiceBroadcastPlaybackState.Stopped) {
            await this.start();
            return;
        }

        if (this.state === VoiceBroadcastPlaybackState.Paused) {
            this.resume();
            return;
        }

        this.pause();
    }

    public getState(): VoiceBroadcastPlaybackState {
        return this.state;
    }

    private setState(state: VoiceBroadcastPlaybackState): void {
        if (this.state === state) {
            return;
        }

        this.state = state;
        this.emit(VoiceBroadcastPlaybackEvent.StateChanged, state, this);
    }

    public getInfoState(): VoiceBroadcastInfoState {
        return this.infoState;
    }

    private setInfoState(state: VoiceBroadcastInfoState): void {
        if (this.infoState === state) {
            return;
        }

        this.infoState = state;
        this.emit(VoiceBroadcastPlaybackEvent.InfoStateChanged, state);
    }

    private destroyQueue(): void {
        this.queue.forEach(p => p.destroy());
        this.queue = [];
    }

    public destroy(): void {
        this.chunkRelationHelper.destroy();
        this.infoRelationHelper.destroy();
        this.removeAllListeners();
        this.destroyQueue();
    }
}