Update APIs to use hydrated sync rules.

Author: rkistner

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

@@ -1,14 +1,14 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { SqlEventDescriptor, SqliteRow, SqliteValue, SqlSyncRules } from '@powersync/service-sync-rules';
+import { SqlEventDescriptor, SqliteRow, SqliteValue, HydratedSyncRules } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 
 import {
   BaseObserver,
   container,
+  logger as defaultLogger,
   ErrorCode,
   errors,
   Logger,
-  logger as defaultLogger,
   ReplicationAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
@@ -22,13 +22,13 @@ import {
   utils
 } from '@powersync/service-core';
 import * as timers from 'node:timers/promises';
+import { idPrefixFilter } from '../../utils/util.js';
 import { PowerSyncMongo } from './db.js';
 import { CurrentBucket, CurrentDataDocument, SourceKey, SyncRuleDocument } from './models.js';
 import { MongoIdSequence } from './MongoIdSequence.js';
 import { batchCreateCustomWriteCheckpoints } from './MongoWriteCheckpointAPI.js';
 import { cacheKey, OperationBatch, RecordOperation } from './OperationBatch.js';
 import { PersistedBatch } from './PersistedBatch.js';
-import { idPrefixFilter } from '../../utils/util.js';
 
 /**
  * 15MB
@@ -44,7 +44,7 @@ const replicationMutex = new utils.Mutex();
 
 export interface MongoBucketBatchOptions {
   db: PowerSyncMongo;
-  syncRules: SqlSyncRules;
+  syncRules: HydratedSyncRules;
   groupId: number;
   slotName: string;
   lastCheckpointLsn: string | null;
@@ -71,7 +71,7 @@ export class MongoBucketBatch
   private readonly client: mongo.MongoClient;
   public readonly db: PowerSyncMongo;
   public readonly session: mongo.ClientSession;
-  private readonly sync_rules: SqlSyncRules;
+  private readonly sync_rules: HydratedSyncRules;
 
   private readonly group_id: number;
 
@@ -474,8 +474,7 @@ export class MongoBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable,
-          bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(`${this.group_id}`)
+          sourceTable
         });
 
         for (let error of syncErrors) {

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

@@ -1,4 +1,4 @@
-import { SqlSyncRules } from '@powersync/service-sync-rules';
+import { SqlSyncRules, HydratedSyncRules } from '@powersync/service-sync-rules';
 
 import { storage } from '@powersync/service-core';
 
@@ -13,4 +13,8 @@ export class MongoPersistedSyncRules implements storage.PersistedSyncRules {
   ) {
     this.slot_name = slot_name ?? `powersync_${id}`;
   }
+
+  hydratedSyncRules(): HydratedSyncRules {
+    return this.sync_rules.hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(`${this.id}`) });
+  }
 }

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

@@ -25,7 +25,7 @@ import {
   WatchWriteCheckpointOptions
 } from '@powersync/service-core';
 import { JSONBig } from '@powersync/service-jsonbig';
-import { ParameterLookup, SqliteJsonRow, SqlSyncRules } from '@powersync/service-sync-rules';
+import { ParameterLookup, SqliteJsonRow, SqlSyncRules, HydratedSyncRules } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 import { LRUCache } from 'lru-cache';
 import * as timers from 'timers/promises';
@@ -61,7 +61,7 @@ export class MongoSyncBucketStorage
   private readonly db: PowerSyncMongo;
   readonly checksums: MongoChecksums;
 
-  private parsedSyncRulesCache: { parsed: SqlSyncRules; options: storage.ParseSyncRulesOptions } | undefined;
+  private parsedSyncRulesCache: { parsed: HydratedSyncRules; options: storage.ParseSyncRulesOptions } | undefined;
   private writeCheckpointAPI: MongoWriteCheckpointAPI;
 
   constructor(
@@ -101,14 +101,14 @@ export class MongoSyncBucketStorage
     });
   }
 
-  getParsedSyncRules(options: storage.ParseSyncRulesOptions): SqlSyncRules {
+  getParsedSyncRules(options: storage.ParseSyncRulesOptions): HydratedSyncRules {
     const { parsed, options: cachedOptions } = this.parsedSyncRulesCache ?? {};
     /**
      * Check if the cached sync rules, if present, had the same options.
      * Parse sync rules if the options are different or if there is no cached value.
      */
     if (!parsed || options.defaultSchema != cachedOptions?.defaultSchema) {
-      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).sync_rules, options };
+      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).hydratedSyncRules(), options };
     }
 
     return this.parsedSyncRulesCache!.parsed;
