Merge remote-tracking branch 'origin/main' into compact-checksums

Author: rkistner

modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts

@@ -1,5 +1,5 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { SqlEventDescriptor, SqliteRow, SqlSyncRules } from '@powersync/service-sync-rules';
+import { SqlEventDescriptor, SqliteRow, SqliteValue, SqlSyncRules } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 
 import {
@@ -319,7 +319,8 @@ export class MongoBucketBatch
     const record = operation.record;
     const beforeId = operation.beforeId;
     const afterId = operation.afterId;
-    let after = record.after;
+    let sourceAfter = record.after;
+    let after = sourceAfter && this.sync_rules.applyRowContext(sourceAfter);
     const sourceTable = record.sourceTable;
 
     let existing_buckets: CurrentBucket[] = [];
@@ -367,7 +368,7 @@ export class MongoBucketBatch
         existing_lookups = result.lookups;
         if (this.storeCurrentData) {
           const data = deserializeBson((result.data as mongo.Binary).buffer) as SqliteRow;
-          after = storage.mergeToast(after!, data);
+          after = storage.mergeToast<SqliteValue>(after!, data);
         }
       }
     } else if (record.tag == SaveOperationTag.DELETE) {

modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts

@@ -687,7 +687,8 @@ export class PostgresBucketBatch
     // We store bytea colums for source keys
     const beforeId = operation.beforeId;
     const afterId = operation.afterId;
-    let after = record.after;
+    let sourceAfter = record.after;
+    let after = sourceAfter && this.sync_rules.applyRowContext(sourceAfter);
     const sourceTable = record.sourceTable;
 
     let existingBuckets: CurrentBucket[] = [];

packages/service-core-tests/src/tests/register-data-storage-tests.ts

@@ -3,9 +3,10 @@ import {
   getUuidReplicaIdentityBson,
   InternalOpId,
   OplogEntry,
+  SaveOptions,
   storage
 } from '@powersync/service-core';
-import { ParameterLookup, RequestParameters } from '@powersync/service-sync-rules';
+import { DateTimeValue, ParameterLookup, RequestParameters } from '@powersync/service-sync-rules';
 import { expect, test, describe, beforeEach } from 'vitest';
 import * as test_utils from '../test-utils/test-utils-index.js';
 import { SqlBucketDescriptor } from '@powersync/service-sync-rules/src/SqlBucketDescriptor.js';
@@ -1995,4 +1996,70 @@ bucket_definitions:
     // we expect 0n and 1n, or 1n and 2n.
     expect(checkpoint2).toBeGreaterThan(checkpoint1);
   });
+
+  test('data with custom types', async () => {
+    await using factory = await generateStorageFactory();
+    const testValue = {
+      sourceTable: TEST_TABLE,
+      tag: storage.SaveOperationTag.INSERT,
+      after: {
+        id: 't1',
+        description: new DateTimeValue('2025-08-28T11:30:00')
+      },
+      afterReplicaId: test_utils.rid('t1')
+    } satisfies SaveOptions;
+
+    {
+      // First, deploy old sync rules and row with date time value
+      const syncRules = await factory.updateSyncRules({
+        content: `
+  bucket_definitions:
+    global:
+      data:
+        - SELECT id, description FROM test
+  `
+      });
+      const bucketStorage = factory.getInstance(syncRules);
+      await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
+        await batch.save(testValue);
+        await batch.commit('1/1');
+      });
+
+      const { checkpoint } = await bucketStorage.getCheckpoint();
+      const batch = await test_utils.fromAsync(
+        bucketStorage.getBucketDataBatch(checkpoint, new Map([['global[]', 0n]]))
+      );
+      expect(batch[0].chunkData.data).toMatchObject([
+        {
+          data: '{"id":"t1","description":"2025-08-28 11:30:00"}'
+        }
+      ]);
+    }
+
+    const syncRules = await factory.updateSyncRules({
+      content: `
+  bucket_definitions:
+    global:
+      data:
+        - SELECT id, description FROM test
+  
+  config:
+    edition: 2
+  `
+    });
+    const bucketStorage = factory.getInstance(syncRules);
+    await bucketStorage.startBatch(test_utils.BATCH_OPTIONS, async (batch) => {
+      await batch.save(testValue);
+      await batch.commit('1/2');
+    });
+    const { checkpoint } = await bucketStorage.getCheckpoint();
+    const batch = await test_utils.fromAsync(
+      bucketStorage.getBucketDataBatch(checkpoint, new Map([['2#global[]', 0n]]))
+    );
+    expect(batch[0].chunkData.data).toMatchObject([
+      {
+        data: '{"id":"t1","description":"2025-08-28T11:30:00"}'
+      }
+    ]);
+  });
 }

