core: rerun sync if there are unsynced changes after sync complete

streetwriters · May 8, 2024 · 4819e17 · 4819e17
1 parent 69a5d5a
commit 4819e17
Show file tree

Hide file tree

Showing 6 changed files with 50 additions and 12 deletions.
diff --git a/packages/core/__e2e__/sync.test.js b/packages/core/__e2e__/sync.test.js
@@ -109,16 +109,17 @@ test(
     });
 
     const id = await deviceA.notes.add({ title: "hello" });
-    for (let i = 0; i < 10; ++i) {
+    for (let i = 0; i < 5; ++i) {
+      if (i > 0) await deviceA.notes.add({ id, title: `edit ${i - 1}` });
       await Promise.all([
         deviceA.sync({ type: "send" }),
-        new Promise((resolve) => setTimeout(resolve), 100).then(() =>
+        new Promise((resolve) => setTimeout(resolve, 40)).then(() =>
           deviceA.notes.add({ id, title: `edit ${i}` })
         )
       ]);
 
-      expect((await deviceA.notes.note(id))?.synced).toBe(false);
-      await deviceA.sync({ type: "send" });
+      expect((await deviceA.notes.note(id))?.title).toBe(`edit ${i}`);
+      expect((await deviceA.notes.note(id))?.synced).toBe(true);
       await deviceB.sync({ type: "fetch" });
       expect((await deviceB.notes.note(id))?.title).toBe(`edit ${i}`);
     }
@@ -139,17 +140,17 @@ test(
       await cleanup(deviceA, deviceB);
     });
 
-    for (let i = 0; i < 10; ++i) {
+    const id = await deviceA.notes.add({ title: "hello" });
+    for (let i = 0; i < 5; ++i) {
+      if (i > 0) await deviceA.notes.add({ id, title: `edit ${i - 1}` });
       await Promise.all([
         deviceA.sync({ type: "send" }),
-        new Promise((resolve) => setTimeout(resolve), 100).then(() =>
+        new Promise((resolve) => setTimeout(resolve, 40)).then(() =>
           deviceA.notes.add({ title: `note ${i}` })
         )
       ]);
-      expect(await deviceB.notes.all.count()).toBe(i);
-      await deviceA.sync({ type: "send" });
       await deviceB.sync({ type: "fetch" });
-      expect(await deviceB.notes.all.count()).toBe(i + 1);
+      expect(await deviceB.notes.all.count()).toBe(i + 2);
     }
   },
   TEST_TIMEOUT * 10

diff --git a/packages/core/src/api/sync/collector.ts b/packages/core/src/api/sync/collector.ts
@@ -33,6 +33,15 @@ class Collector {
   logger = logger.scope("SyncCollector");
   constructor(private readonly db: Database) {}
 
+  async hasUnsyncedChanges() {
+    for (const itemType of SYNC_ITEM_TYPES) {
+      const collectionKey = SYNC_COLLECTIONS_MAP[itemType];
+      const collection = this.db[collectionKey].collection;
+      if ((await collection.unsyncedCount()) > 0) return true;
+    }
+    return false;
+  }
+
   async *collect(
     chunkSize: number,
     isForceSync = false

diff --git a/packages/core/src/api/sync/index.ts b/packages/core/src/api/sync/index.ts
@@ -160,7 +160,7 @@ class Sync {
     )
       this.logger.info("New data sent");
 
-    await this.stop();
+    await this.stop(options);
 
     if (!(await checkSyncStatus(SYNC_CHECK_IDS.autoSync))) {
       await this.connection.stop();
@@ -265,7 +265,15 @@ class Sync {
     return true;
   }
 
-  async stop() {
+  async stop(options: SyncOptions) {
+    if (
+      (options.type === "send" || options.type === "full") &&
+      (await this.collector.hasUnsyncedChanges())
+    ) {
+      this.logger.info("Changes made during last sync. Syncing again...");
+      await this.start({ type: "send" });
+      return;
+    }
     // refresh monographs
     await this.db.monographs.refresh().catch(this.logger.error);
     // update trash cache

diff --git a/packages/core/src/database/index.ts b/packages/core/src/database/index.ts
@@ -167,6 +167,7 @@ export interface DatabaseCollection<T, IsAsync extends boolean> {
   delete(ids: string[]): Promise<void>;
   exists(id: string): AsyncOrSyncResult<IsAsync, boolean>;
   count(): AsyncOrSyncResult<IsAsync, number>;
+  unsyncedCount(): Promise<number>;
   get(id: string): AsyncOrSyncResult<IsAsync, T | undefined>;
   put(items: (T | undefined)[]): Promise<SQLiteItem<T>[]>;
   update(ids: string[], partial: Partial<T>): Promise<void>;

diff --git a/packages/core/src/database/sql-cached-collection.ts b/packages/core/src/database/sql-cached-collection.ts
@@ -21,7 +21,7 @@ import { MaybeDeletedItem, isDeleted } from "../types";
 import EventManager from "../utils/event-manager";
 import { DatabaseAccessor, DatabaseCollection, DatabaseSchema } from ".";
 import { SQLCollection } from "./sql-collection";
-import { Kysely, Transaction } from "kysely";
+import { Kysely } from "kysely";
 import { Sanitizer } from "./sanitizer";
 
 export class SQLCachedCollection<
@@ -171,6 +171,10 @@ export class SQLCachedCollection<
     }
   }
 
+  async unsyncedCount() {
+    return this.collection.unsyncedCount();
+  }
+
   // has(id: string) {
   //   return this.cache.has(id);
   // }

diff --git a/packages/core/src/database/sql-collection.ts b/packages/core/src/database/sql-collection.ts
@@ -278,6 +278,21 @@ export class SQLCollection<
     return items;
   }
 
+  async unsyncedCount() {
+    const { count } =
+      (await this.db()
+        .selectFrom<keyof DatabaseSchema>(this.type)
+        .select((a) => a.fn.count<number>("id").as("count"))
+        .where(isFalse("synced"))
+        .$if(this.type === "attachments", (eb) =>
+          eb.where((eb) =>
+            eb.or([eb("dateUploaded", ">", 0), eb("deleted", "==", true)])
+          )
+        )
+        .executeTakeFirst()) || {};
+    return count || 0;
+  }
+
   async *unsynced(
     chunkSize: number,
     forceSync?: boolean