Merge branch 'bwindels/calls' into thirdroom/dev

.eslintrc.js

@@ -17,6 +17,7 @@ module.exports = {
     "globals": {
         "DEFINE_VERSION": "readonly",
         "DEFINE_GLOBAL_HASH": "readonly",
+        "DEFINE_PROJECT_DIR": "readonly",
         // only available in sw.js
         "DEFINE_UNHASHED_PRECACHED_ASSETS": "readonly",
         "DEFINE_HASHED_PRECACHED_ASSETS": "readonly",

doc/SDK.md

@@ -31,7 +31,8 @@ import {
     createNavigation,
     createRouter,
     RoomViewModel,
-    TimelineView
+    TimelineView,
+    viewClassForTile
 } from "hydrogen-view-sdk";
 import downloadSandboxPath from 'hydrogen-view-sdk/download-sandbox.html?url';
 import workerPath from 'hydrogen-view-sdk/main.js?url';
@@ -53,7 +54,7 @@ import "hydrogen-view-sdk/theme-element-light.css";
 async function main() {
     const app = document.querySelector<HTMLDivElement>('#app')!
     const config = {};
-    const platform = new Platform(app, assetPaths, config, { development: import.meta.env.DEV });
+    const platform = new Platform({container: app, assetPaths, config, options: { development: import.meta.env.DEV }});
     const navigation = createNavigation();
     platform.setNavigation(navigation);
     const urlRouter = createRouter({
@@ -88,7 +89,7 @@ async function main() {
             navigation,
         });
         await vm.load();
-        const view = new TimelineView(vm.timelineViewModel);
+        const view = new TimelineView(vm.timelineViewModel, viewClassForTile);
         app.appendChild(view.mount());
     }
 }

src/domain/ViewModel.ts

@@ -40,9 +40,9 @@ export type Options = {
 export class ViewModel<O extends Options = Options> extends EventEmitter<{change: never}> {
     private disposables?: Disposables;
     private _isDisposed = false;
-    private _options: O;
+    private _options: Readonly<O>;
 
-    constructor(options: O) {
+    constructor(options: Readonly<O>) {
         super();
         this._options = options;
     }
@@ -51,7 +51,7 @@ export class ViewModel<O extends Options = Options> extends EventEmitter<{change
         return Object.assign({}, this._options, explicitOptions);
     }
 
-    get options(): O { return this._options; }
+    get options(): Readonly<O> { return this._options; }
 
     // makes it easier to pass through dependencies of a sub-view model
     getOption<N extends keyof O>(name: N): O[N]  {
@@ -115,10 +115,6 @@ export class ViewModel<O extends Options = Options> extends EventEmitter<{change
         return result;
     }
 
-    updateOptions(options: O): void {
-        this._options = Object.assign(this._options, options);
-    }
-
     emitChange(changedProps: any): void {
         if (this._options.emitChange) {
             this._options.emitChange(changedProps);

src/domain/session/room/timeline/tiles/SimpleTile.js

@@ -22,6 +22,7 @@ export class SimpleTile extends ViewModel {
     constructor(entry, options) {
         super(options);
         this._entry = entry;
+        this._emitUpdate = undefined;
     }
     // view model props for all subclasses
     // hmmm, could also do instanceof ... ?
@@ -67,16 +68,20 @@ export class SimpleTile extends ViewModel {
 
     // TilesCollection contract below
     setUpdateEmit(emitUpdate) {
-        this.updateOptions({emitChange: paramName => {
+        this._emitUpdate = emitUpdate;
+    }
+
+    /** overrides the emitChange in ViewModel to also emit the update over the tiles collection */
+    emitChange(changedProps) {
+        if (this._emitUpdate) {
             // it can happen that after some network call
             // we switched away from the room and the response
             // comes in, triggering an emitChange in a tile that
             // has been disposed already (and hence the change
             // callback has been cleared by dispose) We should just ignore this.
-            if (emitUpdate) {
-                emitUpdate(this, paramName);
-            }
-        }});
+            this._emitUpdate(this, changedProps);
+        }
+        super.emitChange(changedProps);
     }
 
     get upperEntry() {

src/lib.ts

@@ -15,12 +15,20 @@ limitations under the License.
 */
 
 export {Logger} from "./logging/Logger";
+export type {ILogItem} from "./logging/types";
 export {IDBLogPersister} from "./logging/IDBLogPersister";
 export {ConsoleReporter} from "./logging/ConsoleReporter";
 export {Platform} from "./platform/web/Platform.js";
 export {Client, LoadStatus} from "./matrix/Client.js";
 export {RoomStatus} from "./matrix/room/common";
 export {CallIntent} from "./matrix/calls/callEventTypes";
+// export everything needed to observe state events on all rooms using session.observeRoomState
+export type {RoomStateHandler} from "./matrix/room/state/types";
+export type {MemberChange} from "./matrix/room/members/RoomMember";
+export type {Transaction} from "./matrix/storage/idb/Transaction";
+export type {Room} from "./matrix/room/Room";
+export type {StateEvent} from "./matrix/storage/types";
+
 // export main view & view models
 export {createNavigation, createRouter} from "./domain/navigation/index.js";
 export {RootViewModel} from "./domain/RootViewModel.js";

src/matrix/Client.js

@@ -132,14 +132,15 @@ export class Client {
         });
     }
 
-    async startRegistration(homeserver, username, password, initialDeviceDisplayName) {
+    async startRegistration(homeserver, username, password, initialDeviceDisplayName, flowSelector) {
         const request = this._platform.request;
         const hsApi = new HomeServerApi({homeserver, request});
         const registration = new Registration(hsApi, {
             username, 
             password,
             initialDeviceDisplayName,
-        });
+        },
+        flowSelector);
         return registration;
     }
 

src/matrix/Session.js

@@ -48,6 +48,7 @@ import {SecretStorage} from "./ssss/SecretStorage";
 import {ObservableValue} from "../observable/value/ObservableValue";
 import {RetainedObservableValue} from "../observable/value/RetainedObservableValue";
 import {CallHandler} from "./calls/CallHandler";
+import {RoomStateHandlerSet} from "./room/state/RoomStateHandlerSet";
 
 const PICKLE_KEY = "DEFAULT_KEY";
 const PUSHER_KEY = "pusher";
@@ -101,6 +102,8 @@ export class Session {
             }],
             forceTURN: false,
         });
+        this._roomStateHandler = new RoomStateHandlerSet();
+        this.observeRoomState(this._callHandler);
         this._deviceMessageHandler = new DeviceMessageHandler({storage, callHandler: this._callHandler});
         this._olm = olm;
         this._olmUtil = null;
@@ -595,7 +598,7 @@ export class Session {
             user: this._user,
             createRoomEncryption: this._createRoomEncryption,
             platform: this._platform,
-            callHandler: this._callHandler
+            roomStateHandler: this._roomStateHandler
         });
     }
 
@@ -937,6 +940,10 @@ export class Session {
         return observable;
     }
 
+    observeRoomState(roomStateHandler) {
+        return this._roomStateHandler.subscribe(roomStateHandler);
+    }
+
     /**
     Creates an empty (summary isn't loaded) the archived room if it isn't
     loaded already, assuming sync will either remove it (when rejoining) or

src/matrix/calls/CallHandler.ts

@@ -35,6 +35,7 @@ import type {Options as GroupCallOptions} from "./group/GroupCall";
 import type {Transaction} from "../storage/idb/Transaction";
 import type {CallEntry} from "../storage/idb/stores/CallStore";
 import type {Clock} from "../../platform/web/dom/Clock";
+import type {RoomStateHandler} from "../room/state/types";
 
 export type Options = Omit<GroupCallOptions, "emitUpdate" | "createTimeout"> & {
     clock: Clock
@@ -44,7 +45,7 @@ function getRoomMemberKey(roomId: string, userId: string): string {
     return JSON.stringify(roomId)+`,`+JSON.stringify(userId);
 }
 
-export class CallHandler {
+export class CallHandler implements RoomStateHandler {
     // group calls by call id
     private readonly _calls: ObservableMap<string, GroupCall> = new ObservableMap<string, GroupCall>();
     // map of `"roomId","userId"` to set of conf_id's they are in
@@ -143,18 +144,12 @@ export class CallHandler {
     // TODO: check and poll turn server credentials here
 
     /** @internal */
-    handleRoomState(room: Room, events: StateEvent[], txn: Transaction, log: ILogItem) {
-        // first update call events
-        for (const event of events) {
-            if (event.type === EventType.GroupCall) {
-                this.handleCallEvent(event, room.id, txn, log);
-            }
+    handleRoomState(room: Room, event: StateEvent, txn: Transaction, log: ILogItem) {
+        if (event.type === EventType.GroupCall) {
+            this.handleCallEvent(event, room.id, txn, log);
         }
-        // then update members
-        for (const event of events) {
-            if (event.type === EventType.GroupCallMember) {
-                this.handleCallMemberEvent(event, room.id, log);
-            }
+        if (event.type === EventType.GroupCallMember) {
+            this.handleCallMemberEvent(event, room.id, log);
         }
     }
 

src/matrix/room/ArchivedRoom.js

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import {reduceStateEvents} from "./RoomSummary.js";
+import {iterateResponseStateEvents} from "./common";
 import {BaseRoom} from "./BaseRoom.js";
 import {RoomMember, EVENT_TYPE as MEMBER_EVENT_TYPE} from "./members/RoomMember.js";
 
@@ -173,15 +173,15 @@ export class ArchivedRoom extends BaseRoom {
 }
 
 function findKickDetails(roomResponse, ownUserId) {
-    const kickEvent = reduceStateEvents(roomResponse, (kickEvent, event) => {
+    let kickEvent;
+    iterateResponseStateEvents(roomResponse, event => {
         if (event.type === MEMBER_EVENT_TYPE) {
             // did we get kicked?
             if (event.state_key === ownUserId && event.sender !== event.state_key) {
                 kickEvent = event;
             }
         }
-        return kickEvent;
-    }, null);
+    });
     if (kickEvent) {
         return {
             // this is different from the room membership in the sync section, which can only be leave

src/matrix/room/BaseRoom.js

@@ -31,6 +31,8 @@ import {ensureLogItem} from "../../logging/utils";
 import {PowerLevels} from "./PowerLevels.js";
 import {RetainedObservableValue} from "../../observable/value/RetainedObservableValue";
 import {TimelineReader} from "./timeline/persistence/TimelineReader";
+import {ObservedStateTypeMap} from "./state/ObservedStateTypeMap";
+import {ObservedStateKeyValue} from "./state/ObservedStateKeyValue";
 
 const EVENT_ENCRYPTED_TYPE = "m.room.encrypted";
 
@@ -53,11 +55,35 @@ export class BaseRoom extends EventEmitter {
         this._getSyncToken = getSyncToken;
         this._platform = platform;
         this._observedEvents = null;
+        this._roomStateObservers = new Set();
         this._powerLevels = null;
         this._powerLevelLoading = null;
         this._observedMembers = null;
     }
 
+    async observeStateType(type, txn = undefined) {
+        const map = new ObservedStateTypeMap(type);
+        await this._addStateObserver(map, txn);
+        return map;
+    }
+
+    async observeStateTypeAndKey(type, stateKey, txn = undefined) {
+        const value = new ObservedStateKeyValue(type, stateKey);
+        await this._addStateObserver(value, txn);
+        return value;
+    }
+
+    async _addStateObserver(stateObserver, txn) {
+        if (!txn) {
+            txn = await this._storage.readTxn([this._storage.storeNames.roomState]);
+        }
+        await stateObserver.load(this.id, txn);
+        this._roomStateObservers.add(stateObserver);
+        stateObserver.setRemoveCallback(() => {
+            this._roomStateObservers.delete(stateObserver);
+        });
+    }
+
     async _eventIdsToEntries(eventIds, txn) {
         const retryEntries = [];
         await Promise.all(eventIds.map(async eventId => {

src/matrix/room/Room.js

@@ -23,6 +23,7 @@ import {WrappedError} from "../error.js"
 import {Heroes} from "./members/Heroes.js";
 import {AttachmentUpload} from "./AttachmentUpload.js";
 import {DecryptionSource} from "../e2ee/common.js";
+import {iterateResponseStateEvents} from "./common.js";
 import {PowerLevels, EVENT_TYPE as POWERLEVELS_EVENT_TYPE } from "./PowerLevels.js";
 
 const EVENT_ENCRYPTED_TYPE = "m.room.encrypted";
@@ -30,7 +31,7 @@ const EVENT_ENCRYPTED_TYPE = "m.room.encrypted";
 export class Room extends BaseRoom {
     constructor(options) {
         super(options);
-        this._callHandler = options.callHandler;
+        this._roomStateHandler = options.roomStateHandler;
         // TODO: pass pendingEvents to start like pendingOperations?
         const {pendingEvents} = options;
         const relationWriter = new RelationWriter({
@@ -179,8 +180,9 @@ export class Room extends BaseRoom {
             removedPendingEvents = await this._sendQueue.removeRemoteEchos(roomResponse.timeline.events, txn, log);
         }
         const powerLevelsEvent = this._getPowerLevelsEvent(roomResponse);
-        this._updateCallHandler(roomResponse, txn, log);
+        this._runRoomStateHandlers(roomResponse, txn, log);
         return {
+            roomResponse,
             summaryChanges,
             roomEncryption,
             newEntries,
@@ -203,7 +205,7 @@ export class Room extends BaseRoom {
         const {
             summaryChanges, newEntries, updatedEntries, newLiveKey,
             removedPendingEvents, memberChanges, powerLevelsEvent,
-            heroChanges, roomEncryption
+            heroChanges, roomEncryption, roomResponse
         } = changes;
         log.set("id", this.id);
         this._syncWriter.afterSync(newLiveKey);
@@ -217,9 +219,7 @@ export class Room extends BaseRoom {
             if (this._memberList) {
                 this._memberList.afterSync(memberChanges);
             }
-            if (this._callHandler) {
-                this._callHandler.updateRoomMembers(this, memberChanges);
-            }
+            this._roomStateHandler.updateRoomMembers(this, memberChanges);
             if (this._observedMembers) {
                 this._updateObservedMembers(memberChanges);
             }
@@ -265,6 +265,7 @@ export class Room extends BaseRoom {
         if (removedPendingEvents) {
             this._sendQueue.emitRemovals(removedPendingEvents);
         }
+        this._emitSyncRoomState(roomResponse);
     }
 
     _updateObservedMembers(memberChanges) {
@@ -277,8 +278,13 @@ export class Room extends BaseRoom {
     }
 
     _getPowerLevelsEvent(roomResponse) {
-        const isPowerlevelEvent = event => event.state_key === "" && event.type === POWERLEVELS_EVENT_TYPE;
-        const powerLevelEvent = roomResponse.timeline?.events.find(isPowerlevelEvent) ?? roomResponse.state?.events.find(isPowerlevelEvent);
+        let powerLevelEvent;
+        iterateResponseStateEvents(roomResponse, event => {
+            if(event.state_key === "" && event.type === POWERLEVELS_EVENT_TYPE) {
+                powerLevelEvent = event;
+            }
+
+        });
         return powerLevelEvent;
     }
 
@@ -447,20 +453,20 @@ export class Room extends BaseRoom {
         return this._sendQueue.pendingEvents;
     }
 
-    _updateCallHandler(roomResponse, txn, log) {
-        if (this._callHandler) {
-            const stateEvents = roomResponse.state?.events;
-            if (stateEvents?.length) {
-                this._callHandler.handleRoomState(this, stateEvents, txn, log);
+    /** global room state handlers, run during write sync step */
+    _runRoomStateHandlers(roomResponse, txn, log) {
+        iterateResponseStateEvents(roomResponse, event => {
+            this._roomStateHandler.handleRoomState(this, event, txn, log);
+        });
+    }
+
+    /** local room state observers, run during after sync step */
+    _emitSyncRoomState(roomResponse) {
+        iterateResponseStateEvents(roomResponse, event => {
+            for (const handler of this._roomStateObservers) {
+                handler.handleStateEvent(event);
             }
-            let timelineEvents = roomResponse.timeline?.events;
-            if (timelineEvents) {
-                const timelineStateEvents = timelineEvents.filter(e => typeof e.state_key === "string");
-                if (timelineEvents.length !== 0) {
-                    this._callHandler.handleRoomState(this, timelineStateEvents, txn, log);
-                }
-            }
-        }
+        });
     }
 
     /** @package */

src/matrix/room/RoomSummary.js

@@ -15,7 +15,7 @@ limitations under the License.
 */
 
 import {MEGOLM_ALGORITHM} from "../e2ee/common.js";
-
+import {iterateResponseStateEvents} from "./common";
 
 function applyTimelineEntries(data, timelineEntries, isInitialSync, canMarkUnread, ownUserId) {
     if (timelineEntries.length) {
@@ -27,25 +27,6 @@ function applyTimelineEntries(data, timelineEntries, isInitialSync, canMarkUnrea
     return data;
 }
 
-export function reduceStateEvents(roomResponse, callback, value) {
-    const stateEvents = roomResponse?.state?.events;
-    // state comes before timeline
-    if (Array.isArray(stateEvents)) {
-        value = stateEvents.reduce(callback, value);
-    }
-    const timelineEvents = roomResponse?.timeline?.events;
-    // and after that state events in the timeline
-    if (Array.isArray(timelineEvents)) {
-        value = timelineEvents.reduce((data, event) => {
-            if (typeof event.state_key === "string") {
-                value = callback(value, event);
-            }
-            return value;
-        }, value);
-    }
-    return value;
-}
-
 function applySyncResponse(data, roomResponse, membership, ownUserId) {
     if (roomResponse.summary) {
         data = updateSummary(data, roomResponse.summary);
@@ -60,7 +41,9 @@ function applySyncResponse(data, roomResponse, membership, ownUserId) {
     // process state events in state and in timeline.
     // non-state events are handled by applyTimelineEntries
     // so decryption is handled properly
-    data = reduceStateEvents(roomResponse, (data, event) => processStateEvent(data, event, ownUserId), data);
+    iterateResponseStateEvents(roomResponse, event => {
+        data = processStateEvent(data, event, ownUserId);
+    });
     const unreadNotifications = roomResponse.unread_notifications;
     if (unreadNotifications) {
         data = processNotificationCounts(data, unreadNotifications);

src/matrix/room/common.ts

@@ -14,6 +14,12 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+import type {Room} from "./Room";
+import type {StateEvent, TimelineEvent} from "../storage/types";
+import type {Transaction} from "../storage/idb/Transaction";
+import type {ILogItem} from "../../logging/types";
+import type {MemberChange} from "./members/RoomMember";
+
 export function getPrevContentFromStateEvent(event) {
     // where to look for prev_content is a bit of a mess,
     // see https://matrix.to/#/!NasysSDfxKxZBzJJoE:matrix.org/$DvrAbZJiILkOmOIuRsNoHmh2v7UO5CWp_rYhlGk34fQ?via=matrix.org&via=pixie.town&via=amorgan.xyz
@@ -40,3 +46,72 @@ export enum RoomType {
     Private,
     Public
 }
+
+type RoomResponse = {
+    state?: {
+        events?: Array<StateEvent>
+    },
+    timeline?: {
+        events?: Array<StateEvent>
+    }
+}
+
+/** iterates over any state events in a sync room response, in the order that they should be applied (from older to younger events) */
+export function iterateResponseStateEvents(roomResponse: RoomResponse, callback: (StateEvent) => void) {
+    // first iterate over state events, they precede the timeline
+    const stateEvents = roomResponse.state?.events;
+    if (stateEvents) {
+        for (let i = 0; i < stateEvents.length; i++) {
+            callback(stateEvents[i]);
+        }
+    }
+    // now see if there are any state events within the timeline
+    let timelineEvents = roomResponse.timeline?.events;
+    if (timelineEvents) {
+        for (let i = 0; i < timelineEvents.length; i++) {
+            const event = timelineEvents[i];
+            if (typeof event.state_key === "string") {
+                callback(event);
+            }
+        }
+    }
+}
+
+export function tests() {
+    return {
+        "test iterateResponseStateEvents with both state and timeline sections": assert => {
+            const roomResponse = {
+                state: {
+                    events: [
+                        {type: "m.room.member", state_key: "1"},
+                        {type: "m.room.member", state_key: "2", content: {a: 1}},
+                    ]
+                },
+                timeline: {
+                    events: [
+                        {type: "m.room.message"},
+                        {type: "m.room.member", state_key: "3"},
+                        {type: "m.room.message"},
+                        {type: "m.room.member", state_key: "2", content: {a: 2}},
+                    ]
+                }
+            } as unknown as RoomResponse;
+            const expectedStateKeys = ["1", "2", "3", "2"];
+            const expectedAForMember2 = [1, 2];
+            iterateResponseStateEvents(roomResponse, event => {
+                assert.strictEqual(event.type, "m.room.member");
+                assert.strictEqual(expectedStateKeys.shift(), event.state_key);
+                if (event.state_key === "2") {
+                    assert.strictEqual(expectedAForMember2.shift(), event.content.a);
+                }
+            });
+            assert.strictEqual(expectedStateKeys.length, 0);
+            assert.strictEqual(expectedAForMember2.length, 0);
+        },
+        "test iterateResponseStateEvents with empty response": assert => {
+            iterateResponseStateEvents({}, () => {
+                assert.fail("no events expected");
+            });
+        }
+    }
+}

src/matrix/room/state/ObservedStateKeyValue.ts

@@ -0,0 +1,104 @@
+/*
+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 type {StateObserver} from "./types";
+import type {StateEvent} from "../../storage/types";
+import type {Transaction} from "../../storage/idb/Transaction";
+import {BaseObservableValue} from "../../../observable/value/BaseObservableValue";
+
+/**
+ * Observable value for a state event with a given type and state key.
+ * Unsubscribes when last subscription is removed */
+export class ObservedStateKeyValue extends BaseObservableValue<StateEvent | undefined> implements StateObserver {
+    private event?: StateEvent;
+    private removeCallback?: () => void;
+
+    constructor(private readonly type: string, private readonly stateKey: string) {
+        super();
+    }
+    /** @internal */
+    async load(roomId: string, txn: Transaction): Promise<void> {
+        this.event = (await txn.roomState.get(roomId, this.type, this.stateKey))?.event;
+    }
+    /** @internal */
+    handleStateEvent(event: StateEvent) {
+        if (event.type === this.type && event.state_key === this.stateKey) {
+            this.event = event;
+            this.emit(this.get());
+        }
+    }
+
+    get(): StateEvent | undefined {
+        return this.event;
+    }
+
+    setRemoveCallback(callback: () => void) {
+        this.removeCallback = callback;
+    }
+
+    onUnsubscribeLast() {
+        this.removeCallback?.();
+    }
+}
+
+import {createMockStorage} from "../../../mocks/Storage";
+
+export async function tests() {
+    return {
+        "test load and update": async assert => {
+            const storage = await createMockStorage();
+            const writeTxn = await storage.readWriteTxn([storage.storeNames.roomState]);
+            writeTxn.roomState.set("!abc", {
+                event_id: "$abc",
+                type: "m.room.member",
+                state_key: "@alice",
+                sender: "@alice",
+                origin_server_ts: 5,
+                content: {}
+            });
+            await writeTxn.complete();
+            const txn = await storage.readTxn([storage.storeNames.roomState]);
+            const value = new ObservedStateKeyValue("m.room.member", "@alice");
+            await value.load("!abc", txn);
+            const updates: Array<StateEvent | undefined> = [];
+            assert.strictEqual(value.get()?.origin_server_ts, 5);
+            const unsubscribe = value.subscribe(value => updates.push(value));
+            value.handleStateEvent({
+                event_id: "$abc",
+                type: "m.room.member",
+                state_key: "@bob",
+                sender: "@alice",
+                origin_server_ts: 10,
+                content: {}
+            });
+            assert.strictEqual(updates.length, 0);
+            value.handleStateEvent({
+                event_id: "$abc",
+                type: "m.room.member",
+                state_key: "@alice",
+                sender: "@alice",
+                origin_server_ts: 10,
+                content: {}
+            });
+            assert.strictEqual(updates.length, 1);
+            assert.strictEqual(updates[0]?.origin_server_ts, 10);
+            let removed = false;
+            value.setRemoveCallback(() => removed = true);
+            unsubscribe();
+            assert(removed);
+        }
+    }
+}

src/matrix/room/state/ObservedStateTypeMap.ts

@@ -0,0 +1,53 @@
+/*
+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 type {StateObserver} from "./types";
+import type {StateEvent} from "../../storage/types";
+import type {Transaction} from "../../storage/idb/Transaction";
+import {ObservableMap} from "../../../observable/map/ObservableMap";
+
+/**
+ * Observable map for a given type with state keys as map keys.
+ * Unsubscribes when last subscription is removed */
+export class ObservedStateTypeMap extends ObservableMap<string, StateEvent> implements StateObserver {
+    private removeCallback?: () => void;
+    
+    constructor(private readonly type: string) {
+        super();
+    }
+    /** @internal */
+    async load(roomId: string, txn: Transaction): Promise<void> {
+        const events = await txn.roomState.getAllForType(roomId, this.type);
+        for (let i = 0; i < events.length; ++i) {
+            const {event} = events[i];
+            this.add(event.state_key, event);
+        }
+    }
+    /** @internal */
+    handleStateEvent(event: StateEvent) {
+        if (event.type === this.type) {
+            this.set(event.state_key, event);
+        }
+    }
+
+    setRemoveCallback(callback: () => void) {
+        this.removeCallback = callback;
+    }
+
+    onUnsubscribeLast() {
+        this.removeCallback?.();
+    }
+}

src/matrix/room/state/RoomStateHandlerSet.ts

@@ -0,0 +1,37 @@
+/*
+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 type {ILogItem} from "../../../logging/types";
+import type {StateEvent} from "../../storage/types";
+import type {Transaction} from "../../storage/idb/Transaction";
+import type {Room} from "../Room";
+import type {MemberChange} from "../members/RoomMember";
+import type {RoomStateHandler} from "./types";
+import {BaseObservable} from "../../../observable/BaseObservable";
+
+/** keeps track of all handlers registered with Session.observeRoomState */
+export class RoomStateHandlerSet extends BaseObservable<RoomStateHandler> implements RoomStateHandler {
+    handleRoomState(room: Room, stateEvent: StateEvent, txn: Transaction, log: ILogItem) {
+        for(let h of this._handlers) {
+            h.handleRoomState(room, stateEvent, txn, log);
+        }
+    }
+    updateRoomMembers(room: Room, memberChanges: Map<string, MemberChange>) {
+        for(let h of this._handlers) {
+            h.updateRoomMembers(room, memberChanges);
+        }
+    }
+}

src/matrix/room/state/types.ts

@@ -0,0 +1,38 @@
+/*
+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 type {Room} from "../Room";
+import type {StateEvent} from "../../storage/types";
+import type {Transaction} from "../../storage/idb/Transaction";
+import type {ILogItem} from "../../../logging/types";
+import type {MemberChange} from "../members/RoomMember";
+
+/** used for Session.observeRoomState, which observes in all room, but without loading from storage
+ * It receives the sync write transaction, so other stores can be updated as part of the same transaction. */
+export interface RoomStateHandler {
+    handleRoomState(room: Room, stateEvent: StateEvent, syncWriteTxn: Transaction, log: ILogItem);
+    updateRoomMembers(room: Room, memberChanges: Map<string, MemberChange>);
+}
+
+/**
+ * used for Room.observeStateType and Room.observeStateTypeAndKey
+ * @internal
+ * */
+export interface StateObserver {
+    handleStateEvent(event: StateEvent);
+    load(roomId: string, txn: Transaction): Promise<void>;
+    setRemoveCallback(callback: () => void);
+}