wip: uploads

Author: stevensJourney

packages/lite-sdk/src/client/storage/BucketStorage.ts

@@ -9,6 +9,11 @@ export type SavedProgress = {
   sinceLast: number;
 };
 
+export interface LocalState {
+  lastOpId: bigint;
+  targetOpId: bigint;
+}
+
 export type BucketOperationProgress = Record<string, SavedProgress>;
 
 /**
@@ -54,6 +59,13 @@ export interface BucketStorage {
    */
   getBucketStates: () => Promise<Array<BucketState>>;
 
+  /**
+   * Get the current state of the local bucket.
+   * FIXME maybe a better name?
+   * @returns The local bucket state, containing bucket name, operation ID, checksums, etc.
+   */
+  getLocalState: () => Promise<LocalState>;
+
   /**
    * Synchronize the local database with a checkpoint.
    * Validates checksums, applies operations, and updates bucket states.

packages/lite-sdk/src/client/storage/MemoryBucketStorageImpl.ts

@@ -2,7 +2,7 @@ import type { BucketState, Checkpoint } from '@powersync/service-core';
 import { Lock } from '../../utils/Lock.js';
 import { CrudManager } from '../sync/CrudManager.js';
 import { SystemDependencies } from '../system/SystemDependencies.js';
-import type { BucketStorage, SyncDataBatch } from './BucketStorage.js';
+import type { BucketStorage, LocalState, SyncDataBatch } from './BucketStorage.js';
 import type { SyncOperation, SyncOperationsHandler } from './SyncOperationsHandler.js';
 import { constructKey, toStringOrNull } from './bucketHelpers.js';
 import { addChecksums, normalizeChecksum, subtractChecksums } from './checksumUtils.js';
@@ -23,6 +23,8 @@ export type MemoryBucketStorageImplOptions = {
 
 export class MemoryBucketStorageImpl implements BucketStorage {
   protected ps_buckets: PSBucket[];
+  // TODO cleanup
+  protected localBucket: PSBucket;
   protected ps_oplog: PSOplog[];
   protected ps_updated_rows: PsUpdatedRows[];
   protected clientId: string;
@@ -43,6 +45,18 @@ export class MemoryBucketStorageImpl implements BucketStorage {
 
   protected initDefaultState() {
     this.ps_buckets = [];
+    this.localBucket = {
+      id: 0,
+      name: '$local',
+      last_applied_op: 0n,
+      last_op: 0n,
+      target_op: 0n,
+      add_checksum: 0n,
+      op_checksum: 0n,
+      pending_delete: false,
+      count_at_last: 0,
+      count_since_last: 0
+    };
     this.ps_oplog = [];
     this.ps_updated_rows = [];
     this.lastSyncedAt = null;
@@ -76,6 +90,15 @@ export class MemoryBucketStorageImpl implements BucketStorage {
     });
   }
 
+  async getLocalState(): Promise<LocalState> {
+    return await this.lock.runExclusive(async () => {
+      return {
+        lastOpId: this.localBucket.last_op,
+        targetOpId: this.localBucket.target_op
+      };
+    });
+  }
+
   async hasCompletedSync(): Promise<boolean> {
     return await this.lock.runExclusive(async () => {
       return !!this.ps_buckets.find((b) => b.last_applied_op > 0);
@@ -90,18 +113,9 @@ export class MemoryBucketStorageImpl implements BucketStorage {
         if (await this.crudManager?.hasCrud()) {
           return;
         }
-        const localBucket = this.ps_buckets.find((b) => b.name === '$local');
-        if (!localBucket) {
-          throw new Error('Local bucket not found');
-        }
-        localBucket.target_op = normalizeChecksum(writeCheckpoint);
+        this.localBucket.target_op = normalizeChecksum(writeCheckpoint);
       } else {
-        // Set the target to the max op id
-        const localBucket = this.ps_buckets.find((b) => b.name === '$local');
-        if (!localBucket) {
-          throw new Error('Local bucket not found');
-        }
-        localBucket.target_op = normalizeChecksum(MAX_OP_ID);
+        this.localBucket.target_op = normalizeChecksum(MAX_OP_ID);
       }
     });
   }
@@ -124,14 +138,8 @@ export class MemoryBucketStorageImpl implements BucketStorage {
     // SQL: SELECT target_op FROM ps_buckets WHERE name = '$local' AND target_op = CAST(? as INTEGER)
     // TODO: maybe store local state separately
     const shouldProceed = await this.lock.runExclusive(async () => {
-      const localBucket = this.ps_buckets.find((b) => b.name === '$local');
-      if (!localBucket) {
-        // Nothing to update
-        return false;
-      }
-
       const maxOpIdBigint = BigInt(MAX_OP_ID);
-      if (localBucket.target_op !== maxOpIdBigint) {
+      if (this.localBucket.target_op !== maxOpIdBigint) {
         // target_op is not MAX_OP_ID, nothing to update
         return false;
       }
@@ -157,16 +165,10 @@ export class MemoryBucketStorageImpl implements BucketStorage {
     // This is typically called after completing uploads which can
     // be concurrent.
     return await this.lock.runExclusive(async () => {
-      const localBucket = this.ps_buckets.find((b) => b.name === '$local');
-      if (!localBucket) {
-        // Nothing to update
-        return false;
-      }
-
       // Update the '$local' bucket's target_op to the new opId
       // SQL: UPDATE ps_buckets SET target_op = CAST(? as INTEGER) WHERE name='$local'
       const opIdBigint = normalizeChecksum(opId);
-      localBucket.target_op = opIdBigint;
+      this.localBucket.target_op = opIdBigint;
       return true;
     });
   }
@@ -400,10 +402,7 @@ export class MemoryBucketStorageImpl implements BucketStorage {
 
       // Update '$local' bucket if write_checkpoint is provided and it's a complete sync
       if (priority == null && checkpoint.write_checkpoint) {
-        const localBucket = this.ps_buckets.find((b) => b.name === '$local');
-        if (localBucket) {
-          localBucket.last_op = normalizeChecksum(checkpoint.write_checkpoint);
-        }
+        this.localBucket.last_op = normalizeChecksum(checkpoint.write_checkpoint);
       }
     });
 
@@ -452,7 +451,7 @@ export class MemoryBucketStorageImpl implements BucketStorage {
       if (priority == null) {
         const bucketToCount = new Map(checkpoint.buckets.map((b) => [b.bucket, b.count ?? 0]));
         for (const bucket of this.ps_buckets) {
-          if (bucket.name !== '$local' && bucketToCount.has(bucket.name)) {
+          if (bucketToCount.has(bucket.name)) {
             bucket.count_at_last = bucketToCount.get(bucket.name)!;
             bucket.count_since_last = 0;
           }
@@ -479,8 +478,7 @@ export class MemoryBucketStorageImpl implements BucketStorage {
     if (needsCheck) {
       // Check if '$local' bucket has target_op > last_op
       // SQL: SELECT 1 FROM ps_buckets WHERE target_op > last_op AND name = '$local'
-      const localBucket = this.ps_buckets.find((b) => b.name === '$local');
-      if (localBucket && localBucket.target_op > localBucket.last_op) {
+      if (this.localBucket.target_op > this.localBucket.last_op) {
         return false;
       }
 

packages/lite-sdk/src/client/sync/SyncClient.ts

@@ -1,4 +1,4 @@
-import type { BucketRequest, StreamingSyncLine } from '@powersync/service-core';
+import type { BucketRequest, Checkpoint, StreamingSyncLine } from '@powersync/service-core';
 import { BaseListener, BaseObserverInterface, Disposable } from '../../utils/BaseObserver.js';
 import type { BucketStorage } from '../storage/BucketStorage.js';
 import type { SystemDependencies } from '../system/SystemDependencies.js';
@@ -36,11 +36,17 @@ export interface SyncClientListener extends BaseListener {
    * Triggers whenever the status' members have changed in value
    */
   statusChanged?: ((status: SyncStatus) => void) | undefined;
