Live location sharing - Aggregate beacon locations on beacons (#2268)

* add timestamp sorting util

Signed-off-by: Kerry Archibald <kerrya@element.io>

* basic wiring

Signed-off-by: Kerry Archibald <kerrya@element.io>

* quick handle for redacted beacons

Signed-off-by: Kerry Archibald <kerrya@element.io>

* remove fdescribe

Signed-off-by: Kerry Archibald <kerrya@element.io>

* test adding locations

Signed-off-by: Kerry Archibald <kerrya@element.io>

* tidy comments

Signed-off-by: Kerry Archibald <kerrya@element.io>

* test client

Signed-off-by: Kerry Archibald <kerrya@element.io>

* fix monitorLiveness for update

Signed-off-by: Kerry Archibald <kerrya@element.io>

* lint

Signed-off-by: Kerry Archibald <kerrya@element.io>

spec/unit/matrix-client.spec.ts

@@ -33,6 +33,7 @@ import * as testUtils from "../test-utils/test-utils";
 import { makeBeaconInfoContent } from "../../src/content-helpers";
 import { M_BEACON_INFO } from "../../src/@types/beacon";
 import { Room } from "../../src";
+import { makeBeaconEvent } from "../test-utils/beacon";
 
 jest.useFakeTimers();
 
@@ -1025,5 +1026,34 @@ describe("MatrixClient", function() {
             );
             expect(requestContent).toEqual(content);
         });
+
+        describe('processBeaconEvents()', () => {
+            it('does nothing when events is falsy', () => {
+                const room = new Room(roomId, client, userId);
+                const roomStateProcessSpy = jest.spyOn(room.currentState, 'processBeaconEvents');
+
+                client.processBeaconEvents(room, undefined);
+                expect(roomStateProcessSpy).not.toHaveBeenCalled();
+            });
+
+            it('does nothing when events is of length 0', () => {
+                const room = new Room(roomId, client, userId);
+                const roomStateProcessSpy = jest.spyOn(room.currentState, 'processBeaconEvents');
+
+                client.processBeaconEvents(room, []);
+                expect(roomStateProcessSpy).not.toHaveBeenCalled();
+            });
+
+            it('calls room states processBeaconEvents with m.beacon events', () => {
+                const room = new Room(roomId, client, userId);
+                const roomStateProcessSpy = jest.spyOn(room.currentState, 'processBeaconEvents');
+
+                const messageEvent = testUtils.mkMessage({ room: roomId, user: userId, event: true });
+                const beaconEvent = makeBeaconEvent(userId);
+
+                client.processBeaconEvents(room, [messageEvent, beaconEvent]);
+                expect(roomStateProcessSpy).toHaveBeenCalledWith([beaconEvent]);
+            });
+        });
     });
 });

spec/unit/models/beacon.spec.ts

@@ -19,7 +19,7 @@ import {
     Beacon,
     BeaconEvent,
 } from "../../../src/models/beacon";
-import { makeBeaconInfoEvent } from "../../test-utils/beacon";
+import { makeBeaconEvent, makeBeaconInfoEvent } from "../../test-utils/beacon";
 
 jest.useFakeTimers();
 
@@ -282,5 +282,93 @@ describe('Beacon', () => {
                 expect(emitSpy).toHaveBeenCalledTimes(1);
             });
         });
