Add local echo capabilities for rooms

The structure here might need some documentation and work, but overall the idea is that all calls pass through a CachedEcho instance, which are self-updating.

src/stores/local-echo/CachedEcho.ts

@@ -0,0 +1,72 @@
+/*
+Copyright 2020 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 { EchoContext } from "./EchoContext";
+import { RunFn, TransactionStatus } from "./EchoTransaction";
+import { MatrixClient } from "matrix-js-sdk/src/client";
+import { EventEmitter } from "events";
+
+export async function implicitlyReverted() {
+    // do nothing :D
+}
+
+export const PROPERTY_UPDATED = "property_updated";
+
+export abstract class CachedEcho<C extends EchoContext, K, V> extends EventEmitter {
+    private cache = new Map<K, V>();
+    protected matrixClient: MatrixClient;
+
+    protected constructor(protected context: C, private lookupFn: (key: K) => V) {
+        super();
+    }
+
+    public setClient(client: MatrixClient) {
+        const oldClient = this.matrixClient;
+        this.matrixClient = client;
+        this.onClientChanged(oldClient, client);
+    }
+
+    protected abstract onClientChanged(oldClient: MatrixClient, newClient: MatrixClient);
+
+    /**
+     * Gets a value. If the key is in flight, the cached value will be returned. If
+     * the key is not in flight then the lookupFn provided to this class will be
+     * called instead.
+     * @param key The key to look up.
+     * @returns The value for the key.
+     */
+    public getValue(key: K): V {
+        return this.cache.has(key) ? this.cache.get(key) : this.lookupFn(key);
+    }
+
+    private cacheVal(key: K, val: V) {
+        this.cache.set(key, val);
+        this.emit(PROPERTY_UPDATED, key);
+    }
+
+    private decacheKey(key: K) {
+        this.cache.delete(key);
+        this.emit(PROPERTY_UPDATED, key);
+    }
+
+    public setValue(auditName: string, key: K, targetVal: V, runFn: RunFn, revertFn: RunFn) {
+        this.cacheVal(key, targetVal); // set the cache now as it won't be updated by the .when() ladder below.
+        this.context.beginTransaction(auditName, runFn)
+            .when(TransactionStatus.Pending, () => this.cacheVal(key, targetVal))
+            .whenAnyOf([TransactionStatus.DoneError, TransactionStatus.DoneSuccess], () => this.decacheKey(key))
+            .when(TransactionStatus.DoneError, () => revertFn());
+    }
+}

src/stores/local-echo/EchoChamber.ts

@@ -0,0 +1,31 @@
+/*
+Copyright 2020 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 { RoomCachedEcho } from "./RoomCachedEcho";
+import { Room } from "matrix-js-sdk/src/models/room";
+import { EchoStore } from "./EchoStore";
+
+/**
+ * Semantic access to local echo
+ */
+export class EchoChamber {
+    private constructor() {
+    }
+
+    public static forRoom(room: Room): RoomCachedEcho {
+        return EchoStore.instance.getOrCreateEchoForRoom(room);
+    }
+}

src/stores/local-echo/EchoContext.ts

