diff --git a/packages/core/api/sync/auto-sync.js b/packages/core/api/sync/auto-sync.js index 32ba5923a..65f1ad415 100644 --- a/packages/core/api/sync/auto-sync.js +++ b/packages/core/api/sync/auto-sync.js @@ -10,11 +10,14 @@ export class AutoSync { this.db = db; this.interval = interval; this.timeout = null; + this.isAutoSyncing = false; } async start() { if (!(await checkIsUserPremium(CHECK_IDS.databaseSync))) return; + if (this.isAutoSyncing) return; + this.isAutoSyncing = true; this.databaseUpdatedEvent = this.db.eventManager.subscribeSingle( EVENTS.databaseUpdated, this.schedule.bind(this) @@ -22,6 +25,7 @@ export class AutoSync { } stop() { + this.isAutoSyncing = false; clearTimeout(this.timeout); if (this.databaseUpdatedEvent) this.databaseUpdatedEvent.unsubscribe(); } @@ -29,9 +33,12 @@ export class AutoSync { /** * @private */ - schedule() { - this.stop(); + schedule(id, item) { + if (item && item.remote) return; + + clearTimeout(this.timeout); this.timeout = setTimeout(() => { + console.log("SYNC REQUESTED by", id); this.db.eventManager.publish(EVENTS.databaseSyncRequested, false, false); }, this.interval); } diff --git a/packages/core/api/sync/collector.js b/packages/core/api/sync/collector.js index 2fd024b08..f6b2cd857 100644 --- a/packages/core/api/sync/collector.js +++ b/packages/core/api/sync/collector.js @@ -49,7 +49,7 @@ class Collector { if (item.localOnly) { prev.push({ id: item.id, deleted: true, dateModified: Date.now() }); } else if ( - item.dateModified > this._lastSyncedTimestamp || + (item.dateModified > this._lastSyncedTimestamp && !item.synced) || item.migrated ) { prev.push(item); @@ -75,6 +75,7 @@ class Collector { delete item.resolved; // turn the migrated flag off so we don't keep syncing this item repeated delete item.migrated; + delete item.synced; return { id: item.id, diff --git a/packages/core/api/sync/index.js b/packages/core/api/sync/index.js index 37259144e..054dca78d 100644 --- a/packages/core/api/sync/index.js +++ b/packages/core/api/sync/index.js @@ -15,7 +15,6 @@ import Conflicts from "./conflicts"; import { SyncQueue } from "./syncqueue"; import { AutoSync } from "./auto-sync"; import { toChunks } from "../../utils/array"; -import id from "../../utils/id"; const ITEM_TYPE_MAP = { attachments: "attachment", @@ -27,7 +26,6 @@ const ITEM_TYPE_MAP = { /** * @typedef {{ - * syncId: string, * item: string, * itemType: string, * lastSynced: number, @@ -44,7 +42,6 @@ const ITEM_TYPE_MAP = { * lastSynced: number, * current: number, * total: number, - * syncId: string * }} BatchedSyncTransferItem */ @@ -58,14 +55,12 @@ export default class SyncManager { this.syncMutex = new Mutex(); } - start(full, force) { + async start(full, force) { if (this.syncMutex.isLocked()) return false; - return this.syncMutex - .runExclusive(() => { - this.sync.autoSync.stop(); - return this.sync.start(full, force); - }) - .finally(() => this.sync.autoSync.start()); + return this.syncMutex.runExclusive(async () => { + await this.sync.autoSync.start(); + return this.sync.start(full, force); + }); } async acquireLock(callback) { @@ -86,13 +81,11 @@ class Sync { */ constructor(db) { this.db = db; - this.runningSyncs = {}; this.conflicts = new Conflicts(db); this.collector = new Collector(db); this.queue = new SyncQueue(db.storage); this.merger = new Merger(db); this.autoSync = new AutoSync(db, 1000); - this.syncId = null; const tokenManager = new TokenManager(db.storage); this.connection = new signalr.HubConnectionBuilder() @@ -107,13 +100,11 @@ class Sync { }); this.connection.on("SyncItem", async (syncStatus) => { - this.autoSync.stop(); await this.onSyncItem.call(this, syncStatus); - await this.autoSync.start(); }); - this.connection.on("RemoteSyncCompleted", (syncId) => - this.onRemoteSyncCompleted(syncId) + this.connection.on("RemoteSyncCompleted", () => + this.onRemoteSyncCompleted() ); } @@ -123,18 +114,16 @@ class Sync { * @param {boolean} force * @param {Object} ignoredIds */ - async start(full, force, ignoredIds) { + async start(full, force) { this.connection.onclose(() => { throw new Error("Connection closed."); }); - this.syncId = this.getSyncId(); - const { lastSynced, oldLastSynced } = await this.init(force); - const serverResponse = full ? await this.fetch(lastSynced) : null; + const { newLastSynced, data } = await this.collect(lastSynced); - const { newLastSynced, data } = await this.collect(lastSynced, ignoredIds); + const serverResponse = full ? await this.fetch(lastSynced) : null; if (await this.send(data, newLastSynced)) { await this.stop(newLastSynced); @@ -188,12 +177,10 @@ class Sync { return serverResponse; } - async collect(lastSynced, ignoredIds) { + async collect(lastSynced) { const newLastSynced = Date.now(); let data = await this.collector.collect(lastSynced); - if (ignoredIds) - data = this.collector.filter(data, (item) => !ignoredIds[item.id]); let { syncedAt } = await this.queue.get(); if (syncedAt) { @@ -221,6 +208,8 @@ class Sync { if (areAllEmpty(data)) return false; const { itemIds } = await this.queue.get(); + if (!itemIds) return false; + const total = itemIds.length; const arrays = itemIds.reduce( @@ -264,7 +253,6 @@ class Sync { total, items, types, - syncId: this.syncId, }); if (result) { @@ -273,20 +261,17 @@ class Sync { this.db.eventManager, "upload", total, - index + ids.length, - items + index + ids.length ); } } - return await this.connection.invoke( - "SyncCompleted", - lastSynced, - this.syncId - ); + return await this.connection.invoke("SyncCompleted", lastSynced); } async stop(lastSynced) { - await this.db.storage.write("lastSynced", lastSynced); + const storedLastSynced = await this.db.lastSynced(); + if (lastSynced > storedLastSynced) + await this.db.storage.write("lastSynced", lastSynced); this.db.eventManager.publish(EVENTS.syncCompleted); } @@ -321,10 +306,8 @@ class Sync { /** * @private */ - async onRemoteSyncCompleted(syncId) { - const ignoredIds = this.runningSyncs[syncId]; - await this.start(false, false, ignoredIds); - this.runningSyncs[syncId] = {}; + async onRemoteSyncCompleted() { + await this.start(false, false); } /** @@ -332,16 +315,9 @@ class Sync { * @private */ async onSyncItem(syncStatus) { - const { current, id: syncId, item: itemJSON, itemType, total } = syncStatus; + const { current, item: itemJSON, itemType, total } = syncStatus; const item = JSON.parse(itemJSON); - if (syncId) { - this.runningSyncs[syncId] = { - ...this.runningSyncs[syncId], - [item.id]: true, - }; - } - await this.merger.mergeItem(itemType, item); sendSyncProgressEvent(this.db.eventManager, "download", total, current); } @@ -357,8 +333,4 @@ class Sync { const result = await this.connection.invoke("SyncItem", batch); return result === 1; } - - getSyncId() { - return id(); - } } diff --git a/packages/core/api/sync/merger.js b/packages/core/api/sync/merger.js index d0a6ab126..4c6e4b33d 100644 --- a/packages/core/api/sync/merger.js +++ b/packages/core/api/sync/merger.js @@ -122,6 +122,7 @@ class Merger { const deserialized = JSON.parse(decrypted); deserialized.remote = true; + deserialized.synced = true; if (!migrate) return deserialized; return this._migrate(deserialized, item.v); } diff --git a/packages/core/database/indexed-collection.js b/packages/core/database/indexed-collection.js index 3887b7071..eb83c2168 100644 --- a/packages/core/database/indexed-collection.js +++ b/packages/core/database/indexed-collection.js @@ -29,11 +29,13 @@ export default class IndexedCollection { async updateItem(item) { if (!item.id) throw new Error("The item must contain the id field."); - this.eventManager.publish(EVENTS.databaseUpdated, item.remote); + this.eventManager.publish(EVENTS.databaseUpdated, item.id, item); // if item is newly synced, remote will be true. - if (!item.remote) item.dateModified = Date.now(); - + if (!item.remote) { + item.dateModified = Date.now(); + item.synced = false; + } // the item has become local now, so remove the flags delete item.remote; delete item.migrated;