Sync in worker

src/matrix/Client.js

@@ -32,6 +32,7 @@ import {SSOLoginHelper} from "./login/SSOLoginHelper";
 import {getDehydratedDevice} from "./e2ee/Dehydration.js";
 import {Registration} from "./registration/Registration";
 import {FeatureSet} from "../features";
+import {SyncInWorker} from "./SyncInWorker";
 
 export const LoadStatus = createEnum(
     "NotLoading",
@@ -291,7 +292,7 @@ export class Client {
             await log.wrap("createIdentity", log => this._session.createIdentity(log));
         }
 
-        this._sync = new Sync({hsApi: this._requestScheduler.hsApi, storage: this._storage, session: this._session, logger: this._platform.logger});
+        this._sync = new SyncInWorker({hsApi: this._requestScheduler.hsApi, storage: this._storage, session: this._session, logger: this._platform.logger});
         // notify sync and session when back online
         this._reconnectSubscription = this._reconnector.connectionStatus.subscribe(state => {
             if (state === ConnectionStatus.Online) {

src/matrix/SyncInWorker.ts

@@ -0,0 +1,42 @@
+import {Sync, SyncStatus} from "./Sync";
+import {HomeServerApi} from "./net/HomeServerApi";
+import {Session} from "./Session";
+import {Storage} from "./storage/idb/Storage";
+import {Logger} from "../logging/Logger";
+import {WorkerFacade} from "../platform/web/worker-sync/worker/WorkerFacade";
+import {StartSyncRequest} from "../platform/web/worker-sync/SyncWorker";
+
+interface SyncOptions {
+    hsApi: HomeServerApi,
+    session: Session,
+    storage: Storage,
+    logger: Logger
+}
+
+export class SyncInWorker extends Sync {
+    private _worker: WorkerFacade;
+
+    constructor(options: SyncOptions) {
+        super(options);
+        this._worker = new WorkerFacade;
+    }
+
+    get status(): SyncStatus {
+        return super.status;
+    }
+
+    get error(): Error {
+        return super.error;
+    }
+
+    start(): void {
+        const message = new StartSyncRequest({
+            sessionInfo: super.options.sessionInfo,
+        });
+        const result = this._worker.sendAndWaitForReply(message);
+    }
+
+    stop(): void {
+        super.stop();
+    }
+}

src/matrix/sessioninfo/localstorage/SessionInfoStorage.ts

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-interface ISessionInfo {
+export interface ISessionInfo {
     id: string;
     deviceId: string;
     userId: string;

src/matrix/storage/idb/StorageFactory.ts

@@ -52,16 +52,19 @@ async function requestPersistedStorage(): Promise<boolean> {
 }
 
 export class StorageFactory {
-    private _serviceWorkerHandler: ServiceWorkerHandler;
+    private _serviceWorkerHandler?: ServiceWorkerHandler;
     private _idbFactory: IDBFactory;
     private _IDBKeyRange: typeof IDBKeyRange;
-    private _localStorage: IDOMStorage;
+    private _localStorage?: IDOMStorage;
 
-    constructor(serviceWorkerHandler: ServiceWorkerHandler, idbFactory: IDBFactory = window.indexedDB, _IDBKeyRange = window.IDBKeyRange, localStorage: IDOMStorage = window.localStorage) {
+    constructor(serviceWorkerHandler?: ServiceWorkerHandler, idbFactory: IDBFactory = indexedDB, _IDBKeyRange = IDBKeyRange, _localStorage: IDOMStorage | undefined = undefined) {
         this._serviceWorkerHandler = serviceWorkerHandler;
         this._idbFactory = idbFactory;
         this._IDBKeyRange = _IDBKeyRange;
-        this._localStorage = localStorage;
+        if (typeof window !== 'undefined' && _localStorage === undefined) {
+            _localStorage = localStorage;
+        }
+        this._localStorage = _localStorage;
     }
 
     async create(sessionId: string, log: ILogItem): Promise<Storage> {

src/platform/web/Platform.js

@@ -43,6 +43,7 @@ import {MediaDevicesWrapper} from "./dom/MediaDevices";
 import {DOMWebRTC} from "./dom/WebRTC";
 import {ThemeLoader} from "./theming/ThemeLoader";
 import {TimeFormatter} from "./dom/TimeFormatter";
+import {SyncWorkerPool} from "./worker-sync/SyncWorkerPool";
 
 function addScript(src) {
     return new Promise(function (resolve, reject) {
@@ -149,6 +150,7 @@ export class Platform {
             this._serviceWorkerHandler = new ServiceWorkerHandler();
             this._serviceWorkerHandler.registerAndStart(assetPaths.serviceWorker);
         }
+        this._syncWorkerPool = null;
         this.notificationService = undefined;
         // Only try to use crypto when olm is provided
         if(this._assetPaths.olm) {
@@ -173,6 +175,11 @@ export class Platform {
         this.mediaDevices = new MediaDevicesWrapper(navigator.mediaDevices);
         this.webRTC = new DOMWebRTC();
         this._themeLoader = import.meta.env.DEV? null: new ThemeLoader(this);
+
+        if (assetPaths.syncWorker && window.Worker) {
+            this._syncWorkerPool = new SyncWorkerPool(this._assetPaths.syncWorker, this.sessionInfoStorage);
+            this._syncWorkerPool.add("750343279252151");
+        }
     }
 
     async init() {

src/platform/web/sdk/paths/vite.js

@@ -2,6 +2,7 @@
 import _downloadSandboxPath from "../../assets/download-sandbox.html?url";
 // @ts-ignore
 import _workerPath from "../../worker/main.js?url";
+import _syncWorkerPath from "../../worker-sync/sync-worker.js?url";
 // @ts-ignore
 import olmWasmPath from "@matrix-org/olm/olm.wasm?url";
 // @ts-ignore
@@ -12,6 +13,7 @@ import olmLegacyJsPath from "@matrix-org/olm/olm_legacy.js?url";
 export default {
     downloadSandbox: _downloadSandboxPath,
     worker: _workerPath,
+    syncWorker: _syncWorkerPath,
     olm: {
         wasm: olmWasmPath,
         legacyBundle: olmLegacyJsPath,

src/platform/web/worker-sync/SyncWorker.ts

@@ -0,0 +1,115 @@
+import {ISessionInfo} from "../../../matrix/sessioninfo/localstorage/SessionInfoStorage";
+import {HomeServerApi} from "../../../matrix/net/HomeServerApi";
+import {createFetchRequest} from "../dom/request/fetch";
+import {Reconnector} from "../../../matrix/net/Reconnector";
+import {ExponentialRetryDelay} from "../../../matrix/net/ExponentialRetryDelay";
+import {OnlineStatus} from "../dom/OnlineStatus";
+import {Sync} from "../../../matrix/Sync";
+import {Session} from "../../../matrix/Session";
+import {WorkerPlatform} from "./WorkerPlatform";
+import {StorageFactory} from "../../../matrix/storage/idb/StorageFactory";
+import {MediaRepository} from "../../../matrix/net/MediaRepository";
+import {FeatureSet} from "../../../features";
+import {Logger} from "../../../logging/Logger";
+import {ConsoleReporter} from "../../../logging/ConsoleReporter";
+import assetPaths from "../sdk/paths/vite";
+import {Request, RequestData, ResponseData, Worker} from "./worker/Worker";
+import {RequestFunction} from "../../types/types";
+
+enum SyncRequestType {
+    StartSync = "StartSync",
+}
+
+interface StartSyncRequestData extends RequestData {
+    sessionInfo: ISessionInfo,
+}
+
+interface StartSyncResponseData extends ResponseData {
+    success: boolean,
+}
+
+export class StartSyncRequest implements Request {
+    readonly type: SyncRequestType;
+    readonly data: StartSyncRequestData;
+
+    constructor(data: StartSyncRequestData) {
+        this.type = SyncRequestType.StartSync;
+        this.data = data;
+    }
+}
+
+export class SyncWorker extends Worker {
+    private readonly _reconnector: Reconnector;
+    private readonly _request: RequestFunction;
+    private readonly _platform: WorkerPlatform;
+    private readonly _storageFactory: StorageFactory;
+    private readonly _logger: Logger;
+    private readonly _features: FeatureSet;
+    private _sync: Sync;
+    private _olm: any;
+    private _olmWorker: any;
+
+    constructor() {
+        super();
+        this._platform = new WorkerPlatform({assetPaths});
+        this._request = createFetchRequest(this._platform.clock.createTimeout);
+        this._reconnector = new Reconnector({
+            onlineStatus: new OnlineStatus(),
+            retryDelay: new ExponentialRetryDelay(this._platform.clock.createTimeout),
+            createMeasure: this._platform.clock.createMeasure
+        });
+        this._storageFactory = new StorageFactory();
+        this._logger = new Logger({platform: this._platform});
+        this._logger.addReporter(new ConsoleReporter());
+        this._features = new FeatureSet;
+    }
+
+    async init() {
+        this._olm = this._platform.loadOlm();
+        this._olmWorker = await this._platform.loadOlmWorker();
+
+        super.addHandler(SyncRequestType.StartSync, this.startSync.bind(this));
+    }
+
+    async startSync(request: StartSyncRequest): Promise<StartSyncResponseData> {
+        const sessionInfo = request.data.sessionInfo;
+        console.log(`Starting sync worker for session with id ${sessionInfo.id}`);
+
+        const hsApi = new HomeServerApi({
+            request: this._request,
+            reconnector: this._reconnector,
+            homeserver: sessionInfo.homeserver,
+            accessToken: sessionInfo.accessToken,
+        });
+
+        let storage;
+        await this._logger.run("", async log => {
+            storage = await this._storageFactory.create(sessionInfo.id, log)
+        });
+
+        const mediaRepository = new MediaRepository({
+            platform: this._platform,
+            homeserver: sessionInfo.homeServer,
+        });
+
+        const session = new Session({
+            platform: this._platform,
+            features: this._features,
+            olm: this._olm,
+            olmWorker: this._olmWorker,
+            storage,
+            hsApi,
+            sessionInfo,
+            mediaRepository,
+        });
+
+        this._sync = new Sync({
+            logger: this._logger,
+            hsApi,
+            session,
+            storage,
+        });
+
+        return { success: true };
+    }
+}

src/platform/web/worker-sync/SyncWorkerPool.ts

@@ -0,0 +1,48 @@
+import {SessionInfoStorage} from "../../../matrix/sessioninfo/localstorage/SessionInfoStorage";
+import {StartSyncRequest} from "./SyncWorker";
+
+export type SessionId = string;
+
+export class SyncWorkerPool {
+    private readonly _workers: Map<SessionId, Worker> = new Map;
+    private readonly _path: string;
+    private readonly _sessionInfoStorage: SessionInfoStorage;
+
+    constructor(path: string, sessionInfoStorage: SessionInfoStorage) {
+        this._path = path;
+        this._sessionInfoStorage = sessionInfoStorage;
+    }
+
+    async add(sessionId: SessionId) {
+        if (this._workers.size > 0) {
+            throw "Currently there can only be one active sync worker";
+        }
+
+        if (this._workers.has(sessionId)) {
+            throw `Session with id ${sessionId} already has a sync worker`;
+        }
+
+        const worker = new Worker(this._path, {type: "module"});
+        this._workers.set(sessionId, worker);
+        worker.onmessage = event => {
+            const data = event.data;
+            console.log(data);
+        }
+
+        const sessionInfo = await this._sessionInfoStorage.get(sessionId);
+        if (!sessionInfo) {
+            // TODO
+            return;
+        }
+
+        const message = new StartSyncRequest({sessionInfo})
+        worker.postMessage({
+            type: message.type,
+            data: message.data,
+        });
+    }
+
+    remove(sessionId: SessionId) {
+        // TODO
+    }
+}

src/platform/web/worker-sync/WorkerPlatform.ts

@@ -0,0 +1,23 @@
+import {Clock} from "../dom/Clock";
+
+export class WorkerPlatform {
+    private readonly _clock: Clock;
+    private _assetPaths: any;
+
+    constructor({assetPaths}) {
+        this._assetPaths = assetPaths;
+        this._clock = new Clock;
+    }
+
+    loadOlm() {
+        return null;
+    }
+
+    async loadOlmWorker() {
+        return null;
+    }
+
+    get clock(): Clock {
+        return this._clock;
+    }
+}

src/platform/web/worker-sync/sync-worker.js

@@ -0,0 +1,5 @@
+import {SyncWorker} from "./SyncWorker";
+
+const worker = new SyncWorker();
+void worker.start();
+self.syncWorker = worker;