Use server side relations for voice broadcasts (#9534)

src/events/RelationsHelper.ts

@@ -38,6 +38,8 @@ export class RelationsHelper
     extends TypedEventEmitter<RelationsHelperEvent, EventMap>
     implements IDestroyable {
     private relations?: Relations;
+    private eventId: string;
+    private roomId: string;
 
     public constructor(
         private event: MatrixEvent,
@@ -46,6 +48,21 @@ export class RelationsHelper
         private client: MatrixClient,
     ) {
         super();
+
+        const eventId = event.getId();
+
+        if (!eventId) {
+            throw new Error("unable to create RelationsHelper: missing event ID");
+        }
+
+        const roomId = event.getRoomId();
+
+        if (!roomId) {
+            throw new Error("unable to create RelationsHelper: missing room ID");
+        }
+
+        this.eventId = eventId;
+        this.roomId = roomId;
         this.setUpRelations();
     }
 
@@ -73,7 +90,7 @@ export class RelationsHelper
     private setRelations(): void {
         const room = this.client.getRoom(this.event.getRoomId());
         this.relations = room?.getUnfilteredTimelineSet()?.relations?.getChildEventsForEvent(
-            this.event.getId(),
+            this.eventId,
             this.relationType,
             this.relationEventType,
         );
@@ -87,6 +104,32 @@ export class RelationsHelper
         this.relations?.getRelations()?.forEach(e => this.emit(RelationsHelperEvent.Add, e));
     }
 
+    public getCurrent(): MatrixEvent[] {
+        return this.relations?.getRelations() || [];
+    }
+
+    /**
+     * Fetches all related events from the server and emits them.
+     */
+    public async emitFetchCurrent(): Promise<void> {
+        let nextBatch: string | undefined = undefined;
+
+        do {
+            const response = await this.client.relations(
+                this.roomId,
+                this.eventId,
+                this.relationType,
+                this.relationEventType,
+                {
+                    from: nextBatch,
+                    limit: 50,
+                },
+            );
+            nextBatch = response?.nextBatch;
+            response?.events.forEach(e => this.emit(RelationsHelperEvent.Add, e));
+        } while (nextBatch);
+    }
+
     public destroy(): void {
         this.removeAllListeners();
         this.event.off(MatrixEventEvent.RelationsCreated, this.onRelationsCreated);

src/voice-broadcast/models/VoiceBroadcastPlayback.ts

@@ -32,7 +32,6 @@ import { MediaEventHelper } from "../../utils/MediaEventHelper";
 import { IDestroyable } from "../../utils/IDestroyable";
 import { VoiceBroadcastInfoEventType, VoiceBroadcastInfoState } from "..";
 import { RelationsHelper, RelationsHelperEvent } from "../../events/RelationsHelper";
-import { getReferenceRelationsForEvent } from "../../events";
 import { VoiceBroadcastChunkEvents } from "../utils/VoiceBroadcastChunkEvents";
 
 export enum VoiceBroadcastPlaybackState {
@@ -89,15 +88,27 @@ export class VoiceBroadcastPlayback
         this.setUpRelationsHelper();
     }
 
-    private setUpRelationsHelper(): void {
+    private async setUpRelationsHelper(): Promise<void> {
         this.infoRelationHelper = new RelationsHelper(
             this.infoEvent,
             RelationType.Reference,
             VoiceBroadcastInfoEventType,
             this.client,
         );
-        this.infoRelationHelper.on(RelationsHelperEvent.Add, this.addInfoEvent);
-        this.infoRelationHelper.emitCurrent();
+        this.infoRelationHelper.getCurrent().forEach(this.addInfoEvent);
+
+        if (this.infoState !== VoiceBroadcastInfoState.Stopped) {
+            // Only required if not stopped. Stopped is the final state.
+            this.infoRelationHelper.on(RelationsHelperEvent.Add, this.addInfoEvent);
+
+            try {
+                await this.infoRelationHelper.emitFetchCurrent();
+            } catch (err) {
+                logger.warn("error fetching server side relation for voice broadcast info", err);
+                // fall back to local events
+                this.infoRelationHelper.emitCurrent();
+            }
+        }
 
         this.chunkRelationHelper = new RelationsHelper(
             this.infoEvent,
@@ -106,7 +117,15 @@ export class VoiceBroadcastPlayback
             this.client,
         );
         this.chunkRelationHelper.on(RelationsHelperEvent.Add, this.addChunkEvent);
-        this.chunkRelationHelper.emitCurrent();
+
+        try {
+            // TODO Michael W: only fetch events if needed, blocked by PSF-1708
+            await this.chunkRelationHelper.emitFetchCurrent();
+        } catch (err) {
+            logger.warn("error fetching server side relation for voice broadcast chunks", err);
+            // fall back to local events
+            this.chunkRelationHelper.emitCurrent();
+        }
     }
 
     private addChunkEvent = async (event: MatrixEvent): Promise<boolean> => {
@@ -150,23 +169,18 @@ export class VoiceBroadcastPlayback
         this.setInfoState(state);
     };
 
-    private async loadChunks(): Promise<void> {
-        const relations = getReferenceRelationsForEvent(this.infoEvent, EventType.RoomMessage, this.client);
-        const chunkEvents = relations?.getRelations();
-
-        if (!chunkEvents) {
-            return;
-        }
+    private async enqueueChunks(): Promise<void> {
+        const promises = this.chunkEvents.getEvents().reduce((promises, event: MatrixEvent) => {
+            if (!this.playbacks.has(event.getId() || "")) {
+                promises.push(this.enqueueChunk(event));
+            }
+            return promises;
+        }, [] as Promise<void>[]);
 
-        this.chunkEvents.addEvents(chunkEvents);
-        this.setDuration(this.chunkEvents.getLength());
-
-        for (const chunkEvent of chunkEvents) {
-            await this.enqueueChunk(chunkEvent);
-        }
+        await Promise.all(promises);
     }
 
-    private async enqueueChunk(chunkEvent: MatrixEvent) {
+    private async enqueueChunk(chunkEvent: MatrixEvent): Promise<void> {
         const eventId = chunkEvent.getId();
 
         if (!eventId) {
@@ -317,10 +331,7 @@ export class VoiceBroadcastPlayback
     }
 
     public async start(): Promise<void> {
-        if (this.playbacks.size === 0) {
-            await this.loadChunks();
-        }
-
+        await this.enqueueChunks();
         const chunkEvents = this.chunkEvents.getEvents();
 
         const toPlay = this.getInfoState() === VoiceBroadcastInfoState.Stopped

test/@types/common.ts

@@ -0,0 +1,19 @@
+/*
+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 type PublicInterface<T> = {
+    [P in keyof T]: T[P];
+};

test/events/RelationsHelper-test.ts

@@ -28,13 +28,15 @@ import { Relations } from "matrix-js-sdk/src/models/relations";
 import { RelationsContainer } from "matrix-js-sdk/src/models/relations-container";
 
 import { RelationsHelper, RelationsHelperEvent } from "../../src/events/RelationsHelper";
-import { mkEvent, mkStubRoom, stubClient } from "../test-utils";
+import { mkEvent, mkRelationsContainer, mkStubRoom, stubClient } from "../test-utils";
 
 describe("RelationsHelper", () => {
     const roomId = "!room:example.com";
+    let userId: string;
     let event: MatrixEvent;
     let relatedEvent1: MatrixEvent;
     let relatedEvent2: MatrixEvent;
+    let relatedEvent3: MatrixEvent;
     let room: Room;
     let client: MatrixClient;
     let relationsHelper: RelationsHelper;
@@ -46,47 +48,81 @@ describe("RelationsHelper", () => {
 
     beforeEach(() => {
         client = stubClient();
+        userId = client.getUserId() || "";
+        mocked(client.relations).mockClear();
         room = mkStubRoom(roomId, "test room", client);
-        mocked(client.getRoom).mockImplementation((getRoomId: string) => {
+        mocked(client.getRoom).mockImplementation((getRoomId?: string) => {
             if (getRoomId === roomId) {
                 return room;
             }
+
+            return null;
         });
         event = mkEvent({
             event: true,
             type: EventType.RoomMessage,
             room: roomId,
-            user: client.getUserId(),
+            user: userId,
             content: {},
         });
         relatedEvent1 = mkEvent({
             event: true,
             type: EventType.RoomMessage,
             room: roomId,
-            user: client.getUserId(),
-            content: {},
+            user: userId,
+            content: { relatedEvent: 1 },
         });
         relatedEvent2 = mkEvent({
             event: true,
             type: EventType.RoomMessage,
             room: roomId,
-            user: client.getUserId(),
-            content: {},
+            user: userId,
+            content: { relatedEvent: 2 },
+        });
+        relatedEvent3 = mkEvent({
+            event: true,
+            type: EventType.RoomMessage,
+            room: roomId,
+            user: userId,
+            content: { relatedEvent: 3 },
         });
         onAdd = jest.fn();
+        relationsContainer = mkRelationsContainer();
         // TODO Michael W: create test utils, remove casts
-        relationsContainer = {
-            getChildEventsForEvent: jest.fn(),
-        } as unknown as RelationsContainer;
         relations = {
             getRelations: jest.fn(),
             on: jest.fn().mockImplementation((type, l) => relationsOnAdd = l),
+            off: jest.fn(),
         } as unknown as Relations;
         timelineSet = {
             relations: relationsContainer,
         } as unknown as EventTimelineSet;
     });
 
+    afterEach(() => {
+        relationsHelper?.destroy();
+    });
+
+    describe("when there is an event without ID", () => {
+        it("should raise an error", () => {
+            jest.spyOn(event, "getId").mockReturnValue(undefined);
+
+            expect(() => {
+                new RelationsHelper(event, RelationType.Reference, EventType.RoomMessage, client);
+            }).toThrowError("unable to create RelationsHelper: missing event ID");
+        });
+    });
+
+    describe("when there is an event without room ID", () => {
+        it("should raise an error", () => {
+            jest.spyOn(event, "getRoomId").mockReturnValue(undefined);
+
+            expect(() => {
+                new RelationsHelper(event, RelationType.Reference, EventType.RoomMessage, client);
+            }).toThrowError("unable to create RelationsHelper: missing room ID");
+        });
+    });
+
     describe("when there is an event without relations", () => {
         beforeEach(() => {
             relationsHelper = new RelationsHelper(event, RelationType.Reference, EventType.RoomMessage, client);
@@ -118,6 +154,34 @@ describe("RelationsHelper", () => {
         });
     });
 
+    describe("when there is an event with two pages server side relations", () => {
+        beforeEach(() => {
+            mocked(client.relations)
+                .mockResolvedValueOnce({
+                    events: [relatedEvent1, relatedEvent2],
+                    nextBatch: "next",
+                })
+                .mockResolvedValueOnce({
+                    events: [relatedEvent3],
+                    nextBatch: null,
+                });
+            relationsHelper = new RelationsHelper(event, RelationType.Reference, EventType.RoomMessage, client);
+            relationsHelper.on(RelationsHelperEvent.Add, onAdd);
+        });
+
+        describe("emitFetchCurrent", () => {
+            beforeEach(async () => {
+                await relationsHelper.emitFetchCurrent();
+            });
+
+            it("should emit the server side events", () => {
+                expect(onAdd).toHaveBeenCalledWith(relatedEvent1);
+                expect(onAdd).toHaveBeenCalledWith(relatedEvent2);
+                expect(onAdd).toHaveBeenCalledWith(relatedEvent3);
+            });
+        });
+    });
+
     describe("when there is an event with relations", () => {
         beforeEach(() => {
             mocked(room.getUnfilteredTimelineSet).mockReturnValue(timelineSet);

test/test-utils/audio.ts

@@ -20,10 +20,7 @@ import { SimpleObservable } from "matrix-widget-api";
 import { Playback, PlaybackState } from "../../src/audio/Playback";
 import { PlaybackClock } from "../../src/audio/PlaybackClock";
 import { UPDATE_EVENT } from "../../src/stores/AsyncStore";
-
-type PublicInterface<T> = {
-    [P in keyof T]: T[P];
-};
+import { PublicInterface } from "../@types/common";
 
 export const createTestPlayback = (): Playback => {
     const eventEmitter = new EventEmitter();

test/test-utils/index.ts

@@ -25,3 +25,4 @@ export * from './call';
 export * from './wrappers';
 export * from './utilities';
 export * from './date';
+export * from './relations';

test/test-utils/relations.ts

@@ -0,0 +1,35 @@
+/*
+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 { Relations } from "matrix-js-sdk/src/models/relations";
+import { RelationsContainer } from "matrix-js-sdk/src/models/relations-container";
+
+import { PublicInterface } from "../@types/common";
+
+export const mkRelations = (): Relations => {
+    return {
+
+    } as PublicInterface<Relations> as Relations;
+};
+
+export const mkRelationsContainer = (): RelationsContainer => {
+    return {
+        aggregateChildEvent: jest.fn(),
+        aggregateParentEvent: jest.fn(),
+        getAllChildEventsForEvent: jest.fn(),
+        getChildEventsForEvent: jest.fn(),
+    } as PublicInterface<RelationsContainer> as RelationsContainer;
+};