Merge remote-tracking branch 'origin/main' into watches

Author: stevensJourney

.changeset/bright-snakes-clean.md

@@ -29,7 +29,7 @@
     "test": "pnpm build && vitest"
   },
   "peerDependencies": {
-    "@powersync/common": "workspace:^1.33.2"
+    "@powersync/common": "workspace:^1.34.0"
   },
   "devDependencies": {
     "@powersync/common": "workspace:*",

.changeset/cuddly-dingos-check.md

@@ -1,5 +1,22 @@
 # @powersync/common
 
+## 1.34.0
+
+### Minor Changes
+
+- ab33799: Add experimental support for raw tables, giving you full control over the table structure to sync into.
+  While PowerSync manages tables as JSON views by default, raw tables have to be created by the application
+  developer.
+
+  For more information about raw tables, see [the documentation](https://docs.powersync.com/usage/use-case-examples/raw-tables).
+
+- 810c6ad: Propagate logger from PowerSyncDatabase to streaming sync and remote implementations, and tweak some log messages.
+
+### Patch Changes
+
+- a1aa18c: Fix sync stream delays during CRUD upload.
+- 9fb898d: [Internal] Removed shared mutex implementation of `readLock` and `writeLock`.
+
 ## 1.33.2
 
 ### Patch Changes

.changeset/mighty-doors-judge.md

@@ -1,6 +1,6 @@
 {
   "name": "@powersync/common",
-  "version": "1.33.2",
+  "version": "1.34.0",
   "publishConfig": {
     "registry": "https://registry.npmjs.org/",
     "access": "public"

.changeset/orange-baboons-work.md

@@ -143,7 +143,6 @@ export const DEFAULT_POWERSYNC_CLOSE_OPTIONS: PowerSyncCloseOptions = {
 
 export const DEFAULT_POWERSYNC_DB_OPTIONS = {
   retryDelayMs: 5000,
-  logger: Logger.get('PowerSyncDatabase'),
   crudUploadThrottleMs: DEFAULT_CRUD_UPLOAD_THROTTLE_MS
 };
 
@@ -192,6 +191,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
   protected runExclusiveMutex: Mutex;
 
+  logger: ILogger;
+
   constructor(options: PowerSyncDatabaseOptionsWithDBAdapter);
   constructor(options: PowerSyncDatabaseOptionsWithOpenFactory);
   constructor(options: PowerSyncDatabaseOptionsWithSettings);
@@ -215,6 +216,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       throw new Error('The provided `database` option is invalid.');
     }
 
+    this.logger = options.logger ?? Logger.get(`PowerSyncDatabase[${this._database.name}]`);
+
     this.bucketStorageAdapter = this.generateBucketStorageAdapter();
     this.closed = false;
     this.currentStatus = new SyncStatus({});
@@ -435,7 +438,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     try {
       schema.validate();
     } catch (ex) {
-      this.options.logger?.warn('Schema validation failed. Unexpected behaviour could occur', ex);
+      this.logger.warn('Schema validation failed. Unexpected behaviour could occur', ex);
     }
     this._schema = schema;
 
@@ -444,10 +447,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this.iterateListeners(async (cb) => cb.schemaChanged?.(schema));
   }
 
-  get logger() {
-    return this.options.logger!;
-  }
-
   /**
    * Wait for initialization to complete.
    * While initializing is automatic, this helps to catch and report initialization errors.
@@ -951,7 +950,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * @param options Options for configuring watch behavior
    */
   watchWithCallback(sql: string, parameters?: any[], handler?: WatchHandler, options?: SQLWatchOptions): void {
-    const { onResult, onError = (e: Error) => this.options.logger?.error(e) } = handler ?? {};
+    const { onResult, onError = (e: Error) => this.logger.error(e) } = handler ?? {};
     if (!onResult) {
       throw new Error('onResult is required');
     }
@@ -1111,7 +1110,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * @returns A dispose function to stop watching for changes
    */
   onChangeWithCallback(handler?: WatchOnChangeHandler, options?: SQLOnChangeOptions): () => void {
-    const { onChange, onError = (e: Error) => this.options.logger?.error(e) } = handler ?? {};
+    const { onChange, onError = (e: Error) => this.logger.error(e) } = handler ?? {};
     if (!onChange) {
       throw new Error('onChange is required');
     }

.changeset/short-countries-swim.md

@@ -92,11 +92,11 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
   async saveSyncData(batch: SyncDataBatch, fixedKeyFormat: boolean = false) {
     await this.writeTransaction(async (tx) => {
       for (const b of batch.buckets) {
-        const result = await tx.execute('INSERT INTO powersync_operations(op, data) VALUES(?, ?)', [
+        await tx.execute('INSERT INTO powersync_operations(op, data) VALUES(?, ?)', [
           'save',
           JSON.stringify({ buckets: [b.toJSON(fixedKeyFormat)] })
         ]);
-        this.logger.debug('saveSyncData', JSON.stringify(result));
+        this.logger.debug(`Saved batch of data for  bucket: ${b.bucket}, operations: ${b.data.length}`);
       }
     });
   }
@@ -115,7 +115,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
       await tx.execute('INSERT INTO powersync_operations(op, data) VALUES(?, ?)', ['delete_bucket', bucket]);
     });
 
-    this.logger.debug('done deleting bucket');
+    this.logger.debug(`Done deleting bucket ${bucket}`);
   }
 
   async hasCompletedSync() {
@@ -139,6 +139,11 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
       }
       return { ready: false, checkpointValid: false, checkpointFailures: r.checkpointFailures };
     }
+    if (priority == null) {
+      this.logger.debug(`Validated checksums checkpoint ${checkpoint.last_op_id}`);
+    } else {
+      this.logger.debug(`Validated checksums for partial checkpoint ${checkpoint.last_op_id}, priority ${priority}`);
+    }
 
     let buckets = checkpoint.buckets;
     if (priority !== undefined) {
@@ -158,7 +163,6 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
 
     const valid = await this.updateObjectsFromBuckets(checkpoint, priority);
     if (!valid) {
-      this.logger.debug('Not at a consistent checkpoint - cannot update local db');
       return { ready: false, checkpointValid: true };
     }
 
@@ -221,7 +225,6 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     ]);
 
     const resultItem = rs.rows?.item(0);
-    this.logger.debug('validateChecksums priority, checkpoint, result item', priority, checkpoint, resultItem);
     if (!resultItem) {
       return {
         checkpointValid: false,
@@ -262,34 +265,32 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
 
     const opId = await cb();
 
-    this.logger.debug(`[updateLocalTarget] Updating target to checkpoint ${opId}`);
-
     return this.writeTransaction(async (tx) => {
       const anyData = await tx.execute('SELECT 1 FROM ps_crud LIMIT 1');
       if (anyData.rows?.length) {
         // if isNotEmpty
-        this.logger.debug('updateLocalTarget', 'ps crud is not empty');
+        this.logger.debug(`New data uploaded since write checkpoint ${opId} - need new write checkpoint`);
         return false;
       }
 
       const rs = await tx.execute("SELECT seq FROM sqlite_sequence WHERE name = 'ps_crud'");
       if (!rs.rows?.length) {
         // assert isNotEmpty
-        throw new Error('SQlite Sequence should not be empty');
+        throw new Error('SQLite Sequence should not be empty');
       }
 
       const seqAfter: number = rs.rows?.item(0)['seq'];
-      this.logger.debug('seqAfter', JSON.stringify(rs.rows?.item(0)));
       if (seqAfter != seqBefore) {
-        this.logger.debug('seqAfter != seqBefore', seqAfter, seqBefore);
+        this.logger.debug(
+          `New data uploaded since write checpoint ${opId} - need new write checkpoint (sequence updated)`
+        );
+
         // New crud data may have been uploaded since we got the checkpoint. Abort.
         return false;
       }
 
-      const response = await tx.execute("UPDATE ps_buckets SET target_op = CAST(? as INTEGER) WHERE name='$local'", [
-        opId
-      ]);
-      this.logger.debug(['[updateLocalTarget] Response from updating target_op ', JSON.stringify(response)]);
+      this.logger.debug(`Updating target write checkpoint to ${opId}`);
+      await tx.execute("UPDATE ps_buckets SET target_op = CAST(? as INTEGER) WHERE name='$local'", [opId]);
       return true;
     });
   }