+
+        describe('addLocations', () => {
+            it('ignores locations when beacon is not live', () => {
+                const beacon = new Beacon(makeBeaconInfoEvent(userId, roomId, { isLive: false }));
+                const emitSpy = jest.spyOn(beacon, 'emit');
+
+                beacon.addLocations([
+                    makeBeaconEvent(userId, { beaconInfoId: beacon.beaconInfoId, timestamp: now + 1 }),
+                ]);
+
+                expect(beacon.latestLocationState).toBeFalsy();
+                expect(emitSpy).not.toHaveBeenCalled();
+            });
+
+            it('ignores locations outside the beacon live duration', () => {
+                const beacon = new Beacon(makeBeaconInfoEvent(userId, roomId, { isLive: true, timeout: 60000 }));
+                const emitSpy = jest.spyOn(beacon, 'emit');
+
+                beacon.addLocations([
+                    // beacon has now + 60000 live period
+                    makeBeaconEvent(userId, { beaconInfoId: beacon.beaconInfoId, timestamp: now + 100000 }),
+                ]);
+
+                expect(beacon.latestLocationState).toBeFalsy();
+                expect(emitSpy).not.toHaveBeenCalled();
+            });
+
+            it('sets latest location state to most recent location', () => {
+                const beacon = new Beacon(makeBeaconInfoEvent(userId, roomId, { isLive: true, timeout: 60000 }));
+                const emitSpy = jest.spyOn(beacon, 'emit');
+
+                const locations = [
+                    // older
+                    makeBeaconEvent(
+                        userId, { beaconInfoId: beacon.beaconInfoId, uri: 'geo:foo', timestamp: now + 1 },
+                    ),
+                    // newer
+                    makeBeaconEvent(
+                        userId, { beaconInfoId: beacon.beaconInfoId, uri: 'geo:bar', timestamp: now + 10000 },
+                    ),
+                    // not valid
+                    makeBeaconEvent(
+                        userId, { beaconInfoId: beacon.beaconInfoId, uri: 'geo:baz', timestamp: now - 5 },
+                    ),
+                ];
+
+                beacon.addLocations(locations);
+
+                const expectedLatestLocation = {
+                    description: undefined,
+                    timestamp: now + 10000,
+                    uri: 'geo:bar',
+                };
+
+                // the newest valid location
+                expect(beacon.latestLocationState).toEqual(expectedLatestLocation);
+                expect(emitSpy).toHaveBeenCalledWith(BeaconEvent.LocationUpdate, expectedLatestLocation);
+            });
+
+            it('ignores locations that are less recent that the current latest location', () => {
+                const beacon = new Beacon(makeBeaconInfoEvent(userId, roomId, { isLive: true, timeout: 60000 }));
+
+                const olderLocation = makeBeaconEvent(
+                    userId, { beaconInfoId: beacon.beaconInfoId, uri: 'geo:foo', timestamp: now + 1 },
+                );
+                const newerLocation = makeBeaconEvent(
+                    userId, { beaconInfoId: beacon.beaconInfoId, uri: 'geo:bar', timestamp: now + 10000 },
+                );
+
+                beacon.addLocations([newerLocation]);
+                // latest location set to newerLocation
+                expect(beacon.latestLocationState).toEqual(expect.objectContaining({
+                    uri: 'geo:bar',
+                }));
+
+                const emitSpy = jest.spyOn(beacon, 'emit').mockClear();
+
+                // add older location
+                beacon.addLocations([olderLocation]);
+
+                // no change
+                expect(beacon.latestLocationState).toEqual(expect.objectContaining({
+                    uri: 'geo:bar',
+                }));
+                // no emit
+                expect(emitSpy).not.toHaveBeenCalled();
+            });
+        });
     });
 });

spec/unit/room-state.spec.js

@@ -1,5 +1,5 @@
 import * as utils from "../test-utils/test-utils";
-import { makeBeaconInfoEvent } from "../test-utils/beacon";
+import { makeBeaconEvent, makeBeaconInfoEvent } from "../test-utils/beacon";
 import { filterEmitCallsByEventType } from "../test-utils/emitter";
 import { RoomState, RoomStateEvent } from "../../src/models/room-state";
 import { BeaconEvent, getBeaconInfoIdentifier } from "../../src/models/beacon";
@@ -712,4 +712,57 @@ describe("RoomState", function() {
             expect(state.maySendEvent('m.room.other_thing', userB)).toEqual(false);
         });
     });