+  /**
+   * Triggers whenever a checkpoint is completed.
+   * @param checkpoint - The checkpoint that was completed
+   */
+  checkpointCompleted?: ((checkpoint: Checkpoint) => void) | undefined;
 }
 
 export type CreateCheckpointResponse = {
   targetUpdated: boolean;
   targetCheckpoint?: string;
+  waitForValidation: (signal?: AbortSignal) => Promise<void>;
 };
 
 /**
@@ -71,8 +77,21 @@ export interface SyncClient extends BaseObserverInterface<SyncClientListener>, D
    * Sets the target write checkpoint.
    * @param customCheckpoint - Optional custom checkpoint to set the target to
    * defaults to creating a managed PowerSync checkpoint
+   *
+   * FIXME
+   * This is a no-op if there are pending CRUD items.
+   * This does not answer the generic question "have I synced to this point in time?"
+   * It answers the question "have I synced to after uploads have been completed?"
    */
   checkpoint: (customCheckpoint?: string) => Promise<CreateCheckpointResponse>;
+
+  /**
+   * Triggers a CRUD upload.
+   * This will perform an upload of the CRUD items if there are any.
+   * If there are no CRUD items, it will do nothing.
+   * If there are CRUD items, the {@link CrudManager} will be used to perform the upload.
+   */
+  triggerCrudUpload: () => void;
 }
 
 export type StreamOptions = {

packages/lite-sdk/src/client/sync/SyncClientImpl.ts

@@ -1,6 +1,7 @@
 import type { Checkpoint, CheckpointBucket } from '@powersync/service-core';
 import { BaseObserver } from '../../utils/BaseObserver.js';
 import type { BucketStorage } from '../storage/BucketStorage.js';
+import { normalizeChecksum } from '../storage/checksumUtils.js';
 import { Connector } from './Connector.js';
 import { CrudManager } from './CrudManager.js';
 import type {
@@ -92,10 +93,15 @@ export class SyncClientImpl extends BaseObserver<SyncClientListener> implements
     const controller = new AbortController();
     this.abortController = controller;
 
-    while (!this.abortController.signal.aborted) {
+    // Don't await this, this will run until the connection is aborted
+    this.connectInternal(connector, controller.signal);
+  }
+
+  protected async connectInternal(connector: Connector, signal: AbortSignal): Promise<void> {
+    while (!signal.aborted) {
       try {
         this.updateSyncStatus({ connecting: true });
-        await this.syncIteration(connector, controller.signal);
+        await this.syncIteration(connector, signal);
       } catch (error) {
         this.updateSyncStatus({
           connected: false,
@@ -110,13 +116,16 @@ export class SyncClientImpl extends BaseObserver<SyncClientListener> implements
       }
     }
   }
-
   disconnect() {
     this.abortController.abort();
   }
 
   async checkpoint(customCheckpoint?: string): Promise<CreateCheckpointResponse> {
     let targetCheckpoint: string | undefined = customCheckpoint;
+    // FIXME, could this be optimized?
+    // If there are no crud items, we set the target to the custom checkpoint or the max op id
+    // We then set the local target to the new checkpoint
+    await this.bucketStorage.handleCrudUploaded(customCheckpoint);
     const targetUpdated = await this.bucketStorage.updateLocalTarget(async () => {
       if (targetCheckpoint) {
         return targetCheckpoint;
@@ -126,11 +135,39 @@ export class SyncClientImpl extends BaseObserver<SyncClientListener> implements
     });
     return {
       targetUpdated,
-      targetCheckpoint
+      targetCheckpoint,
+      waitForValidation: async (signal?: AbortSignal) => {
+        if (!targetCheckpoint || !targetUpdated) {
+          // FIXME throw an error in this case?
+          return;
+        }
+        const localBucketState = await this.bucketStorage.getLocalState();
+        if (localBucketState.lastOpId >= normalizeChecksum(targetCheckpoint)!) {
+          return;
+        }
+
+        return new Promise((resolve, reject) => {
+          let disposeListener: (() => void) | null = null;
+          const signalListener = () => {
+            reject(new Error('Aborted'));
+            disposeListener?.();
+          };
+          disposeListener = this.registerListener({
+            checkpointCompleted: (checkpoint) => {
+              if (checkpoint.write_checkpoint && checkpoint.write_checkpoint >= targetCheckpoint!) {
+                resolve();
+                disposeListener?.();
+                signal?.removeEventListener('abort', signalListener);
+              }
+            }
+          });
+          signal?.addEventListener('abort', disposeListener, { once: true });
+        });
+      }
     };
   }
 
-  protected triggerCrudUpload(): void {
+  triggerCrudUpload(): void {
     if (!this.crudManager) {
       return;
     }
@@ -422,6 +459,10 @@ export class SyncClientImpl extends BaseObserver<SyncClientListener> implements
       lastSyncedAt: new Date()
     });
 
+    this.iterateListeners((listener) => {
+      listener.checkpointCompleted?.(checkpoint);
+    });
+
     return { applied: true, endIteration: false };
   }