packages/service-core/src/storage/BucketStorage.ts

@@ -39,8 +39,8 @@ export enum SyncRuleState {
 export const DEFAULT_DOCUMENT_BATCH_LIMIT = 1000;
 export const DEFAULT_DOCUMENT_CHUNK_LIMIT_BYTES = 1 * 1024 * 1024;
 
-export function mergeToast(record: ToastableSqliteRow, persisted: ToastableSqliteRow): ToastableSqliteRow {
-  const newRecord: ToastableSqliteRow = {};
+export function mergeToast<V>(record: ToastableSqliteRow<V>, persisted: ToastableSqliteRow<V>): ToastableSqliteRow<V> {
+  const newRecord: ToastableSqliteRow<V> = {};
   for (let key in record) {
     if (typeof record[key] == 'undefined') {
       newRecord[key] = persisted[key];

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -37,10 +37,7 @@ export class SqlBucketDescriptor implements BucketSource {
   name: string;
   bucketParameters?: string[];
 
-  constructor(
-    name: string,
-    private readonly compatibility: CompatibilityContext
-  ) {
+  constructor(name: string) {
     this.name = name;
   }
 
@@ -61,11 +58,11 @@ export class SqlBucketDescriptor implements BucketSource {
 
   parameterIdSequence = new IdSequence();
 
-  addDataQuery(sql: string, options: SyncRulesOptions): QueryParseResult {
+  addDataQuery(sql: string, options: SyncRulesOptions, compatibility: CompatibilityContext): QueryParseResult {
     if (this.bucketParameters == null) {
       throw new Error('Bucket parameters must be defined');
     }
-    const dataRows = SqlDataQuery.fromSql(this.name, this.bucketParameters, sql, options, this.compatibility);
+    const dataRows = SqlDataQuery.fromSql(this.name, this.bucketParameters, sql, options, compatibility);
 
     this.dataQueries.push(dataRows);
 
@@ -105,13 +102,7 @@ export class SqlBucketDescriptor implements BucketSource {
         continue;
       }
 
-      results.push(
-        ...query.evaluateRow(
-          options.sourceTable,
-          applyRowContext(options.record, this.compatibility),
-          options.bucketIdTransformer
-        )
-      );
+      results.push(...query.evaluateRow(options.sourceTable, options.record, options.bucketIdTransformer));
     }
     return results;
   }

packages/sync-rules/src/SqlSyncRules.ts

@@ -22,13 +22,17 @@ import {
   RequestParameters,
   SourceSchema,
   SqliteInputRow,
+  SqliteInputValue,
   SqliteJsonRow,
+  SqliteRow,
+  SqliteValue,
   StreamParseOptions,
   SyncRules
 } from './types.js';
 import { BucketSource } from './BucketSource.js';
 import { syncStreamFromSql } from './streams/from_sql.js';
 import { CompatibilityContext, CompatibilityEdition, CompatibilityOption } from './compatibility.js';
+import { applyRowContext } from './utils.js';
 
 const ACCEPT_POTENTIALLY_DANGEROUS_QUERIES = Symbol('ACCEPT_POTENTIALLY_DANGEROUS_QUERIES');
 
@@ -217,7 +221,7 @@ export class SqlSyncRules implements SyncRules {
       const parameters = value.get('parameters', true) as unknown;
       const dataQueries = value.get('data', true) as unknown;
 
-      const descriptor = new SqlBucketDescriptor(key, compatibility);
+      const descriptor = new SqlBucketDescriptor(key);
 
       if (parameters instanceof Scalar) {
         rules.withScalar(parameters, (q) => {
@@ -239,7 +243,7 @@ export class SqlSyncRules implements SyncRules {
       }
       for (let query of dataQueries.items) {
         rules.withScalar(query, (q) => {
-          return descriptor.addDataQuery(q, queryOptions);
+          return descriptor.addDataQuery(q, queryOptions, compatibility);
         });
       }
       rules.bucketSources.push(descriptor);
@@ -382,6 +386,12 @@ export class SqlSyncRules implements SyncRules {
     this.content = content;
   }
 
+  applyRowContext<MaybeToast extends undefined = never>(
+    source: SqliteRow<SqliteInputValue | MaybeToast>
+  ): SqliteRow<SqliteValue | MaybeToast> {
+    return applyRowContext(source, this.compatibility);
+  }
+
   /**
    * Throws errors.
    */

packages/sync-rules/src/events/SqlEventDescriptor.ts

@@ -51,10 +51,7 @@ export class SqlEventDescriptor {
       };
     }
 
-    return matchingQuery.evaluateRowWithErrors(
-      options.sourceTable,
-      applyRowContext(options.record, this.compatibility)
-    );
+    return matchingQuery.evaluateRowWithErrors(options.sourceTable, options.record);
   }
 
   getSourceTables(): Set<TablePattern> {

packages/sync-rules/src/streams/from_sql.ts

@@ -100,8 +100,7 @@ class SyncStreamCompiler {
 
     const stream = new SyncStream(
       this.descriptorName,
-      new BaseSqlDataQuery(this.compileDataQuery(tools, query, alias, sourceTable)),
-      this.options.compatibility
+      new BaseSqlDataQuery(this.compileDataQuery(tools, query, alias, sourceTable))
     );
     stream.subscribedToByDefault = this.options.auto_subscribe ?? false;
     if (filter.isValid(tools)) {

packages/sync-rules/src/streams/stream.ts

@@ -27,11 +27,7 @@ export class SyncStream implements BucketSource {
   variants: StreamVariant[];
   data: BaseSqlDataQuery;
 
-  constructor(
-    name: string,
-    data: BaseSqlDataQuery,
-    private readonly compatibility: CompatibilityContext
-  ) {
+  constructor(name: string, data: BaseSqlDataQuery) {
     this.name = name;
     this.subscribedToByDefault = false;
     this.priority = DEFAULT_BUCKET_PRIORITY;
@@ -172,15 +168,14 @@ export class SyncStream implements BucketSource {
     }
 
     const stream = this;
-    const mappedRow = applyRowContext(options.record, this.compatibility);
     const row: TableRow = {
       sourceTable: options.sourceTable,
-      record: mappedRow
+      record: options.record
     };
 
     return this.data.evaluateRowWithOptions({
       table: options.sourceTable,
-      row: applyRowContext(options.record, this.compatibility),
+      row: options.record,
       bucketIds() {
         const bucketIds: string[] = [];
         for (const variant of stream.variants) {

packages/sync-rules/src/types.ts

@@ -218,7 +218,7 @@ export type SqliteInputRow = SqliteRow<SqliteInputValue>;
  *
  * Toasted values are `undefined`.
  */
-export type ToastableSqliteRow = SqliteRow<SqliteInputValue | undefined>;
+export type ToastableSqliteRow<V = SqliteInputValue> = SqliteRow<V | undefined>;
 
 /**
  * A value as received from the database.
@@ -297,7 +297,7 @@ export interface InputParameter {
  */
 export type BucketIdTransformer = (regularId: string) => string;
 
-export interface EvaluateRowOptions extends TableRow<SqliteInputRow> {
+export interface EvaluateRowOptions extends TableRow {
   bucketIdTransformer: BucketIdTransformer;
 }