+
+    describe('processBeaconEvents', () => {
+        const beacon1 = makeBeaconInfoEvent(userA, roomId, {}, '$beacon1', '$beacon1');
+        const beacon2 = makeBeaconInfoEvent(userB, roomId, {}, '$beacon2', '$beacon2');
+
+        it('does nothing when state has no beacons', () => {
+            const emitSpy = jest.spyOn(state, 'emit');
+            state.processBeaconEvents([makeBeaconEvent(userA, { beaconInfoId: '$beacon1' })]);
+            expect(emitSpy).not.toHaveBeenCalled();
+        });
+
+        it('does nothing when there are no events', () => {
+            state.setStateEvents([beacon1, beacon2]);
+            const emitSpy = jest.spyOn(state, 'emit').mockClear();
+            state.processBeaconEvents([]);
+            expect(emitSpy).not.toHaveBeenCalled();
+        });
+
+        it('discards events for beacons that are not in state', () => {
+            const location = makeBeaconEvent(userA, {
+                beaconInfoId: 'some-other-beacon',
+            });
+            state.setStateEvents([beacon1, beacon2]);
+            const emitSpy = jest.spyOn(state, 'emit').mockClear();
+            state.processBeaconEvents([location]);
+            expect(emitSpy).not.toHaveBeenCalled();
+        });
+
+        it('adds locations to beacons', () => {
+            const location1 = makeBeaconEvent(userA, {
+                beaconInfoId: '$beacon1', timestamp: Date.now() + 1,
+            });
+            const location2 = makeBeaconEvent(userA, {
+                beaconInfoId: '$beacon1', timestamp: Date.now() + 2,
+            });
+            const location3 = makeBeaconEvent(userB, {
+                beaconInfoId: 'some-other-beacon',
+            });
+
+            state.setStateEvents([beacon1, beacon2]);
+
+            expect(state.beacons.size).toEqual(2);
+
+            const beaconInstance = state.beacons.get(getBeaconInfoIdentifier(beacon1));
+            const addLocationsSpy = jest.spyOn(beaconInstance, 'addLocations');
+
+            state.processBeaconEvents([location1, location2, location3]);
+
+            expect(addLocationsSpy).toHaveBeenCalledTimes(1);
+            // only called with locations for beacon1
+            expect(addLocationsSpy).toHaveBeenCalledWith([location1, location2]);
+        });
+    });
 });

spec/unit/utils.spec.ts

@@ -10,8 +10,11 @@ import {
     prevString,
     simpleRetryOperation,
     stringToBase,
+    sortEventsByLatestContentTimestamp,
 } from "../../src/utils";
 import { logger } from "../../src/logger";
+import { mkMessage } from "../test-utils/test-utils";
+import { makeBeaconEvent } from "../test-utils/beacon";
 
 // TODO: Fix types throughout
 
@@ -506,4 +509,30 @@ describe("utils", function() {
             });
         });
     });
+
+    describe('sortEventsByLatestContentTimestamp', () => {
+        const roomId = '!room:server';
+        const userId = '@user:server';
+        const eventWithoutContentTimestamp = mkMessage({ room: roomId, user: userId, event: true });
+        // m.beacon events have timestamp in content
+        const beaconEvent1 = makeBeaconEvent(userId, { timestamp: 1648804528557 });
+        const beaconEvent2 = makeBeaconEvent(userId, { timestamp: 1648804528558 });
+        const beaconEvent3 = makeBeaconEvent(userId, { timestamp: 1648804528000 });
+        const beaconEvent4 = makeBeaconEvent(userId, { timestamp: 0 });
+
+        it('sorts events with timestamps as later than events without', () => {
+            expect(
+                [beaconEvent4, eventWithoutContentTimestamp, beaconEvent1]
+                    .sort(utils.sortEventsByLatestContentTimestamp),
+            ).toEqual([
+                beaconEvent1, beaconEvent4, eventWithoutContentTimestamp,
+            ]);
+        });
+
+        it('sorts by content timestamps correctly', () => {
+            expect(
+                [beaconEvent1, beaconEvent2, beaconEvent3].sort(sortEventsByLatestContentTimestamp),
+            ).toEqual([beaconEvent2, beaconEvent1, beaconEvent3]);
+        });
+    });
 });

src/client.ts

