Merge remote-tracking branch 'origin/main' into improve-logging

Author: rkistner

.changeset/bright-snakes-clean.md

@@ -0,0 +1,12 @@
+---
+'@powersync/common': minor
+'@powersync/node': minor
+'@powersync/web': minor
+'@powersync/react-native': minor
+---
+
+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).

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -27,6 +27,7 @@ import { CrudTransaction } from './sync/bucket/CrudTransaction.js';
 import {
   DEFAULT_CRUD_UPLOAD_THROTTLE_MS,
   DEFAULT_RETRY_DELAY_MS,
+  InternalConnectionOptions,
   StreamingSyncImplementation,
   StreamingSyncImplementationListener,
   type AdditionalConnectionOptions,
@@ -461,7 +462,10 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Connects to stream of events from the PowerSync instance.
    */
   async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) {
-    return this.connectionManager.connect(connector, options);
+    const resolvedOptions: InternalConnectionOptions = options ?? {};
+    resolvedOptions.serializedSchema = this.schema.toJSON();
+
+    return this.connectionManager.connect(connector, resolvedOptions);
   }
 
   /**

packages/common/src/client/ConnectionManager.ts

@@ -2,7 +2,7 @@ import { ILogger } from 'js-logger';
 import { BaseListener, BaseObserver } from '../utils/BaseObserver.js';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js';
 import {
-  PowerSyncConnectionOptions,
+  InternalConnectionOptions,
   StreamingSyncImplementation
 } from './sync/stream/AbstractStreamingSyncImplementation.js';
 
@@ -24,14 +24,14 @@ export interface ConnectionManagerSyncImplementationResult {
 export interface ConnectionManagerOptions {
   createSyncImplementation(
     connector: PowerSyncBackendConnector,
-    options: PowerSyncConnectionOptions
+    options: InternalConnectionOptions
   ): Promise<ConnectionManagerSyncImplementationResult>;
   logger: ILogger;
 }
 
 type StoredConnectionOptions = {
   connector: PowerSyncBackendConnector;
-  options: PowerSyncConnectionOptions;
+  options: InternalConnectionOptions;
 };
 
 /**
@@ -95,7 +95,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
     await this.syncDisposer?.();
   }
 
-  async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) {
+  async connect(connector: PowerSyncBackendConnector, options: InternalConnectionOptions) {
     // Keep track if there were pending operations before this call
     const hadPendingOptions = !!this.pendingConnectionOptions;
 
@@ -140,7 +140,7 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
   }
 
   protected async connectInternal() {
-    let appliedOptions: PowerSyncConnectionOptions | null = null;
+    let appliedOptions: InternalConnectionOptions | null = null;
 
     // This method ensures a disconnect before any connection attempt
     await this.disconnectInternal();

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

@@ -123,7 +123,9 @@ export interface StreamingSyncImplementationListener extends BaseListener {
  * Configurable options to be used when connecting to the PowerSync
  * backend instance.
  */
