From f5e43f753f7db3fdad7af8dcb671e696778b7ff7 Mon Sep 17 00:00:00 2001 From: Abdullah Atta Date: Tue, 26 Dec 2023 11:49:10 +0500 Subject: [PATCH] core: auto sync not triggered on note update --- packages/core/src/api/sync/auto-sync.ts | 6 +- packages/core/src/api/sync/index.ts | 96 +++++++++++--------- packages/core/src/database/sql-collection.ts | 8 +- 3 files changed, 62 insertions(+), 48 deletions(-) diff --git a/packages/core/src/api/sync/auto-sync.ts b/packages/core/src/api/sync/auto-sync.ts index 5a5c944dd..ddeac4b83 100644 --- a/packages/core/src/api/sync/auto-sync.ts +++ b/packages/core/src/api/sync/auto-sync.ts @@ -54,7 +54,7 @@ export class AutoSync { this.logger.info(`Auto sync stopped`); } - private schedule(id: string, item?: Item) { + private schedule(id: string | string[], item?: Item) { if ( item && (item.remote || @@ -73,7 +73,9 @@ export class AutoSync { // are equal causing the item to not be synced. const interval = item && item.type === "tiptap" ? 100 : this.interval; this.timeout = setTimeout(() => { - this.logger.info(`Sync requested by: ${id}`); + this.logger.info( + `Sync requested by: ${Array.isArray(id) ? id.join(", ") : id}` + ); this.db.eventManager.publish(EVENTS.databaseSyncRequested, false, false); }, interval) as unknown as number; } diff --git a/packages/core/src/api/sync/index.ts b/packages/core/src/api/sync/index.ts index 4a77eee9a..3b15866d6 100644 --- a/packages/core/src/api/sync/index.ts +++ b/packages/core/src/api/sync/index.ts @@ -102,48 +102,20 @@ class Sync { autoSync = new AutoSync(this.db, 1000); logger = logger.scope("Sync"); syncConnectionMutex = new Mutex(); - connection: signalr.HubConnection; + connection?: signalr.HubConnection; devices = new SyncDevices(this.db.storage, this.db.tokenManager); constructor(private readonly db: Database) { - const tokenManager = new TokenManager(db.storage); - this.connection = new signalr.HubConnectionBuilder() - .withUrl(`${Constants.API_HOST}/hubs/sync/v2`, { - accessTokenFactory: async () => { - const token = await tokenManager.getAccessToken(); - if (!token) throw new Error("Failed to get access token."); - return token; - }, - skipNegotiation: true, - transport: signalr.HttpTransportType.WebSockets, - logger: { - log(level, message) { - const scopedLogger = logger.scope("SignalR::SyncHub"); - switch (level) { - case signalr.LogLevel.Critical: - return scopedLogger.fatal(new Error(message)); - case signalr.LogLevel.Error: { - db.eventManager.publish(EVENTS.syncAborted, message); - return scopedLogger.error(new Error(message)); - } - case signalr.LogLevel.Warning: - return scopedLogger.warn(message); - } - } - } - }) - .withHubProtocol(new MessagePackHubProtocol({ ignoreUndefined: true })) - .build(); - this.connection.serverTimeoutInMilliseconds = 60 * 1000 * 5; EV.subscribe(EVENTS.userLoggedOut, async () => { - await this.connection.stop(); + await this.connection?.stop(); this.autoSync.stop(); }); - - this.connection.on("PushCompleted", () => this.onPushCompleted()); } async start(full?: boolean, force?: boolean) { + this.createConnection(); + + if (!this.connection) return; if (!(await checkSyncStatus(SYNC_CHECK_IDS.sync))) { await this.connection.stop(); return; @@ -208,9 +180,9 @@ class Sync { } let count = 0; - this.connection.off("SendItems"); - this.connection.on("SendItems", async (chunk) => { - if (this.connection.state !== signalr.HubConnectionState.Connected) + this.connection?.off("SendItems"); + this.connection?.on("SendItems", async (chunk) => { + if (this.connection?.state !== signalr.HubConnectionState.Connected) return false; await this.processChunk(chunk, key); @@ -220,7 +192,7 @@ class Sync { return true; }); - const serverResponse = await this.connection.invoke( + const serverResponse = await this.connection?.invoke( "RequestFetch", deviceId ); @@ -235,7 +207,7 @@ class Sync { await this.db.vault.setKey(serverResponse.vaultKey); } - this.connection.off("SendItems"); + this.connection?.off("SendItems"); } async send(deviceId: string, isForceSync?: boolean) { @@ -246,7 +218,7 @@ class Sync { for await (const item of this.collector.collect(100, isForceSync)) { if (!isSyncInitialized) { const vaultKey = await this.db.vault.getKey(); - await this.connection.send("InitializePush", { + await this.connection?.send("InitializePush", { vaultKey, synced: false }); @@ -266,7 +238,7 @@ class Sync { } } if (!isSyncInitialized) return false; - await this.connection.send("PushCompleted"); + await this.connection?.send("PushCompleted"); return true; } @@ -281,7 +253,7 @@ class Sync { async cancel() { this.logger.info("Sync canceled"); - await this.connection.stop(); + await this.connection?.stop(); } /** @@ -355,13 +327,51 @@ class Sync { private async pushItem(deviceId: string, item: SyncTransferItem) { await this.checkConnection(); - return (await this.connection.invoke("PushItems", deviceId, item)) === 1; + return (await this.connection?.invoke("PushItems", deviceId, item)) === 1; + } + + private createConnection() { + if (this.connection) return; + + const tokenManager = new TokenManager(this.db.storage); + this.connection = new signalr.HubConnectionBuilder() + .withUrl(`${Constants.API_HOST}/hubs/sync/v2`, { + accessTokenFactory: async () => { + const token = await tokenManager.getAccessToken(); + if (!token) throw new Error("Failed to get access token."); + return token; + }, + skipNegotiation: true, + transport: signalr.HttpTransportType.WebSockets, + logger: { + log: (level, message) => { + const scopedLogger = logger.scope("SignalR::SyncHub"); + switch (level) { + case signalr.LogLevel.Critical: + return scopedLogger.fatal(new Error(message)); + case signalr.LogLevel.Error: { + this.db.eventManager.publish(EVENTS.syncAborted, message); + return scopedLogger.error(new Error(message)); + } + case signalr.LogLevel.Warning: + return scopedLogger.warn(message); + } + } + } + }) + .withHubProtocol(new MessagePackHubProtocol({ ignoreUndefined: true })) + .build(); + this.connection.serverTimeoutInMilliseconds = 60 * 1000 * 5; + this.connection.on("PushCompleted", () => this.onPushCompleted()); } private async checkConnection() { await this.syncConnectionMutex.runExclusive(async () => { try { - if (this.connection.state !== signalr.HubConnectionState.Connected) { + if ( + this.connection && + this.connection.state !== signalr.HubConnectionState.Connected + ) { if ( this.connection.state !== signalr.HubConnectionState.Disconnected ) { diff --git a/packages/core/src/database/sql-collection.ts b/packages/core/src/database/sql-collection.ts index ac19d2c8e..778aa99c7 100644 --- a/packages/core/src/database/sql-collection.ts +++ b/packages/core/src/database/sql-collection.ts @@ -69,7 +69,6 @@ export class SQLCollection< async upsert(item: SQLiteItem) { if (!item.id) throw new Error("The item must contain the id field."); if (!item.deleted) item.dateCreated = item.dateCreated || Date.now(); - this.eventManager.publish(EVENTS.databaseUpdated, item.id, item); // if item is newly synced, remote will be true. if (!item.remote) { @@ -83,10 +82,11 @@ export class SQLCollection< .replaceInto(this.type) .values(item) .execute(); + + this.eventManager.publish(EVENTS.databaseUpdated, item.id, item); } async softDelete(ids: string[]) { - this.eventManager.publish(EVENTS.databaseUpdated, ids); await this.db() .replaceInto(this.type) .values( @@ -98,14 +98,15 @@ export class SQLCollection< })) ) .execute(); + this.eventManager.publish(EVENTS.databaseUpdated, ids); } async delete(ids: string[]) { - this.eventManager.publish(EVENTS.databaseUpdated, ids); await this.db() .deleteFrom(this.type) .where("id", "in", ids) .execute(); + this.eventManager.publish(EVENTS.databaseUpdated, ids); } async exists(id: string) { @@ -175,6 +176,7 @@ export class SQLCollection< synced: partial.synced || false }) .execute(); + this.eventManager.publish(EVENTS.databaseUpdated, ids); } async ids(sortOptions: GroupOptions): Promise {