@@ -180,7 +180,7 @@ import { MediaHandler } from "./webrtc/mediaHandler";
 import { IRefreshTokenResponse } from "./@types/auth";
 import { TypedEventEmitter } from "./models/typed-event-emitter";
 import { Thread, THREAD_RELATION_TYPE } from "./models/thread";
-import { MBeaconInfoEventContent, M_BEACON_INFO } from "./@types/beacon";
+import { MBeaconInfoEventContent, M_BEACON, M_BEACON_INFO } from "./@types/beacon";
 
 export type Store = IStore;
 export type SessionStore = WebStorageSessionStore;
@@ -5169,6 +5169,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
 
                 const [timelineEvents, threadedEvents] = room.partitionThreadedEvents(matrixEvents);
 
+                this.processBeaconEvents(room, matrixEvents);
                 room.addEventsToTimeline(timelineEvents, true, room.getLiveTimeline());
                 await this.processThreadEvents(room, threadedEvents, true);
 
@@ -5308,6 +5309,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
         timelineSet.addEventsToTimeline(timelineEvents, true, timeline, res.start);
         // The target event is not in a thread but process the contextual events, so we can show any threads around it.
         await this.processThreadEvents(timelineSet.room, threadedEvents, true);
+        this.processBeaconEvents(timelineSet.room, events);
 
         // There is no guarantee that the event ended up in "timeline" (we might have switched to a neighbouring
         // timeline) - so check the room's index again. On the other hand, there's no guarantee the event ended up
@@ -5438,6 +5440,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
                 // in the notification timeline set
                 const timelineSet = eventTimeline.getTimelineSet();
                 timelineSet.addEventsToTimeline(matrixEvents, backwards, eventTimeline, token);
+                this.processBeaconEvents(timelineSet.room, matrixEvents);
 
                 // if we've hit the end of the timeline, we need to stop trying to
                 // paginate. We need to keep the 'forwards' token though, to make sure
@@ -5474,6 +5477,7 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
                 const timelineSet = eventTimeline.getTimelineSet();
                 const [timelineEvents, threadedEvents] = timelineSet.room.partitionThreadedEvents(matrixEvents);
                 timelineSet.addEventsToTimeline(timelineEvents, backwards, eventTimeline, token);
+                this.processBeaconEvents(timelineSet.room, matrixEvents);
                 await this.processThreadEvents(room, threadedEvents, backwards);
 
                 // if we've hit the end of the timeline, we need to stop trying to
@@ -8851,6 +8855,17 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
         await room.processThreadedEvents(threadedEvents, toStartOfTimeline);
     }
 
+    public processBeaconEvents(
+        room: Room,
+        events?: MatrixEvent[],
+    ): void {
+        if (!events?.length) {
+            return;
+        }
+        const beaconEvents = events.filter(event => M_BEACON.matches(event.getType()));
+        room.currentState.processBeaconEvents(beaconEvents);
+    }
+
     /**
      * Fetches the user_id of the configured access token.
      */

src/content-helpers.ts

@@ -261,3 +261,18 @@ export const makeBeaconContent: MakeBeaconContent = (
         event_id: beaconInfoEventId,
     },
 });
+
+export type BeaconLocationState = MLocationContent & {
+    timestamp: number;
+};
+
+export const parseBeaconContent = (content: MBeaconEventContent): BeaconLocationState => {
+    const { description, uri } = M_LOCATION.findIn<MLocationContent>(content);
+    const timestamp = M_TIMESTAMP.findIn<number>(content);
+
+    return {
+        description,
+        uri,
+        timestamp,
+    };
+};

src/models/beacon.ts

@@ -14,21 +14,27 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import { BeaconInfoState, parseBeaconInfoContent } from "../content-helpers";
+import { MBeaconEventContent } from "../@types/beacon";
+import { M_TIMESTAMP } from "../@types/location";
+import { BeaconInfoState, BeaconLocationState, parseBeaconContent, parseBeaconInfoContent } from "../content-helpers";
 import { MatrixEvent } from "../matrix";
