Support older sync services

Author: simolus3

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -149,7 +149,7 @@ export const isPowerSyncDatabaseOptionsWithSettings = (test: any): test is Power
 /**
  * The priority used by the core extension to indicate that a full sync was completed.
  */
-const fullSyncPriority = 2147483647;
+const FULL_SYNC_PRIORITY = 2147483647;
 
 export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDBListener> {
   /**
@@ -357,7 +357,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     for (const { priority, last_synced_at } of result) {
       const parsedDate = new Date(last_synced_at + 'Z');
 
-      if (priority == fullSyncPriority) {
+      if (priority == FULL_SYNC_PRIORITY) {
         // This lowest-possible priority represents a complete sync.
         lastCompleteSync = parsedDate;
       } else {

packages/common/src/client/sync/bucket/BucketStorageAdapter.ts

@@ -32,7 +32,7 @@ export interface SyncLocalDatabaseResult {
 
 export interface BucketChecksum {
   bucket: string;
-  priority: number;
+  priority?: number;
   /**
    * 32-bit unsigned hash.
    */

packages/common/src/client/sync/bucket/SqliteBucketStorage.ts

@@ -4,6 +4,7 @@ import { DBAdapter, Transaction, extractTableUpdates } from '../../../db/DBAdapt
 import { BaseObserver } from '../../../utils/BaseObserver.js';
 import { MAX_OP_ID } from '../../constants.js';
 import {
+  BucketChecksum,
   BucketState,
   BucketStorageAdapter,
   BucketStorageListener,
@@ -147,7 +148,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
 
     const buckets = checkpoint.buckets;
     if (priority !== undefined) {
-      buckets.filter((b) => b.priority <= priority);
+      buckets.filter((b) => hasMatchingPriority(priority, b));
     }
     const bucketNames = buckets.map((b) => b.bucket);
     await this.writeTransaction(async (tx) => {
@@ -185,7 +186,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     if (priority !== undefined) {
       const affectedBuckets: string[] = [];
       for (const desc of checkpoint.buckets) {
-        if (desc.priority <= priority) {
+        if (hasMatchingPriority(priority, desc)) {
           affectedBuckets.push(desc.bucket);
         }
       }
@@ -205,7 +206,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
   async validateChecksums(checkpoint: Checkpoint, priority: number | undefined): Promise<SyncLocalDatabaseResult> {
     if (priority !== undefined) {
       // Only validate the buckets within the priority we care about
-      const newBuckets = checkpoint.buckets.filter((cs) => cs.priority <= priority);
+      const newBuckets = checkpoint.buckets.filter((cs) => hasMatchingPriority(priority, cs));
       checkpoint = {...checkpoint, buckets: newBuckets};
     }
 
@@ -391,3 +392,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     // No-op for now
   }
 }
+
+function hasMatchingPriority(priority: number, bucket: BucketChecksum) {
+  return bucket.priority != null && bucket.priority <= priority;
+}

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -146,6 +146,12 @@ export const DEFAULT_STREAM_CONNECTION_OPTIONS: RequiredPowerSyncConnectionOptio
   params: {}
 };
 
+// The priority we assume when we receive checkpoint lines where no priority is set.
+// This is the default priority used by the sync service, but can be set to an arbitrary
+// value since sync services without priorities also won't send partial sync completion
+// messages.
+const FALLBACK_PRIORITY = 3;
+
 export abstract class AbstractStreamingSyncImplementation
   extends BaseObserver<StreamingSyncImplementationListener>
   implements StreamingSyncImplementation
@@ -551,7 +557,7 @@ The next upload iteration will be delayed.`);
             const bucketsToDelete = new Set<string>(bucketMap.keys());
             const newBuckets = new Map<string, BucketDescription>();
             for (const checksum of line.checkpoint.buckets) {
-              newBuckets.set(checksum.bucket, { name: checksum.bucket, priority: checksum.priority });
+              newBuckets.set(checksum.bucket, { name: checksum.bucket, priority: checksum.priority ?? FALLBACK_PRIORITY });
               bucketsToDelete.delete(checksum.bucket);
             }
             if (bucketsToDelete.size > 0) {
@@ -641,7 +647,7 @@ The next upload iteration will be delayed.`);
             newBuckets.forEach((checksum, name) =>
               bucketMap.set(name, {
                 name: checksum.bucket,
-                priority: checksum.priority
+                priority: checksum.priority ?? FALLBACK_PRIORITY
               })
             );