@@ -0,0 +1,68 @@
+/*
+Copyright 2020 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 { EchoTransaction, RunFn, TransactionStatus } from "./EchoTransaction";
+import { arrayFastClone } from "../../utils/arrays";
+import { IDestroyable } from "../../utils/IDestroyable";
+import { Whenable } from "../../utils/Whenable";
+
+export enum ContextTransactionState {
+    NotStarted,
+    PendingErrors,
+    AllSuccessful
+}
+
+export abstract class EchoContext extends Whenable<ContextTransactionState> implements IDestroyable {
+    private _transactions: EchoTransaction[] = [];
+    public readonly startTime: Date = new Date();
+
+    public get transactions(): EchoTransaction[] {
+        return arrayFastClone(this._transactions);
+    }
+
+    public beginTransaction(auditName: string, runFn: RunFn): EchoTransaction {
+        const txn = new EchoTransaction(auditName, runFn);
+        this._transactions.push(txn);
+        txn.whenAnything(this.checkTransactions);
+
+        // We have no intent to call the transaction again if it succeeds (in fact, it'll
+        // be really angry at us if we do), so call that the end of the road for the events.
+        txn.when(TransactionStatus.DoneSuccess, () => txn.destroy());
+
+        return txn;
+    }
+
+    private checkTransactions = () => {
+        let status = ContextTransactionState.AllSuccessful;
+        for (const txn of this.transactions) {
+            if (txn.status === TransactionStatus.DoneError) {
+                status = ContextTransactionState.PendingErrors;
+                break;
+            } else if (txn.status === TransactionStatus.Pending) {
+                status = ContextTransactionState.NotStarted;
+                // no break as we might hit something which broke
+            }
+        }
+        this.notifyCondition(status);
+    };
+
+    public destroy() {
+        for (const txn of this.transactions) {
+            txn.destroy();
+        }
+        super.destroy();
+    }
+}

src/stores/local-echo/EchoStore.ts

@@ -0,0 +1,72 @@
+/*
+Copyright 2020 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 { EventEmitter } from "events";
+import { CachedEcho } from "./CachedEcho";
+import { Room } from "matrix-js-sdk/src/models/room";
+import { RoomCachedEcho } from "./RoomCachedEcho";
+import { RoomEchoContext } from "./RoomEchoContext";
+import { AsyncStoreWithClient } from "../AsyncStoreWithClient";
+import defaultDispatcher from "../../dispatcher/dispatcher";
+import { ActionPayload } from "../../dispatcher/payloads";
+
+type ContextKey = string;
+
+const roomContextKey = (room: Room): ContextKey => `room-${room.roomId}`;
+
+export class EchoStore extends AsyncStoreWithClient<any> {
+    private static _instance: EchoStore;
+
+    private caches = new Map<ContextKey, CachedEcho<any, any, any>>();
+
+    constructor() {
+        super(defaultDispatcher);
+    }
+
+    public static get instance(): EchoStore {
+        if (!EchoStore._instance) {
+            EchoStore._instance = new EchoStore();
+        }
+        return EchoStore._instance;
+    }
+
+    public getOrCreateEchoForRoom(room: Room): RoomCachedEcho {
+        if (this.caches.has(roomContextKey(room))) {
+            return this.caches.get(roomContextKey(room)) as RoomCachedEcho;
+        }
+        const echo = new RoomCachedEcho(new RoomEchoContext(room));
+        echo.setClient(this.matrixClient);
+        this.caches.set(roomContextKey(room), echo);
+        return echo;
+    }
+
+    protected async onReady(): Promise<any> {
+        for (const echo of this.caches.values()) {
+            echo.setClient(this.matrixClient);
+        }
+    }
+
+    protected async onNotReady(): Promise<any> {
+        for (const echo of this.caches.values()) {
+            echo.setClient(null);
+        }
+    }
+
+    protected async onAction(payload: ActionPayload): Promise<any> {
+        // We have nothing to actually listen for
+        return Promise.resolve();
+    }
+}

src/stores/local-echo/EchoTransaction.ts

@@ -0,0 +1,65 @@
+/*
+Copyright 2020 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 { Whenable } from "../../utils/Whenable";
+
+export type RunFn = () => Promise<void>;
+
+export enum TransactionStatus {
+    Pending,
+    DoneSuccess,
+    DoneError,
+}
+
+export class EchoTransaction extends Whenable<TransactionStatus> {
+    private _status = TransactionStatus.Pending;
+    private didFail = false;
+
+    public constructor(
+        public readonly auditName,
+        public runFn: RunFn,
+    ) {
+        super();
+    }
+
+    public get didPreviouslyFail(): boolean {
+        return this.didFail;
+    }
+
+    public get status(): TransactionStatus {
+        return this._status;
+    }
+
+    public run() {
+        if (this.status === TransactionStatus.DoneSuccess) {
+            throw new Error("Cannot re-run a successful echo transaction");
+        }
+        this.setStatus(TransactionStatus.Pending);
+        this.runFn()
+            .then(() => this.setStatus(TransactionStatus.DoneSuccess))
+            .catch(() => this.setStatus(TransactionStatus.DoneError));
+    }
+
+    private setStatus(status: TransactionStatus) {
+        this._status = status;
+        if (status === TransactionStatus.DoneError) {
+            this.didFail = true;
+        } else if (status === TransactionStatus.DoneSuccess) {
+            this.didFail = false;
+        }
+        this.notifyCondition(status);
+    }
+}

src/stores/local-echo/RoomCachedEcho.ts

@@ -0,0 +1,77 @@
+/*
+Copyright 2020 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 { CachedEcho, implicitlyReverted, PROPERTY_UPDATED } from "./CachedEcho";
+import { getRoomNotifsState, setRoomNotifsState } from "../../RoomNotifs";
+import { RoomEchoContext } from "./RoomEchoContext";
+import { _t } from "../../languageHandler";
+import { Volume } from "../../RoomNotifsTypes";
+import { MatrixEvent } from "matrix-js-sdk/src/models/event";
+
+export type CachedRoomValues = Volume;
+
+export enum CachedRoomKey {
+    NotificationVolume,
+}
+
+export class RoomCachedEcho extends CachedEcho<RoomEchoContext, CachedRoomKey, CachedRoomValues> {
+    private properties = new Map<CachedRoomKey, CachedRoomValues>();
+
+    public constructor(context: RoomEchoContext) {
+        super(context, (k) => this.properties.get(k));
+    }
+
+    protected onClientChanged(oldClient, newClient) {
+        this.properties.clear();
+        if (oldClient) {
+            oldClient.removeListener("accountData", this.onAccountData);
+        }
+        if (newClient) {
+            // Register the listeners first
+            newClient.on("accountData", this.onAccountData);
+
+            // Then populate the properties map
+            this.updateNotificationVolume();
+        }
+    }
+
+    private onAccountData = (event: MatrixEvent) => {
+        if (event.getType() === "m.push_rules") {
+            const currentVolume = this.properties.get(CachedRoomKey.NotificationVolume) as Volume;
+            const newVolume = getRoomNotifsState(this.context.room.roomId) as Volume;
+            if (currentVolume !== newVolume) {
+                this.updateNotificationVolume();
+            }
+        }
+    };
+
+    private updateNotificationVolume() {
+        this.properties.set(CachedRoomKey.NotificationVolume, getRoomNotifsState(this.context.room.roomId));
+        this.emit(PROPERTY_UPDATED, CachedRoomKey.NotificationVolume);
+    }
+
+    // ---- helpers below here ----
+
+    public get notificationVolume(): Volume {
+        return this.getValue(CachedRoomKey.NotificationVolume);
+    }
+
+    public set notificationVolume(v: Volume) {
+        this.setValue(_t("Change notification settings"), CachedRoomKey.NotificationVolume, v, async () => {
+            setRoomNotifsState(this.context.room.roomId, v);
+        }, implicitlyReverted);
+    }
+}

src/stores/local-echo/RoomEchoContext.ts

@@ -0,0 +1,24 @@
+/*
+Copyright 2020 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 { EchoContext } from "./EchoContext";
+import { Room } from "matrix-js-sdk/src/models/room";
+
+export class RoomEchoContext extends EchoContext {
+    constructor(public readonly room: Room) {
+        super();
+    }
+}

src/utils/Whenable.ts

@@ -0,0 +1,86 @@
+/*
+Copyright 2020 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 { IDestroyable } from "./IDestroyable";
+import { arrayFastClone } from "./arrays";
+
+export type WhenFn<T> = (w: Whenable<T>) => void;
+
+/**
+ * Whenables are a cheap way to have Observable patterns mixed with typical
+ * usage of Promises, without having to tear down listeners or calls. Whenables
+ * are intended to be used when a condition will be met multiple times and
+ * the consumer needs to know *when* that happens.
+ */
+export abstract class Whenable<T> implements IDestroyable {
+    private listeners: {condition: T | null, fn: WhenFn<T>}[] = [];
+
+    /**
+     * Sets up a call to `fn` *when* the `condition` is met.
+     * @param condition The condition to match.
+     * @param fn The function to call.
+     * @returns This.
+     */
+    public when(condition: T, fn: WhenFn<T>): Whenable<T> {
+        this.listeners.push({condition, fn});
+        return this;
+    }
+
+    /**
+     * Sets up a fall to `fn` *when* any of the `conditions` are met.
+     * @param conditions The conditions to match.
+     * @param fn The function to call.
+     * @returns This.
+     */
+    public whenAnyOf(conditions: T[], fn: WhenFn<T>): Whenable<T> {
+        for (const condition of conditions) {
+            this.when(condition, fn);
+        }
+        return this;
+    }
+
+    /**
+     * Sets up a call to `fn` *when* any condition is met.
+     * @param fn The function to call.
+     * @returns This.
+     */
+    public whenAnything(fn: WhenFn<T>): Whenable<T> {
+        this.listeners.push({condition: null, fn});
+        return this;
+    }
+
+    /**
+     * Notifies all the whenables of a given condition.
+     * @param condition The new condition that has been met.
+     */
+    protected notifyCondition(condition: T) {
+        const listeners = arrayFastClone(this.listeners); // clone just in case the handler modifies us
+        for (const listener of listeners) {
+            if (listener.condition === null || listener.condition === condition) {
+                try {
+                    listener.fn(this);
+                } catch (e) {
+                    console.error(`Error calling whenable listener for ${condition}:`, e);
+                }
+            }
+        }
+    }
+
+    public destroy() {
+        this.listeners = [];
+    }
+}