+import { sortEventsByLatestContentTimestamp } from "../utils";
 import { TypedEventEmitter } from "./typed-event-emitter";
 
 export enum BeaconEvent {
     New = "Beacon.new",
     Update = "Beacon.update",
     LivenessChange = "Beacon.LivenessChange",
-    Destroy = "Destroy",
+    Destroy = "Beacon.Destroy",
+    LocationUpdate = "Beacon.LocationUpdate",
 }
 
 export type BeaconEventHandlerMap = {
     [BeaconEvent.Update]: (event: MatrixEvent, beacon: Beacon) => void;
     [BeaconEvent.LivenessChange]: (isLive: boolean, beacon: Beacon) => void;
     [BeaconEvent.Destroy]: (beaconIdentifier: string) => void;
+    [BeaconEvent.LocationUpdate]: (locationState: BeaconLocationState) => void;
+    [BeaconEvent.Destroy]: (beaconIdentifier: string) => void;
 };
 
 export const isTimestampInDuration = (
@@ -49,6 +55,7 @@ export class Beacon extends TypedEventEmitter<Exclude<BeaconEvent, BeaconEvent.N
     private _beaconInfo: BeaconInfoState;
     private _isLive: boolean;
     private livenessWatchInterval: number;
+    private _latestLocationState: BeaconLocationState | undefined;
 
     constructor(
         private rootEvent: MatrixEvent,
@@ -82,6 +89,10 @@ export class Beacon extends TypedEventEmitter<Exclude<BeaconEvent, BeaconEvent.N
         return this._beaconInfo;
     }
 
+    public get latestLocationState(): BeaconLocationState | undefined {
+        return this._latestLocationState;
+    }
+
     public update(beaconInfoEvent: MatrixEvent): void {
         if (getBeaconInfoIdentifier(beaconInfoEvent) !== this.identifier) {
             throw new Error('Invalid updating event');
@@ -94,6 +105,7 @@ export class Beacon extends TypedEventEmitter<Exclude<BeaconEvent, BeaconEvent.N
         this.setBeaconInfo(this.rootEvent);
 
         this.emit(BeaconEvent.Update, beaconInfoEvent, this);
+        this.clearLatestLocation();
     }
 
     public destroy(): void {
@@ -114,14 +126,51 @@ export class Beacon extends TypedEventEmitter<Exclude<BeaconEvent, BeaconEvent.N
             clearInterval(this.livenessWatchInterval);
         }
 
+        this.checkLiveness();
         if (this.isLive) {
-            const expiryInMs = (this._beaconInfo?.timestamp + this._beaconInfo?.timeout + 1) - Date.now();
+            const expiryInMs = (this._beaconInfo?.timestamp + this._beaconInfo?.timeout) - Date.now();
             if (expiryInMs > 1) {
-                this.livenessWatchInterval = setInterval(this.checkLiveness.bind(this), expiryInMs);
+                this.livenessWatchInterval = setInterval(
+                    () => { this.monitorLiveness(); },
+                    expiryInMs,
+                );
             }
         }
     }
 
+    /**
+     * Process Beacon locations
+     * Emits BeaconEvent.LocationUpdate
+     */
+    public addLocations(beaconLocationEvents: MatrixEvent[]): void {
+        // discard locations for beacons that are not live
+        if (!this.isLive) {
+            return;
+        }
+
+        const validLocationEvents = beaconLocationEvents.filter(event => {
+            const content = event.getContent<MBeaconEventContent>();
+            const timestamp = M_TIMESTAMP.findIn<number>(content);
+            return (
+                // only include positions that were taken inside the beacon's live period
+                isTimestampInDuration(this._beaconInfo.timestamp, this._beaconInfo.timeout, timestamp) &&
+                // ignore positions older than our current latest location
+                (!this.latestLocationState || timestamp > this.latestLocationState.timestamp)
+            );
+        });
+        const latestLocationEvent = validLocationEvents.sort(sortEventsByLatestContentTimestamp)?.[0];
+
+        if (latestLocationEvent) {
+            this._latestLocationState = parseBeaconContent(latestLocationEvent.getContent());
+            this.emit(BeaconEvent.LocationUpdate, this.latestLocationState);
+        }
+    }
+
+    private clearLatestLocation = () => {
+        this._latestLocationState = undefined;
+        this.emit(BeaconEvent.LocationUpdate, this.latestLocationState);
+    };
+
     private setBeaconInfo(event: MatrixEvent): void {
         this._beaconInfo = parseBeaconInfoContent(event.getContent());
         this.checkLiveness();

src/models/room-state.ts

@@ -407,6 +407,37 @@ export class RoomState extends TypedEventEmitter<EmittedEvents, EventHandlerMap>
         this.emit(RoomStateEvent.Update, this);
     }
 
+    public processBeaconEvents(events: MatrixEvent[]): void {
+        if (
+            !events.length ||
+            // discard locations if we have no beacons
+            !this.beacons.size
+        ) {
+            return;
+        }
+
+        // names are confusing here
+        // a Beacon is the parent event, but event type is 'm.beacon_info'
+        // a location is the 'child' related to the Beacon, but the event type is 'm.beacon'
+        // group locations by beaconInfo event id
+        const locationEventsByBeaconEventId = events.reduce<Record<string, MatrixEvent[]>>((acc, event) => {
+            const beaconInfoEventId = event.getRelation()?.event_id;
+            if (!acc[beaconInfoEventId]) {
+                acc[beaconInfoEventId] = [];
+            }
+            acc[beaconInfoEventId].push(event);
+            return acc;
+        }, {});
+
+        Object.entries(locationEventsByBeaconEventId).forEach(([beaconInfoEventId, events]) => {
+            const beacon = [...this.beacons.values()].find(beacon => beacon.beaconInfoId === beaconInfoEventId);
+
+            if (beacon) {
+                beacon.addLocations(events);
+            }
+        });
+    }
+
     /**
      * Looks up a member by the given userId, and if it doesn't exist,
      * create it and emit the `RoomState.newMember` event.
@@ -441,6 +472,7 @@ export class RoomState extends TypedEventEmitter<EmittedEvents, EventHandlerMap>
      */
     private setBeacon(event: MatrixEvent): void {
         const beaconIdentifier = getBeaconInfoIdentifier(event);
+
         if (this.beacons.has(beaconIdentifier)) {
             const beacon = this.beacons.get(beaconIdentifier);
 
@@ -470,6 +502,7 @@ export class RoomState extends TypedEventEmitter<EmittedEvents, EventHandlerMap>
 
         this.emit(BeaconEvent.New, event, beacon);
         beacon.on(BeaconEvent.LivenessChange, this.onBeaconLivenessChange.bind(this));
+
         this.beacons.set(beacon.identifier, beacon);
     }
 

src/sync.ts

@@ -1636,6 +1636,7 @@ export class SyncApi {
         // This also needs to be done before running push rules on the events as they need
         // to be decorated with sender etc.
         room.addLiveEvents(timelineEventList || [], null, fromCache);
+        this.client.processBeaconEvents(room, timelineEventList);
     }
 
     /**

src/utils.ts

@@ -24,6 +24,8 @@ import unhomoglyph from "unhomoglyph";
 import promiseRetry from "p-retry";
 
 import type NodeCrypto from "crypto";
+import { MatrixEvent } from ".";
+import { M_TIMESTAMP } from "./@types/location";
 
 /**
  * Encode a dictionary of query parameters.
@@ -708,3 +710,15 @@ export function recursivelyAssign(target: Object, source: Object, ignoreNullish
     }
     return target;
 }
+
+function getContentTimestampWithFallback(event: MatrixEvent): number {
+    return M_TIMESTAMP.findIn<number>(event.getContent()) ?? -1;
+}
+
+/**
+ * Sort events by their content m.ts property
+ * Latest timestamp first
+ */
+export function sortEventsByLatestContentTimestamp(left: MatrixEvent, right: MatrixEvent): number {
+    return getContentTimestampWithFallback(right) - getContentTimestampWithFallback(left);
+}