-export interface PowerSyncConnectionOptions extends BaseConnectionOptions, AdditionalConnectionOptions {}
+export type PowerSyncConnectionOptions = Omit<InternalConnectionOptions, 'serializedSchema'>;
+
+export interface InternalConnectionOptions extends BaseConnectionOptions, AdditionalConnectionOptions {}
 
 /** @internal */
 export interface BaseConnectionOptions {
@@ -152,6 +154,11 @@ export interface BaseConnectionOptions {
    * These parameters are passed to the sync rules, and will be available under the`user_parameters` object.
    */
   params?: Record<string, StreamingSyncRequestParameterType>;
+
+  /**
+   * The serialized schema - mainly used to forward information about raw tables to the sync client.
+   */
+  serializedSchema?: any;
 }
 
 /** @internal */
@@ -176,7 +183,7 @@ export interface StreamingSyncImplementation extends BaseObserver<StreamingSyncI
   /**
    * Connects to the sync service
    */
-  connect(options?: PowerSyncConnectionOptions): Promise<void>;
+  connect(options?: InternalConnectionOptions): Promise<void>;
   /**
    * Disconnects from the sync services.
    * @throws if not connected or if abort is not controlled internally
@@ -208,7 +215,8 @@ export const DEFAULT_STREAM_CONNECTION_OPTIONS: RequiredPowerSyncConnectionOptio
   connectionMethod: SyncStreamConnectionMethod.WEB_SOCKET,
   clientImplementation: DEFAULT_SYNC_CLIENT_IMPLEMENTATION,
   fetchStrategy: FetchStrategy.Buffered,
-  params: {}
+  params: {},
+  serializedSchema: undefined
 };
 
 // The priority we assume when we receive checkpoint lines where no priority is set.
@@ -621,6 +629,10 @@ The next upload iteration will be delayed.`);
   }
 
   private async legacyStreamingSyncIteration(signal: AbortSignal, resolvedOptions: RequiredPowerSyncConnectionOptions) {
+    if (resolvedOptions.serializedSchema?.raw_tables != null) {
+      this.logger.warn('Raw tables require the Rust-based sync client. The JS client will ignore them.');
+    }
+
     this.logger.debug('Streaming sync iteration started');
     this.options.adapter.startSession();
     let [req, bucketMap] = await this.collectLocalBucketState();
@@ -714,6 +726,8 @@ The next upload iteration will be delayed.`);
 
       if (isStreamingSyncCheckpoint(line)) {
         targetCheckpoint = line.checkpoint;
+        // New checkpoint - existing validated checkpoint is no longer valid
+        pendingValidatedCheckpoint = null;
         const bucketsToDelete = new Set<string>(bucketMap.keys());
         const newBuckets = new Map<string, BucketDescription>();
         for (const checksum of line.checkpoint.buckets) {
@@ -735,7 +749,13 @@ The next upload iteration will be delayed.`);
         if (result.endIteration) {
           return;
         } else if (!result.applied) {
+          // "Could not apply checkpoint due to local data". We need to retry after
+          // finishing uploads.
           pendingValidatedCheckpoint = targetCheckpoint;
+        } else {
+          // Nothing to retry later. This would likely already be null from the last
+          // checksum or checksum_diff operation, but we make sure.
+          pendingValidatedCheckpoint = null;
         }
       } else if (isStreamingSyncCheckpointPartiallyComplete(line)) {
         const priority = line.partial_checkpoint_complete.priority;
@@ -771,6 +791,8 @@ The next upload iteration will be delayed.`);
         if (targetCheckpoint == null) {
           throw new Error('Checkpoint diff without previous checkpoint');
         }
+        // New checkpoint - existing validated checkpoint is no longer valid
+        pendingValidatedCheckpoint = null;
         const diff = line.checkpoint_diff;
         const newBuckets = new Map<string, BucketChecksum>();
         for (const checksum of targetCheckpoint.buckets) {
@@ -1025,12 +1047,12 @@ The next upload iteration will be delayed.`);
     }
 
     try {
-      await control(
-        PowerSyncControlCommand.START,
-        JSON.stringify({
-          parameters: resolvedOptions.params
-        })
-      );
+      const options: any = { parameters: resolvedOptions.params };
+      if (resolvedOptions.serializedSchema) {
+        options.schema = resolvedOptions.serializedSchema;
+      }
+
+      await control(PowerSyncControlCommand.START, JSON.stringify(options));
 
       this.notifyCompletedUploads = () => {
         controlInvocations?.enqueueData({ command: PowerSyncControlCommand.NOTIFY_CRUD_UPLOAD_COMPLETED });

packages/common/src/db/schema/RawTable.ts

@@ -0,0 +1,67 @@
+/**
+ * A pending variant of a {@link RawTable} that doesn't have a name (because it would be inferred when creating the
+ * schema).
+ */
+export type RawTableType = {
+  /**
+   * The statement to run when PowerSync detects that a row needs to be inserted or updated.
+   */
+  put: PendingStatement;
+  /**
+   * The statement to run when PowerSync detects that a row needs to be deleted.
+   */
+  delete: PendingStatement;
+};
+
+/**
+ * A parameter to use as part of {@link PendingStatement}.
+ *
+ * For delete statements, only the `"Id"` value is supported - the sync client will replace it with the id of the row to
+ * be synced.
+ *
+ * For insert and replace operations, the values of columns in the table are available as parameters through
+ * `{Column: 'name'}`.
+ */
+export type PendingStatementParameter = 'Id' | { Column: string };
+
+/**
+ * A statement that the PowerSync client should use to insert or delete data into a table managed by the user.
+ */
+export type PendingStatement = {
+  sql: string;
+  params: PendingStatementParameter[];
+};
+
+/**
+ * Instructs PowerSync to sync data into a "raw" table.
+ *
+ * Since raw tables are not backed by JSON, running complex queries on them may be more efficient. Further, they allow
+ * using client-side table and column constraints.
+ *
+ * To collect local writes to raw tables with PowerSync, custom triggers are required. See
+ * {@link https://docs.powersync.com/usage/use-case-examples/raw-tables the documentation} for details and an example on
+ * using raw tables.
+ *
+ * Note that raw tables are only supported when using the new `SyncClientImplementation.rust` sync client.
+ *
+ * @experimental Please note that this feature is experimental at the moment, and not covered by PowerSync semver or
+ * stability guarantees.
+ */
+export class RawTable implements RawTableType {
+  /**
+   * The name of the table.
+   *
+   * This does not have to match the actual table name in the schema - {@link put} and {@link delete} are free to use
+   * another table. Instead, this name is used by the sync client to recognize that operations on this table (as it
+   * appears in the source / backend database) are to be handled specially.
+   */
+  name: string;
+  put: PendingStatement;
+  delete: PendingStatement;
+
+  constructor(name: string, type: RawTableType) {
+    this.name = name;
+    this.put = type.put;
+    this.delete = type.delete;
+  }
+}

packages/common/src/db/schema/Schema.ts

@@ -1,3 +1,4 @@
+import { RawTable, RawTableType } from './RawTable.js';
 import { RowType, Table } from './Table.js';
 
 type SchemaType = Record<string, Table<any>>;
@@ -16,6 +17,7 @@ export class Schema<S extends SchemaType = SchemaType> {
   readonly types: SchemaTableType<S>;
   readonly props: S;
   readonly tables: Table[];
+  readonly rawTables: RawTable[];
 
   constructor(tables: Table[] | S) {
     if (Array.isArray(tables)) {
@@ -36,6 +38,24 @@ export class Schema<S extends SchemaType = SchemaType> {
       this.props = tables as S;
       this.tables = this.convertToClassicTables(this.props);
     }
+
+    this.rawTables = [];
+  }
+
+  /**
+   * Adds raw tables to this schema. Raw tables are identified by their name, but entirely managed by the application
+   * developer instead of automatically by PowerSync.
+   * Since raw tables are not backed by JSON, running complex queries on them may be more efficient. Further, they allow
+   * using client-side table and column constraints.
+   * Note that raw tables are only supported when using the new `SyncClientImplementation.rust` sync client.
+   *
+   * @param tables An object of (table name, raw table definition) entries.
+   * @experimental Note that the raw tables API is still experimental and may change in the future.
+   */
+  withRawTables(tables: Record<string, RawTableType>) {
+    for (const [name, rawTableDefinition] of Object.entries(tables)) {
+      this.rawTables.push(new RawTable(name, rawTableDefinition));
+    }
   }
 
   validate() {
@@ -47,7 +67,8 @@ export class Schema<S extends SchemaType = SchemaType> {
   toJSON() {
     return {
       // This is required because "name" field is not present in TableV2
-      tables: this.tables.map((t) => t.toJSON())
+      tables: this.tables.map((t) => t.toJSON()),
+      raw_tables: this.rawTables
     };
   }
 

packages/common/tests/db/schema/Schema.test.ts

@@ -80,6 +80,18 @@ describe('Schema', () => {
         content: column.text
       })
     });
+    schema.withRawTables({
+      lists: {
+        put: {
+          sql: 'SELECT 1',
+          params: [{ Column: 'foo' }]
+        },
+        delete: {
+          sql: 'SELECT 2',
+          params: ['Id']
+        }
+      }
+    });
 
     const json = schema.toJSON();
 
@@ -115,6 +127,19 @@ describe('Schema', () => {
           ],
           indexes: []
         }
+      ],
+      raw_tables: [
+        {
+          name: 'lists',
+          delete: {
+            sql: 'SELECT 2',
+            params: ['Id']
+          },
+          put: {
+            sql: 'SELECT 1',
+            params: [{ Column: 'foo' }]
+          }
+        }
       ]
     });
   });

packages/node/src/db/PowerSyncDatabase.ts

@@ -78,7 +78,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
     connector: PowerSyncBackendConnector,
     options: NodeAdditionalConnectionOptions
   ): AbstractStreamingSyncImplementation {
-    const remote = new NodeRemote(connector, this.options.logger, {
+    const logger = this.options.logger;
+    const remote = new NodeRemote(connector, logger, {
       dispatcher: options.dispatcher,
       ...(this.options as NodePowerSyncDatabaseOptions).remoteOptions
     });
@@ -93,7 +94,7 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase {
       retryDelayMs: this.options.retryDelayMs,
       crudUploadThrottleMs: this.options.crudUploadThrottleMs,
       identifier: this.database.name,
-      logger: this.logger
+      logger
     });
   }
 }