@@ -170,7 +170,7 @@ export class MongoSyncBucketStorage
     await using batch = new MongoBucketBatch({
       logger: options.logger,
       db: this.db,
-      syncRules: this.sync_rules.parsed(options).sync_rules,
+      syncRules: this.sync_rules.parsed(options).hydratedSyncRules(),
       groupId: this.group_id,
       slotName: this.slot_name,
       lastCheckpointLsn: checkpoint_lsn,

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -10,15 +10,20 @@ import {
   ServiceError
 } from '@powersync/lib-services-framework';
 import {
-  InternalOpId,
   MetricsEngine,
   RelationCache,
   SaveOperationTag,
   SourceEntityDescriptor,
   SourceTable,
   storage
 } from '@powersync/service-core';
-import { DatabaseInputRow, SqliteInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
+import {
+  DatabaseInputRow,
+  SqliteInputRow,
+  SqliteRow,
+  HydratedSyncRules,
+  TablePattern
+} from '@powersync/service-sync-rules';
 import { ReplicationMetric } from '@powersync/service-types';
 import { MongoLSN } from '../common/MongoLSN.js';
 import { PostImagesOption } from '../types/types.js';
@@ -75,7 +80,7 @@ export class ChangeStreamInvalidatedError extends DatabaseConnectionError {
 }
 
 export class ChangeStream {
-  sync_rules: SqlSyncRules;
+  sync_rules: HydratedSyncRules;
   group_id: number;
 
   connection_id = 1;

modules/module-mssql/src/replication/CDCStream.ts

@@ -10,7 +10,13 @@ import {
 } from '@powersync/lib-services-framework';
 import { getUuidReplicaIdentityBson, MetricsEngine, SourceEntityDescriptor, storage } from '@powersync/service-core';
 
-import { SqliteInputRow, SqliteRow, SqlSyncRules, TablePattern } from '@powersync/service-sync-rules';
+import {
+  SqliteInputRow,
+  SqliteRow,
+  SqlSyncRules,
+  HydratedSyncRules,
+  TablePattern
+} from '@powersync/service-sync-rules';
 
 import { ReplicationMetric } from '@powersync/service-types';
 import { BatchedSnapshotQuery, MSSQLSnapshotQuery, SimpleSnapshotQuery } from './MSSQLSnapshotQuery.js';
@@ -82,7 +88,7 @@ export class CDCDataExpiredError extends DatabaseConnectionError {
 }
 
 export class CDCStream {
-  private readonly syncRules: SqlSyncRules;
+  private readonly syncRules: HydratedSyncRules;
   private readonly storage: storage.SyncRulesBucketStorage;
   private readonly connections: MSSQLConnectionManager;
   private readonly abortSignal: AbortSignal;

modules/module-mysql/src/replication/BinLogStream.ts

@@ -61,7 +61,7 @@ function createTableId(schema: string, tableName: string): string {
 }
 
 export class BinLogStream {
-  private readonly syncRules: sync_rules.SqlSyncRules;
+  private readonly syncRules: sync_rules.HydratedSyncRules;
   private readonly groupId: number;
 
   private readonly storage: storage.SyncRulesBucketStorage;

modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts

@@ -60,7 +60,9 @@ export class PostgresSyncRulesStorage
   protected writeCheckpointAPI: PostgresWriteCheckpointAPI;
 
   //   TODO we might be able to share this in an abstract class
-  private parsedSyncRulesCache: { parsed: sync_rules.SqlSyncRules; options: storage.ParseSyncRulesOptions } | undefined;
+  private parsedSyncRulesCache:
+    | { parsed: sync_rules.HydratedSyncRules; options: storage.ParseSyncRulesOptions }
+    | undefined;
   private _checksumCache: storage.ChecksumCache | undefined;
 
   constructor(protected options: PostgresSyncRulesStorageOptions) {
@@ -96,14 +98,14 @@ export class PostgresSyncRulesStorage
   }
 
   //   TODO we might be able to share this in an abstract class
-  getParsedSyncRules(options: storage.ParseSyncRulesOptions): sync_rules.SqlSyncRules {
+  getParsedSyncRules(options: storage.ParseSyncRulesOptions): sync_rules.HydratedSyncRules {
     const { parsed, options: cachedOptions } = this.parsedSyncRulesCache ?? {};
     /**
      * Check if the cached sync rules, if present, had the same options.
      * Parse sync rules if the options are different or if there is no cached value.
      */
     if (!parsed || options.defaultSchema != cachedOptions?.defaultSchema) {
-      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).sync_rules, options };
+      this.parsedSyncRulesCache = { parsed: this.sync_rules.parsed(options).hydratedSyncRules(), options };
     }
 
     return this.parsedSyncRulesCache!.parsed;
@@ -349,7 +351,7 @@ export class PostgresSyncRulesStorage
     const batch = new PostgresBucketBatch({
       logger: options.logger ?? framework.logger,
       db: this.db,
-      sync_rules: this.sync_rules.parsed(options).sync_rules,
+      sync_rules: this.sync_rules.parsed(options).hydratedSyncRules(),
       group_id: this.group_id,
       slot_name: this.slot_name,
       last_checkpoint_lsn: checkpoint_lsn,

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

@@ -24,7 +24,7 @@ import { PostgresPersistedBatch } from './PostgresPersistedBatch.js';
 export interface PostgresBucketBatchOptions {
   logger: Logger;
   db: lib_postgres.DatabaseClient;
-  sync_rules: sync_rules.SqlSyncRules;
+  sync_rules: sync_rules.HydratedSyncRules;
   group_id: number;
   slot_name: string;
   last_checkpoint_lsn: string | null;
@@ -72,7 +72,7 @@ export class PostgresBucketBatch
   protected persisted_op: InternalOpId | null;
 
   protected write_checkpoint_batch: storage.CustomWriteCheckpointOptions[];
-  protected readonly sync_rules: sync_rules.SqlSyncRules;
+  protected readonly sync_rules: sync_rules.HydratedSyncRules;
   protected batch: OperationBatch | null;
   private lastWaitingLogThrottled = 0;
   private markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
@@ -840,8 +840,7 @@ export class PostgresBucketBatch
       if (sourceTable.syncData) {
         const { results: evaluated, errors: syncErrors } = this.sync_rules.evaluateRowWithErrors({
           record: after,
-          sourceTable,
-          bucketIdTransformer: sync_rules.SqlSyncRules.versionedBucketIdTransformer(`${this.group_id}`)
+          sourceTable
         });
 
         for (const error of syncErrors) {

modules/module-postgres-storage/src/storage/sync-rules/PostgresPersistedSyncRulesContent.ts

@@ -35,7 +35,12 @@ export class PostgresPersistedSyncRulesContent implements storage.PersistedSyncR
     return {
       id: this.id,
       slot_name: this.slot_name,
-      sync_rules: SqlSyncRules.fromYaml(this.sync_rules_content, options)
+      sync_rules: SqlSyncRules.fromYaml(this.sync_rules_content, options),
+      hydratedSyncRules() {
+        return this.sync_rules.hydrate({
+          bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(`${this.id}`)
+        });
+      }
     };
   }
 

modules/module-postgres/src/replication/WalStream.ts

@@ -27,6 +27,7 @@ import {
   SqliteInputValue,
   SqliteRow,
   SqlSyncRules,
+  HydratedSyncRules,
   TablePattern,
   ToastableSqliteRow,
   toSyncRulesRow
@@ -107,7 +108,7 @@ export class MissingReplicationSlotError extends Error {
 }
 
 export class WalStream {
-  sync_rules: SqlSyncRules;
+  sync_rules: HydratedSyncRules;
   group_id: number;
 
   connection_id = 1;