Merge branch 'poljar/seshat-pr' into develop

src/BasePlatform.js

@@ -19,6 +19,7 @@ limitations under the License.
 */
 
 import dis from './dispatcher';
+import BaseEventIndexManager from './indexing/BaseEventIndexManager';
 
 /**
  * Base class for classes that provide platform-specific functionality
@@ -151,4 +152,14 @@ export default class BasePlatform {
     async setMinimizeToTrayEnabled(enabled: boolean): void {
         throw new Error("Unimplemented");
     }
+
+    /**
+     * Get our platform specific EventIndexManager.
+     *
+     * @return {BaseEventIndexManager} The EventIndex manager for our platform,
+     * can be null if the platform doesn't support event indexing.
+     */
+    getEventIndexingManager(): BaseEventIndexManager | null {
+        return null;
+    }
 }

src/Lifecycle.js

@@ -20,6 +20,7 @@ import Promise from 'bluebird';
 import Matrix from 'matrix-js-sdk';
 
 import MatrixClientPeg from './MatrixClientPeg';
+import EventIndexPeg from './indexing/EventIndexPeg';
 import createMatrixClient from './utils/createMatrixClient';
 import Analytics from './Analytics';
 import Notifier from './Notifier';
@@ -589,6 +590,7 @@ async function startMatrixClient(startSyncing=true) {
 
     if (startSyncing) {
         await MatrixClientPeg.start();
+        await EventIndexPeg.init();
     } else {
         console.warn("Caller requested only auxiliary services be started");
         await MatrixClientPeg.assign();
@@ -607,20 +609,20 @@ async function startMatrixClient(startSyncing=true) {
  * Stops a running client and all related services, and clears persistent
  * storage. Used after a session has been logged out.
  */
-export function onLoggedOut() {
+export async function onLoggedOut() {
     _isLoggingOut = false;
     // Ensure that we dispatch a view change **before** stopping the client so
     // so that React components unmount first. This avoids React soft crashes
     // that can occur when components try to use a null client.
     dis.dispatch({action: 'on_logged_out'}, true);
     stopMatrixClient();
-    _clearStorage().done();
+    await _clearStorage();
 }
 
 /**
  * @returns {Promise} promise which resolves once the stores have been cleared
  */
-function _clearStorage() {
+async function _clearStorage() {
     Analytics.logout();
 
     if (window.localStorage) {
@@ -632,7 +634,9 @@ function _clearStorage() {
         // we'll never make any requests, so can pass a bogus HS URL
         baseUrl: "",
     });
-    return cli.clearStores();
+
+    await EventIndexPeg.deleteEventIndex();
+    await cli.clearStores();
 }
 
 /**
@@ -649,6 +653,7 @@ export function stopMatrixClient(unsetClient=true) {
     IntegrationManagers.sharedInstance().stopWatching();
     Mjolnir.sharedInstance().stop();
     if (DMRoomMap.shared()) DMRoomMap.shared().stop();
+    EventIndexPeg.stop();
     const cli = MatrixClientPeg.get();
     if (cli) {
         cli.stopClient();
@@ -656,6 +661,7 @@ export function stopMatrixClient(unsetClient=true) {
 
         if (unsetClient) {
             MatrixClientPeg.unset();
+            EventIndexPeg.unset();
         }
     }
 }

src/Searching.js

@@ -0,0 +1,138 @@
+/*
+Copyright 2019 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 EventIndexPeg from "./indexing/EventIndexPeg";
+import MatrixClientPeg from "./MatrixClientPeg";
+
+function serverSideSearch(term, roomId = undefined) {
+    let filter;
+    if (roomId !== undefined) {
+        // XXX: it's unintuitive that the filter for searching doesn't have
+        // the same shape as the v2 filter API :(
+        filter = {
+            rooms: [roomId],
+        };
+    }
+
+    const searchPromise = MatrixClientPeg.get().searchRoomEvents({
+        filter,
+        term,
+    });
+
+    return searchPromise;
+}
+
+async function combinedSearch(searchTerm) {
+    // Create two promises, one for the local search, one for the
+    // server-side search.
+    const serverSidePromise = serverSideSearch(searchTerm);
+    const localPromise = localSearch(searchTerm);
+
+    // Wait for both promises to resolve.
+    await Promise.all([serverSidePromise, localPromise]);
+
+    // Get both search results.
+    const localResult = await localPromise;
+    const serverSideResult = await serverSidePromise;
+
+    // Combine the search results into one result.
+    const result = {};
+
+    // Our localResult and serverSideResult are both ordered by
+    // recency separately, when we combine them the order might not
+    // be the right one so we need to sort them.
+    const compare = (a, b) => {
+        const aEvent = a.context.getEvent().event;
+        const bEvent = b.context.getEvent().event;
+
+        if (aEvent.origin_server_ts >
+            bEvent.origin_server_ts) return -1;
+        if (aEvent.origin_server_ts <
+            bEvent.origin_server_ts) return 1;
+        return 0;
+    };
+
+    result.count = localResult.count + serverSideResult.count;
+    result.results = localResult.results.concat(
+        serverSideResult.results).sort(compare);
+    result.highlights = localResult.highlights.concat(
+        serverSideResult.highlights);
+
+    return result;
+}
+
+async function localSearch(searchTerm, roomId = undefined) {
+    const searchArgs = {
+        search_term: searchTerm,
+        before_limit: 1,
+        after_limit: 1,
+        order_by_recency: true,
+        room_id: undefined,
+    };
+
+    if (roomId !== undefined) {
+        searchArgs.room_id = roomId;
+    }
+
+    const eventIndex = EventIndexPeg.get();
+
+    const localResult = await eventIndex.search(searchArgs);
+
+    const response = {
+        search_categories: {
+            room_events: localResult,
+        },
+    };
+
+    const emptyResult = {
+        results: [],
+        highlights: [],
+    };
+
+    const result = MatrixClientPeg.get()._processRoomEventsSearch(
+        emptyResult, response);
+
+    return result;
+}
+
+function eventIndexSearch(term, roomId = undefined) {
+    let searchPromise;
+
+    if (roomId !== undefined) {
+        if (MatrixClientPeg.get().isRoomEncrypted(roomId)) {
+            // The search is for a single encrypted room, use our local
+            // search method.
+            searchPromise = localSearch(term, roomId);
+        } else {
+            // The search is for a single non-encrypted room, use the
+            // server-side search.
+            searchPromise = serverSideSearch(term, roomId);
+        }
+    } else {
+        // Search across all rooms, combine a server side search and a
+        // local search.
+        searchPromise = combinedSearch(term);
+    }
+
+    return searchPromise;
+}
+
+export default function eventSearch(term, roomId = undefined) {
+    const eventIndex = EventIndexPeg.get();
+
+    if (eventIndex === null) return serverSideSearch(term, roomId);
+    else return eventIndexSearch(term, roomId);
+}

src/components/structures/RoomView.js

@@ -43,6 +43,7 @@ import Tinter from '../../Tinter';
 import rate_limited_func from '../../ratelimitedfunc';
 import ObjectUtils from '../../ObjectUtils';
 import * as Rooms from '../../Rooms';
+import eventSearch from '../../Searching';
 
 import { KeyCode, isOnlyCtrlOrCmdKeyEvent } from '../../Keyboard';
 
@@ -1129,22 +1130,11 @@ module.exports = createReactClass({
         // todo: should cancel any previous search requests.
         this.searchId = new Date().getTime();
 
-        let filter;
+        let roomId;
-        if (scope === "Room") {
+        if (scope === "Room") roomId = this.state.room.roomId;
-            filter = {
-                // XXX: it's unintuitive that the filter for searching doesn't have the same shape as the v2 filter API :(
-                rooms: [
-                    this.state.room.roomId,
-                ],
-            };
-        }
 
         debuglog("sending search request");
-
+        const searchPromise = eventSearch(term, roomId);
-        const searchPromise = MatrixClientPeg.get().searchRoomEvents({
-            filter: filter,
-            term: term,
-        });
         this._handleSearchResult(searchPromise).done();
     },
 

src/i18n/strings/en_EN.json

@@ -343,6 +343,7 @@
     "Try out new ways to ignore people (experimental)": "Try out new ways to ignore people (experimental)",
     "Send verification requests in direct message, including a new verification UX in the member panel.": "Send verification requests in direct message, including a new verification UX in the member panel.",
     "Enable cross-signing to verify per-user instead of per-device": "Enable cross-signing to verify per-user instead of per-device",
+    "Enable local event indexing and E2EE search (requires restart)": "Enable local event indexing and E2EE search (requires restart)",
     "Use the new, faster, composer for writing messages": "Use the new, faster, composer for writing messages",
     "Enable Emoji suggestions while typing": "Enable Emoji suggestions while typing",
     "Use compact timeline layout": "Use compact timeline layout",

src/indexing/BaseEventIndexManager.js

@@ -0,0 +1,223 @@
+/*
+Copyright 2019 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 interface MatrixEvent {
+    type: string;
+    sender: string;
+    content: {};
+    event_id: string;
+    origin_server_ts: number;
+    unsigned: ?{};
+    room_id: string;
+}
+
+export interface MatrixProfile {
+    avatar_url: string;
+    displayname: string;
+}
+
+export interface CrawlerCheckpoint {
+    roomId: string;
+    token: string;
+    fullCrawl: boolean;
+    direction: string;
+}
+
+export interface ResultContext {
+    events_before: [MatrixEvent];
+    events_after: [MatrixEvent];
+    profile_info: Map<string, MatrixProfile>;
+}
+
+export interface ResultsElement {
+    rank: number;
+    result: MatrixEvent;
+    context: ResultContext;
+}
+
+export interface SearchResult {
+    count: number;
+    results: [ResultsElement];
+    highlights: [string];
+}
+
+export interface SearchArgs {
+    search_term: string;
+    before_limit: number;
+    after_limit: number;
+    order_by_recency: boolean;
+    room_id: ?string;
+}
+
+export interface HistoricEvent {
+    event: MatrixEvent;
+    profile: MatrixProfile;
+}
+
+/**
+ * Base class for classes that provide platform-specific event indexing.
+ *
+ * Instances of this class are provided by the application.
+ */
+export default class BaseEventIndexManager {
+    /**
+     * Does our EventIndexManager support event indexing.
+     *
+     * If an EventIndexManager implementor has runtime dependencies that
+     * optionally enable event indexing they may override this method to perform
+     * the necessary runtime checks here.
+     *
+     * @return {Promise} A promise that will resolve to true if event indexing
+     * is supported, false otherwise.
+     */
+    async supportsEventIndexing(): Promise<boolean> {
+        return true;
+    }
+    /**
+     * Initialize the event index for the given user.
+     *
+     * @return {Promise} A promise that will resolve when the event index is
+     * initialized.
+     */
+    async initEventIndex(): Promise<> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Queue up an event to be added to the index.
+     *
+     * @param {MatrixEvent} ev The event that should be added to the index.
+     * @param {MatrixProfile} profile The profile of the event sender at the
+     * time of the event receival.
+     *
+     * @return {Promise} A promise that will resolve when the was queued up for
+     * addition.
+     */
+    async addEventToIndex(ev: MatrixEvent, profile: MatrixProfile): Promise<> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Check if our event index is empty.
+     */
+    indexIsEmpty(): Promise<boolean> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Commit the previously queued up events to the index.
+     *
+     * @return {Promise} A promise that will resolve once the queued up events
+     * were added to the index.
+     */
+    async commitLiveEvents(): Promise<> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Search the event index using the given term for matching events.
+     *
+     * @param {SearchArgs} searchArgs The search configuration sets what should
+     * be searched for and what should be contained in the search result.
+     *
+     * @return {Promise<[SearchResult]>} A promise that will resolve to an array
+     * of search results once the search is done.
+     */
+    async searchEventIndex(searchArgs: SearchArgs): Promise<SearchResult> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Add events from the room history to the event index.
+     *
+     * This is used to add a batch of events to the index.
+     *
+     * @param {[HistoricEvent]} events The list of events and profiles that
+     * should be added to the event index.
+     * @param {[CrawlerCheckpoint]} checkpoint A new crawler checkpoint that
+     * should be stored in the index which should be used to continue crawling
+     * the room.
+     * @param {[CrawlerCheckpoint]} oldCheckpoint The checkpoint that was used
+     * to fetch the current batch of events. This checkpoint will be removed
+     * from the index.
+     *
+     * @return {Promise} A promise that will resolve to true if all the events
+     * were already added to the index, false otherwise.
+     */
+    async addHistoricEvents(
+        events: [HistoricEvent],
+        checkpoint: CrawlerCheckpoint | null,
+        oldCheckpoint: CrawlerCheckpoint | null,
+    ): Promise<bool> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Add a new crawler checkpoint to the index.
+     *
+     * @param {CrawlerCheckpoint} checkpoint The checkpoint that should be added
+     * to the index.
+     *
+     * @return {Promise} A promise that will resolve once the checkpoint has
+     * been stored.
+     */
+    async addCrawlerCheckpoint(checkpoint: CrawlerCheckpoint): Promise<> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Add a new crawler checkpoint to the index.
+     *
+     * @param {CrawlerCheckpoint} checkpoint The checkpoint that should be
+     * removed from the index.
+     *
+     * @return {Promise} A promise that will resolve once the checkpoint has
+     * been removed.
+     */
+    async removeCrawlerCheckpoint(checkpoint: CrawlerCheckpoint): Promise<> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Load the stored checkpoints from the index.
+     *
+     * @return {Promise<[CrawlerCheckpoint]>} A promise that will resolve to an
+     * array of crawler checkpoints once they have been loaded from the index.
+     */
+    async loadCheckpoints(): Promise<[CrawlerCheckpoint]> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * close our event index.
+     *
+     * @return {Promise} A promise that will resolve once the event index has
+     * been closed.
+     */
+    async closeEventIndex(): Promise<> {
+        throw new Error("Unimplemented");
+    }
+
+    /**
+     * Delete our current event index.
+     *
+     * @return {Promise} A promise that will resolve once the event index has
+     * been deleted.
+     */
+    async deleteEventIndex(): Promise<> {
+        throw new Error("Unimplemented");
+    }
+}

src/indexing/EventIndex.js

@@ -0,0 +1,412 @@
+/*
+Copyright 2019 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 PlatformPeg from "../PlatformPeg";
+import MatrixClientPeg from "../MatrixClientPeg";
+
+/*
+ * Event indexing class that wraps the platform specific event indexing.
+ */
+export default class EventIndex {
+    constructor() {
+        this.crawlerCheckpoints = [];
+        // The time that the crawler will wait between /rooms/{room_id}/messages
+        // requests
+        this._crawlerTimeout = 3000;
+        // The maximum number of events our crawler should fetch in a single
+        // crawl.
+        this._eventsPerCrawl = 100;
+        this._crawler = null;
+        this.liveEventsForIndex = new Set();
+    }
+
+    async init() {
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+        await indexManager.initEventIndex();
+
+        this.registerListeners();
+    }
+
+    registerListeners() {
+        const client = MatrixClientPeg.get();
+
+        client.on('sync', this.onSync);
+        client.on('Room.timeline', this.onRoomTimeline);
+        client.on('Event.decrypted', this.onEventDecrypted);
+        client.on('Room.timelineReset', this.onTimelineReset);
+    }
+
+    removeListeners() {
+        const client = MatrixClientPeg.get();
+        if (client === null) return;
+
+        client.removeListener('sync', this.onSync);
+        client.removeListener('Room.timeline', this.onRoomTimeline);
+        client.removeListener('Event.decrypted', this.onEventDecrypted);
+        client.removeListener('Room.timelineReset', this.onTimelineReset);
+    }
+
+    onSync = async (state, prevState, data) => {
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+
+        if (prevState === null && state === "PREPARED") {
+            // Load our stored checkpoints, if any.
+            this.crawlerCheckpoints = await indexManager.loadCheckpoints();
+            console.log("EventIndex: Loaded checkpoints",
+                        this.crawlerCheckpoints);
+            return;
+        }
+
+        if (prevState === "PREPARED" && state === "SYNCING") {
+            const addInitialCheckpoints = async () => {
+                const client = MatrixClientPeg.get();
+                const rooms = client.getRooms();
+
+                const isRoomEncrypted = (room) => {
+                    return client.isRoomEncrypted(room.roomId);
+                };
+
+                // We only care to crawl the encrypted rooms, non-encrypted.
+                // rooms can use the search provided by the homeserver.
+                const encryptedRooms = rooms.filter(isRoomEncrypted);
+
+                console.log("EventIndex: Adding initial crawler checkpoints");
+
+                // Gather the prev_batch tokens and create checkpoints for
+                // our message crawler.
+                await Promise.all(encryptedRooms.map(async (room) => {
+                    const timeline = room.getLiveTimeline();
+                    const token = timeline.getPaginationToken("b");
+
+                    console.log("EventIndex: Got token for indexer",
+                                room.roomId, token);
+
+                    const backCheckpoint = {
+                        roomId: room.roomId,
+                        token: token,
+                        direction: "b",
+                    };
+
+                    const forwardCheckpoint = {
+                        roomId: room.roomId,
+                        token: token,
+                        direction: "f",
+                    };
+
+                    await indexManager.addCrawlerCheckpoint(backCheckpoint);
+                    await indexManager.addCrawlerCheckpoint(forwardCheckpoint);
+                    this.crawlerCheckpoints.push(backCheckpoint);
+                    this.crawlerCheckpoints.push(forwardCheckpoint);
+                }));
+            };
+
+            // If our indexer is empty we're most likely running Riot the
+            // first time with indexing support or running it with an
+            // initial sync. Add checkpoints to crawl our encrypted rooms.
+            const eventIndexWasEmpty = await indexManager.isEventIndexEmpty();
+            if (eventIndexWasEmpty) await addInitialCheckpoints();
+
+            // Start our crawler.
+            this.startCrawler();
+            return;
+        }
+
+        if (prevState === "SYNCING" && state === "SYNCING") {
+            // A sync was done, presumably we queued up some live events,
+            // commit them now.
+            console.log("EventIndex: Committing events");
+            await indexManager.commitLiveEvents();
+            return;
+        }
+    }
+
+    onRoomTimeline = async (ev, room, toStartOfTimeline, removed, data) => {
+        // We only index encrypted rooms locally.
+        if (!MatrixClientPeg.get().isRoomEncrypted(room.roomId)) return;
+
+        // If it isn't a live event or if it's redacted there's nothing to
+        // do.
+        if (toStartOfTimeline || !data || !data.liveEvent
+            || ev.isRedacted()) {
+            return;
+        }
+
+        // If the event is not yet decrypted mark it for the
+        // Event.decrypted callback.
+        if (ev.isBeingDecrypted()) {
+            const eventId = ev.getId();
+            this.liveEventsForIndex.add(eventId);
+        } else {
+            // If the event is decrypted or is unencrypted add it to the
+            // index now.
+            await this.addLiveEventToIndex(ev);
+        }
+    }
+
+    onEventDecrypted = async (ev, err) => {
+        const eventId = ev.getId();
+
+        // If the event isn't in our live event set, ignore it.
+        if (!this.liveEventsForIndex.delete(eventId)) return;
+        if (err) return;
+        await this.addLiveEventToIndex(ev);
+    }
+
+    async addLiveEventToIndex(ev) {
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+
+        if (["m.room.message", "m.room.name", "m.room.topic"]
+            .indexOf(ev.getType()) == -1) {
+            return;
+        }
+
+        const e = ev.toJSON().decrypted;
+        const profile = {
+            displayname: ev.sender.rawDisplayName,
+            avatar_url: ev.sender.getMxcAvatarUrl(),
+        };
+
+        indexManager.addEventToIndex(e, profile);
+    }
+
+    async crawlerFunc() {
+        // TODO either put this in a better place or find a library provided
+        // method that does this.
+        const sleep = async (ms) => {
+            return new Promise(resolve => setTimeout(resolve, ms));
+        };
+
+        let cancelled = false;
+
+        console.log("EventIndex: Started crawler function");
+
+        const client = MatrixClientPeg.get();
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+
+        this._crawler = {};
+
+        this._crawler.cancel = () => {
+            cancelled = true;
+        };
+
+        while (!cancelled) {
+            // This is a low priority task and we don't want to spam our
+            // homeserver with /messages requests so we set a hefty timeout
+            // here.
+            await sleep(this._crawlerTimeout);
+
+            console.log("EventIndex: Running the crawler loop.");
+
+            if (cancelled) {
+                break;
+            }
+
+            const checkpoint = this.crawlerCheckpoints.shift();
+
+            /// There is no checkpoint available currently, one may appear if
+            // a sync with limited room timelines happens, so go back to sleep.
+            if (checkpoint === undefined) {
+                continue;
+            }
+
+            console.log("EventIndex: crawling using checkpoint", checkpoint);
+
+            // We have a checkpoint, let us fetch some messages, again, very
+            // conservatively to not bother our homeserver too much.
+            const eventMapper = client.getEventMapper();
+            // TODO we need to ensure to use member lazy loading with this
+            // request so we get the correct profiles.
+            let res;
+
+            try {
+                res = await client._createMessagesRequest(
+                    checkpoint.roomId, checkpoint.token, this._eventsPerCrawl,
+                    checkpoint.direction);
+            } catch (e) {
+                console.log("EventIndex: Error crawling events:", e);
+                this.crawlerCheckpoints.push(checkpoint);
+                continue;
+            }
+
+            if (res.chunk.length === 0) {
+                console.log("EventIndex: Done with the checkpoint", checkpoint);
+                // We got to the start/end of our timeline, lets just
+                // delete our checkpoint and go back to sleep.
+                await indexManager.removeCrawlerCheckpoint(checkpoint);
+                continue;
+            }
+
+            // Convert the plain JSON events into Matrix events so they get
+            // decrypted if necessary.
+            const matrixEvents = res.chunk.map(eventMapper);
+            let stateEvents = [];
+            if (res.state !== undefined) {
+                stateEvents = res.state.map(eventMapper);
+            }
+
+            const profiles = {};
+
+            stateEvents.forEach(ev => {
+                if (ev.event.content &&
+                    ev.event.content.membership === "join") {
+                    profiles[ev.event.sender] = {
+                        displayname: ev.event.content.displayname,
+                        avatar_url: ev.event.content.avatar_url,
+                    };
+                }
+            });
+
+            const decryptionPromises = [];
+
+            matrixEvents.forEach(ev => {
+                if (ev.isBeingDecrypted() || ev.isDecryptionFailure()) {
+                    // TODO the decryption promise is a private property, this
+                    // should either be made public or we should convert the
+                    // event that gets fired when decryption is done into a
+                    // promise using the once event emitter method:
+                    // https://nodejs.org/api/events.html#events_events_once_emitter_name
+                    decryptionPromises.push(ev._decryptionPromise);
+                }
+            });
+
+            // Let us wait for all the events to get decrypted.
+            await Promise.all(decryptionPromises);
+
+            // We filter out events for which decryption failed, are redacted
+            // or aren't of a type that we know how to index.
+            const isValidEvent = (value) => {
+                return ([
+                        "m.room.message",
+                        "m.room.name",
+                        "m.room.topic",
+                    ].indexOf(value.getType()) >= 0
+                        && !value.isRedacted() && !value.isDecryptionFailure()
+                );
+                // TODO do we need to check if the event has all the valid
+                // attributes?
+            };
+
+            // TODO if there are no events at this point we're missing a lot
+            // decryption keys, do we want to retry this checkpoint at a later
+            // stage?
+            const filteredEvents = matrixEvents.filter(isValidEvent);
+
+            // Let us convert the events back into a format that EventIndex can
+            // consume.
+            const events = filteredEvents.map((ev) => {
+                const jsonEvent = ev.toJSON();
+
+                let e;
+                if (ev.isEncrypted()) e = jsonEvent.decrypted;
+                else e = jsonEvent;
+
+                let profile = {};
+                if (e.sender in profiles) profile = profiles[e.sender];
+                const object = {
+                    event: e,
+                    profile: profile,
+                };
+                return object;
+            });
+
+            // Create a new checkpoint so we can continue crawling the room for
+            // messages.
+            const newCheckpoint = {
+                roomId: checkpoint.roomId,
+                token: res.end,
+                fullCrawl: checkpoint.fullCrawl,
+                direction: checkpoint.direction,
+            };
+
+            console.log(
+                "EventIndex: Crawled room",
+                client.getRoom(checkpoint.roomId).name,
+                "and fetched", events.length, "events.",
+            );
+
+            try {
+                const eventsAlreadyAdded = await indexManager.addHistoricEvents(
+                    events, newCheckpoint, checkpoint);
+                // If all events were already indexed we assume that we catched
+                // up with our index and don't need to crawl the room further.
+                // Let us delete the checkpoint in that case, otherwise push
+                // the new checkpoint to be used by the crawler.
+                if (eventsAlreadyAdded === true && newCheckpoint.fullCrawl !== true) {
+                    console.log("EventIndex: Checkpoint had already all events",
+                                "added, stopping the crawl", checkpoint);
+                    await indexManager.removeCrawlerCheckpoint(newCheckpoint);
+                } else {
+                    this.crawlerCheckpoints.push(newCheckpoint);
+                }
+            } catch (e) {
+                console.log("EventIndex: Error durring a crawl", e);
+                // An error occurred, put the checkpoint back so we
+                // can retry.
+                this.crawlerCheckpoints.push(checkpoint);
+            }
+        }
+
+        this._crawler = null;
+
+        console.log("EventIndex: Stopping crawler function");
+    }
+
+    onTimelineReset = async (room, timelineSet, resetAllTimelines) => {
+        if (room === null) return;
+
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+        if (!MatrixClientPeg.get().isRoomEncrypted(room.roomId)) return;
+
+        const timeline = room.getLiveTimeline();
+        const token = timeline.getPaginationToken("b");
+
+        const backwardsCheckpoint = {
+            roomId: room.roomId,
+            token: token,
+            fullCrawl: false,
+            direction: "b",
+        };
+
+        console.log("EventIndex: Added checkpoint because of a limited timeline",
+            backwardsCheckpoint);
+
+        await indexManager.addCrawlerCheckpoint(backwardsCheckpoint);
+
+        this.crawlerCheckpoints.push(backwardsCheckpoint);
+    }
+
+    startCrawler() {
+        if (this._crawler !== null) return;
+        this.crawlerFunc();
+    }
+
+    stopCrawler() {
+        if (this._crawler === null) return;
+        this._crawler.cancel();
+    }
+
+    async close() {
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+        this.removeListeners();
+        this.stopCrawler();
+        return indexManager.closeEventIndex();
+    }
+
+    async search(searchArgs) {
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+        return indexManager.searchEventIndex(searchArgs);
+    }
+}

src/indexing/EventIndexPeg.js

@@ -0,0 +1,115 @@
+/*
+Copyright 2019 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.
+*/
+
+/*
+ * Object holding the global EventIndex object. Can only be initialized if the
+ * platform supports event indexing.
+ */
+
+import PlatformPeg from "../PlatformPeg";
+import EventIndex from "../indexing/EventIndex";
+import SettingsStore from '../settings/SettingsStore';
+
+class EventIndexPeg {
+    constructor() {
+        this.index = null;
+    }
+
+    /**
+     * Create a new EventIndex and initialize it if the platform supports it.
+     *
+     * @return {Promise<bool>} A promise that will resolve to true if an
+     * EventIndex was successfully initialized, false otherwise.
+     */
+    async init() {
+        if (!SettingsStore.isFeatureEnabled("feature_event_indexing")) {
+            return false;
+        }
+
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+        if (!indexManager || await indexManager.supportsEventIndexing() !== true) {
+            console.log("EventIndex: Platform doesn't support event indexing,",
+                        "not initializing.");
+            return false;
+        }
+
+        const index = new EventIndex();
+
+        try {
+            await index.init();
+        } catch (e) {
+            console.log("EventIndex: Error initializing the event index", e);
+            return false;
+        }
+
+        console.log("EventIndex: Successfully initialized the event index");
+
+        this.index = index;
+
+        return true;
+    }
+
+    /**
+     * Get the current event index.
+     *
+     * @return {EventIndex} The current event index.
+     */
+    get() {
+        return this.index;
+    }
+
+    stop() {
+        if (this.index === null) return;
+        this.index.stopCrawler();
+    }
+
+    /**
+     * Unset our event store
+     *
+     * After a call to this the init() method will need to be called again.
+     *
+     * @return {Promise} A promise that will resolve once the event index is
+     * closed.
+     */
+    async unset() {
+        if (this.index === null) return;
+        this.index.close();
+        this.index = null;
+    }
+
+    /**
+     * Delete our event indexer.
+     *
+     * After a call to this the init() method will need to be called again.
+     *
+     * @return {Promise} A promise that will resolve once the event index is
+     * deleted.
+     */
+    async deleteEventIndex() {
+        const indexManager = PlatformPeg.get().getEventIndexingManager();
+
+        if (indexManager !== null) {
+            this.unset();
+            console.log("EventIndex: Deleting event index.");
+            await indexManager.deleteEventIndex();
+        }
+    }
+}
+
+if (!global.mxEventIndexPeg) {
+    global.mxEventIndexPeg = new EventIndexPeg();
+}
+module.exports = global.mxEventIndexPeg;

src/settings/Settings.js

@@ -148,6 +148,12 @@ export const SETTINGS = {
         default: false,
         controller: new ReloadOnChangeController(),
     },
+    "feature_event_indexing": {
+        isFeature: true,
+        supportedLevels: LEVELS_FEATURE,
+        displayName: _td("Enable local event indexing and E2EE search (requires restart)"),
+        default: false,
+    },
     "useCiderComposer": {
         displayName: _td("Use the new, faster, composer for writing messages"),
         supportedLevels: LEVELS_ACCOUNT_SETTINGS,