Merge pull request #348 from matrix-org/rav/device_list_stream

Use the device change notifications interface

src/base-apis.js

@@ -1018,20 +1018,35 @@ MatrixBaseApis.prototype.uploadKeysRequest = function(content, opts, callback) {
  *
  * @param {string[]} userIds  list of users to get keys for
  *
- * @param {module:client.callback=} callback
+ * @param {Object=} opts
+ *
+ * @param {string=} opts.token   sync token to pass in the query request, to help
+ *   the HS give the most recent results
  *
  * @return {module:client.Promise} Resolves: result object. Rejects: with
  *     an error response ({@link module:http-api.MatrixError}).
  */
-MatrixBaseApis.prototype.downloadKeysForUsers = function(userIds, callback) {
-    const downloadQuery = {};
-
-    for (let i = 0; i < userIds.length; ++i) {
-        downloadQuery[userIds[i]] = {};
+MatrixBaseApis.prototype.downloadKeysForUsers = function(userIds, opts) {
+    if (utils.isFunction(opts)) {
+        // opts used to be 'callback'.
+        throw new Error(
+            'downloadKeysForUsers no longer accepts a callback parameter',
+        );
     }
-    const content = {device_keys: downloadQuery};
+    opts = opts || {};
+
+    const content = {
+        device_keys: {},
+    };
+    if ('token' in opts) {
+        content.token = opts.token;
+    }
+    userIds.forEach((u) => {
+        content.device_keys[u] = {};
+    });
+
     return this._http.authedRequestWithPrefix(
-        callback, "POST", "/keys/query", undefined, content,
+        undefined, "POST", "/keys/query", undefined, content,
         httpApi.PREFIX_UNSTABLE,
     );
 };
@@ -1067,6 +1082,28 @@ MatrixBaseApis.prototype.claimOneTimeKeys = function(devices, key_algorithm) {
     );
 };
 
+/**
+ * Ask the server for a list of users who have changed their device lists
+ * between a pair of sync tokens
+ *
+ * @param {string} oldToken
+ * @param {string} newToken
+ *
+ * @return {module:client.Promise} Resolves: result object. Rejects: with
+ *     an error response ({@link module:http-api.MatrixError}).
+ */
+MatrixBaseApis.prototype.getKeyChanges = function(oldToken, newToken) {
+    const qps = {
+        from: oldToken,
+        to: newToken,
+    };
+
+    return this._http.authedRequestWithPrefix(
+        undefined, "GET", "/keys/changes", qps, undefined,
+        httpApi.PREFIX_UNSTABLE,
+    );
+};
+
 
 // Identity Server Operations
 // ==========================

src/client.js

@@ -158,6 +158,7 @@ function MatrixClient(opts) {
             this, this,
             opts.sessionStore,
             userId, this.deviceId,
+            this.store,
         );
 
         this.olmVersion = Crypto.getOlmVersion();
@@ -2665,8 +2666,6 @@ MatrixClient.prototype.startClient = function(opts) {
         };
     }
 
-    this._clientOpts = opts;
-
     if (this._crypto) {
         this._crypto.uploadKeys(5).done();
         const tenMinutes = 1000 * 60 * 10;
@@ -2684,6 +2683,13 @@ MatrixClient.prototype.startClient = function(opts) {
         console.error("Still have sync object whilst not running: stopping old one");
         this._syncApi.stop();
     }
+
+    // shallow-copy the opts dict before modifying and storing it
+    opts = Object.assign({}, opts);
+
+    opts.crypto = this._crypto;
+    this._clientOpts = opts;
+
     this._syncApi = new SyncApi(this, opts);
     this._syncApi.sync();
 };
@@ -3067,12 +3073,26 @@ module.exports.CRYPTO_ENABLED = CRYPTO_ENABLED;
  * </ul>
  *
  * @event module:client~MatrixClient#"sync"
+ *
  * @param {string} state An enum representing the syncing state. One of "PREPARED",
  * "SYNCING", "ERROR", "STOPPED".
+ *
  * @param {?string} prevState An enum representing the previous syncing state.
  * One of "PREPARED", "SYNCING", "ERROR", "STOPPED" <b>or null</b>.
+ *
  * @param {?Object} data Data about this transition.
+ *
  * @param {MatrixError} data.err The matrix error if <code>state=ERROR</code>.
+ *
+ * @param {String} data.oldSyncToken The 'since' token passed to /sync.
+ *    <code>null</code> for the first successful sync since this client was
+ *    started. Only present if <code>state=PREPARED</code> or
+ *    <code>state=SYNCING</code>.
+ *
+ * @param {String} data.nextSyncToken The 'next_batch' result from /sync, which
+ *    will become the 'since' token for the next call to /sync. Only present if
+ *    <code>state=PREPARED</code> or <code>state=SYNCING</code>.
+ *
  * @example
  * matrixClient.on("sync", function(state, prevState, data) {
  *   switch (state) {

src/crypto/DeviceList.js

@@ -41,6 +41,8 @@ export default class DeviceList {
 
         // userId -> promise
         this._keyDownloadsInProgressByUser = {};
+
+        this.lastKnownSyncToken = null;
     }
 
     /**
@@ -288,8 +290,13 @@ export default class DeviceList {
     _doKeyDownloadForUsers(downloadUsers) {
         console.log('Starting key download for ' + downloadUsers);
 
+        const token = this.lastKnownSyncToken;
+        const opts = {};
+        if (token) {
+            opts.token = token;
+        }
         return this._baseApis.downloadKeysForUsers(
-            downloadUsers,
+            downloadUsers, opts,
         ).then((res) => {
             const dk = res.device_keys || {};
 
@@ -319,6 +326,10 @@ export default class DeviceList {
                 this._sessionStore.storeEndToEndDevicesForUser(
                     userId, storage,
                 );
+
+                if (token) {
+                    this._sessionStore.storeEndToEndDeviceSyncToken(token);
+                }
             }
         });
     }

src/crypto/index.js

@@ -48,14 +48,16 @@ const DeviceList = require('./DeviceList').default;
  * @param {string} userId The user ID for the local user
  *
  * @param {string} deviceId The identifier for this device.
+ *
+ * @param {Object} clientStore the MatrixClient data store.
  */
-function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId) {
+function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId,
+                clientStore) {
     this._baseApis = baseApis;
     this._sessionStore = sessionStore;
     this._userId = userId;
     this._deviceId = deviceId;
-
-    this._initialSyncCompleted = false;
+    this._clientStore = clientStore;
 
     this._olmDevice = new OlmDevice(sessionStore);
     this._deviceList = new DeviceList(baseApis, sessionStore, this._olmDevice);
@@ -111,11 +113,8 @@ function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId) {
 function _registerEventHandlers(crypto, eventEmitter) {
     eventEmitter.on("sync", function(syncState, oldState, data) {
         try {
-            if (syncState == "PREPARED") {
-                // XXX ugh. we're assuming the eventEmitter is a MatrixClient.
-                // how can we avoid doing so?
-                const rooms = eventEmitter.getRooms();
-                crypto._onInitialSyncCompleted(rooms);
+            if (syncState === "SYNCING") {
+                crypto._onSyncCompleted(data);
             }
         } catch (e) {
             console.error("Error handling sync", e);
@@ -710,6 +709,18 @@ Crypto.prototype.decryptEvent = function(event) {
     alg.decryptEvent(event);
 };
 
+/**
+ * Handle the notification from /sync that a user has updated their device list.
+ *
+ * @param {String} userId
+ */
+Crypto.prototype.userDeviceListChanged = function(userId) {
+    this._deviceList.invalidateUserDeviceList(userId);
+
+    // don't flush the outdated device list yet - we do it once we finish
+    // processing the sync.
+};
+
 /**
  * handle an m.room.encryption event
  *
@@ -729,19 +740,50 @@ Crypto.prototype._onCryptoEvent = function(event) {
 };
 
 /**
- * handle the completion of the initial sync.
+ * handle the completion of a /sync
  *
- * Announces the new device.
+ * This is called after the processing of each successful /sync response.
+ * It is an opportunity to do a batch process on the information received.
+ *
+ * @param {Object} syncData  the data from the 'MatrixClient.sync' event
+ */
+Crypto.prototype._onSyncCompleted = function(syncData) {
+    this._deviceList.lastKnownSyncToken = syncData.nextSyncToken;
+
+    if (!syncData.oldSyncToken) {
+        // an initialsync.
+        this._sendNewDeviceEvents();
+
+        // if we have a deviceSyncToken, we can tell the deviceList to
+        // invalidate devices which have changed since then.
+        const oldSyncToken = this._sessionStore.getEndToEndDeviceSyncToken();
+        if (oldSyncToken) {
+            this._invalidateDeviceListsSince(oldSyncToken).catch((e) => {
+                // if that failed, we fall back to invalidating everyone.
+                console.warn("Error fetching changed device list", e);
+                this._invalidateDeviceListForAllActiveUsers();
+                return this._deviceList.flushNewDeviceRequests();
+            }).done();
+        } else {
+            // otherwise, we have to invalidate all devices for all users we
+            // share a room with.
+            this._invalidateDeviceListForAllActiveUsers();
+        }
+    }
+
+    // catch up on any new devices we got told about during the sync.
+    this._deviceList.refreshOutdatedDeviceLists().done();
+};
+
+/**
+ * Send m.new_device messages to any devices we share a room with.
+ *
+ * (TODO: we can get rid of this once a suitable number of homeservers and
+ * clients support the more reliable device list update stream mechanism)
  *
  * @private
- * @param {module:models/room[]} rooms list of rooms the client knows about
  */
-Crypto.prototype._onInitialSyncCompleted = function(rooms) {
-    this._initialSyncCompleted = true;
-
-    // catch up on any m.new_device events which arrived during the initial sync.
-    this._deviceList.refreshOutdatedDeviceLists().done();
-
+Crypto.prototype._sendNewDeviceEvents = function() {
     if (this._sessionStore.getDeviceAnnounced()) {
         return;
     }
@@ -750,23 +792,7 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) {
     // we have arrived.
     // build a list of rooms for each user.
     const roomsByUser = {};
-    for (let i = 0; i < rooms.length; i++) {
-        const room = rooms[i];
-
-        // check for rooms with encryption enabled
-        const alg = this._roomEncryptors[room.roomId];
-        if (!alg) {
-            continue;
-        }
-
-        // ignore any rooms which we have left
-        const me = room.getMember(this._userId);
-        if (!me || (
-            me.membership !== "join" && me.membership !== "invite"
-        )) {
-            continue;
-        }
-
+    for (const room of this._getE2eRooms()) {
         const members = room.getJoinedMembers();
         for (let j = 0; j < members.length; j++) {
             const m = members[j];
@@ -800,6 +826,88 @@ Crypto.prototype._onInitialSyncCompleted = function(rooms) {
     });
 };
 
+/**
+ * Ask the server which users have new devices since a given token,
+ * invalidate them, and start an update query.
+ *
+ * @param {String} oldSyncToken
+ *
+ * @returns {Promise} resolves once the query is complete. Rejects if the
+ *   keyChange query fails.
+ */
+Crypto.prototype._invalidateDeviceListsSince = function(oldSyncToken) {
+    return this._baseApis.getKeyChanges(
+        oldSyncToken, this.lastKnownSyncToken,
+    ).then((r) => {
+        if (!r.changed || !Array.isArray(r.changed)) {
+            return;
+        }
+
+        // only invalidate users we share an e2e room with - we don't
+        // care about users in non-e2e rooms.
+        const filteredUserIds = this._getE2eRoomMembers();
+        r.changed.forEach((u) => {
+            if (u in filteredUserIds) {
+                this._deviceList.invalidateUserDeviceList(u);
+            }
+        });
+        return this._deviceList.flushNewDeviceRequests();
+    });
+};
+
+/**
+ * Invalidate any stored device list for any users we share an e2e room with
+ *
+ * @private
+ */
+Crypto.prototype._invalidateDeviceListForAllActiveUsers = function() {
+    Object.keys(this._getE2eRoomMembers()).forEach((m) => {
+        this._deviceList.invalidateUserDeviceList(m);
+    });
+};
+
+/**
+ * get the users we share an e2e-enabled room with
+ *
+ * @returns {Object<string>} userid->userid map (should be a Set but argh ES6)
+ */
+Crypto.prototype._getE2eRoomMembers = function() {
+    const userIds = Object.create(null);
+
+    const rooms = this._getE2eRooms();
+    for (const r of rooms) {
+        const members = r.getJoinedMembers();
+        members.forEach((m) => { userIds[m.userId] = m.userId; });
+    }
+
+    return userIds;
+};
+
+/**
+ * Get a list of the e2e-enabled rooms we are members of
+ *
+ * @returns {module:models.Room[]}
+ */
+Crypto.prototype._getE2eRooms = function() {
+    return this._clientStore.getRooms().filter((room) => {
+        // check for rooms with encryption enabled
+        const alg = this._roomEncryptors[room.roomId];
+        if (!alg) {
+            return false;
+        }
+
+        // ignore any rooms which we have left
+        const me = room.getMember(this._userId);
+        if (!me || (
+            me.membership !== "join" && me.membership !== "invite"
+        )) {
+            return false;
+        }
+
+        return true;
+    });
+};
+
 /**
  * Handle a key event
  *
@@ -873,12 +981,6 @@ Crypto.prototype._onNewDeviceEvent = function(event) {
     }
 
     this._deviceList.invalidateUserDeviceList(userId);
-
-    // we delay handling these until the intialsync has completed, so that we
-    // can do all of them together.
-    if (this._initialSyncCompleted) {
-        this._deviceList.refreshOutdatedDeviceLists().done();
-    }
 };
 
 

src/store/session/webstorage.js

@@ -99,6 +99,27 @@ WebStorageSessionStore.prototype = {
         return getJsonItem(this.store, keyEndToEndDevicesForUser(userId));
     },
 
+    /**
+     * Store the sync token corresponding to the device list.
+     *
+     * This is used when starting the client, to get a list of the users who
+     * have changed their device list since the list time we were running.
+     *
+     * @param {String?} token
+     */
+    storeEndToEndDeviceSyncToken: function(token) {
+        setJsonItem(this.store, KEY_END_TO_END_DEVICE_SYNC_TOKEN, token);
+    },
+
+    /**
+     * Get the sync token corresponding to the device list.
+     *
+     * @return {String?} token
+     */
+    getEndToEndDeviceSyncToken: function() {
+        return getJsonItem(this.store, KEY_END_TO_END_DEVICE_SYNC_TOKEN);
+    },
+
     /**
      * Store a session between the logged-in user and another device
      * @param {string} deviceKey The public key of the other device.
@@ -180,6 +201,7 @@ WebStorageSessionStore.prototype = {
 
 const KEY_END_TO_END_ACCOUNT = E2E_PREFIX + "account";
 const KEY_END_TO_END_ANNOUNCED = E2E_PREFIX + "announced";
+const KEY_END_TO_END_DEVICE_SYNC_TOKEN = E2E_PREFIX + "device_sync_token";
 
 function keyEndToEndDevicesForUser(userId) {
     return E2E_PREFIX + "devices/" + userId;
@@ -199,6 +221,8 @@ function keyEndToEndRoom(roomId) {
 
 function getJsonItem(store, key) {
     try {
+        // if the key is absent, store.getItem() returns null, and
+        // JSON.parse(null) === null, so this returns null.
         return JSON.parse(store.getItem(key));
     } catch (e) {
         debuglog("Failed to get key %s: %s", key, e);

src/sync.js

@@ -58,6 +58,7 @@ function debuglog() {
  * @constructor
  * @param {MatrixClient} client The matrix client instance to use.
  * @param {Object} opts Config options
+ * @param {module:crypto=} opts.crypto Crypto manager
  */
 function SyncApi(client, opts) {
     this.client = client;
@@ -529,13 +530,18 @@ SyncApi.prototype._sync = function(syncOptions) {
         }
 
         // emit synced events
+        const syncEventData = {
+            oldSyncToken: syncToken,
+            nextSyncToken: data.next_batch,
+        };
+
         if (!syncOptions.hasSyncedBefore) {
-            self._updateSyncState("PREPARED");
+            self._updateSyncState("PREPARED", syncEventData);
             syncOptions.hasSyncedBefore = true;
         }
 
         // keep emitting SYNCING -> SYNCING for clients who want to do bulk updates
-        self._updateSyncState("SYNCING");
+        self._updateSyncState("SYNCING", syncEventData);
 
         self._sync(syncOptions);
     }, function(err) {
@@ -584,6 +590,7 @@ SyncApi.prototype._processSyncResponse = function(syncToken, data) {
     //    next_batch: $token,
     //    presence: { events: [] },
     //    account_data: { events: [] },
+    //    device_lists: { changed: ["@user:server", ... ]},
     //    to_device: { events: [] },
     //    rooms: {
     //      invite: {
@@ -859,6 +866,13 @@ SyncApi.prototype._processSyncResponse = function(syncToken, data) {
             client.getNotifTimelineSet().addLiveEvent(event);
         });
     }
+
+    // Handle device list updates
+    if (this.opts.crypto && data.device_lists && data.device_lists.changed) {
+        data.device_lists.changed.forEach((u) => {
+            this.opts.crypto.userDeviceListChanged(u);
+        });
+    }
 };
 
 /**