Remove BucketIdTransformer.

Author: rkistner

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

@@ -1,6 +1,7 @@
 import { SqlSyncRules, HydratedSyncRules } from '@powersync/service-sync-rules';
 
 import { storage } from '@powersync/service-core';
+import { versionedHydrationState } from '@powersync/service-sync-rules/src/HydrationState.js';
 
 export class MongoPersistedSyncRules implements storage.PersistedSyncRules {
   public readonly slot_name: string;
@@ -15,6 +16,6 @@ export class MongoPersistedSyncRules implements storage.PersistedSyncRules {
   }
 
   hydratedSyncRules(): HydratedSyncRules {
-    return this.sync_rules.hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(`${this.id}`) });
+    return this.sync_rules.hydrate({ hydrationState: versionedHydrationState(this.id) });
   }
 }

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

@@ -4,6 +4,7 @@ import { storage } from '@powersync/service-core';
 import { SqlSyncRules } from '@powersync/service-sync-rules';
 
 import { models } from '../../types/types.js';
+import { versionedHydrationState } from '@powersync/service-sync-rules/src/HydrationState.js';
 
 export class PostgresPersistedSyncRulesContent implements storage.PersistedSyncRulesContent {
   public readonly slot_name: string;
@@ -38,7 +39,7 @@ export class PostgresPersistedSyncRulesContent implements storage.PersistedSyncR
       sync_rules: SqlSyncRules.fromYaml(this.sync_rules_content, options),
       hydratedSyncRules() {
         return this.sync_rules.hydrate({
-          bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer(`${this.id}`)
+          hydrationState: versionedHydrationState(this.id)
         });
       }
     };

packages/service-core-tests/src/test-utils/general-utils.ts

@@ -1,5 +1,6 @@
 import { storage, utils } from '@powersync/service-core';
 import { GetQuerierOptions, RequestParameters, SqlSyncRules } from '@powersync/service-sync-rules';
+import { versionedHydrationState } from '@powersync/service-sync-rules/src/HydrationState.js';
 import * as bson from 'bson';
 
 export const ZERO_LSN = '0/0';
@@ -27,7 +28,7 @@ export function testRules(content: string): storage.PersistedSyncRulesContent {
         sync_rules: SqlSyncRules.fromYaml(content, options),
         slot_name: 'test',
         hydratedSyncRules() {
-          return this.sync_rules.hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer('1') });
+          return this.sync_rules.hydrate({ hydrationState: versionedHydrationState(1) });
         }
       };
     },

packages/service-core/test/src/sync/BucketChecksumState.test.ts

@@ -12,8 +12,9 @@ import {
   WatchFilterEvent
 } from '@/index.js';
 import { JSONBig } from '@powersync/service-jsonbig';
-import { SqliteJsonRow, ParameterLookup, SqlSyncRules, RequestJwtPayload } from '@powersync/service-sync-rules';
-import { describe, expect, test, beforeEach } from 'vitest';
+import { ParameterLookup, RequestJwtPayload, SqliteJsonRow, SqlSyncRules } from '@powersync/service-sync-rules';
+import { versionedHydrationState } from '@powersync/service-sync-rules/src/HydrationState.js';
+import { beforeEach, describe, expect, test } from 'vitest';
 
 describe('BucketChecksumState', () => {
   // Single global[] bucket.
@@ -25,7 +26,7 @@ bucket_definitions:
     data: []
     `,
     { defaultSchema: 'public' }
-  ).hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer('1') });
+  ).hydrate({ hydrationState: versionedHydrationState(1) });
 
   // global[1] and global[2]
   const SYNC_RULES_GLOBAL_TWO = SqlSyncRules.fromYaml(
@@ -38,7 +39,7 @@ bucket_definitions:
     data: []
     `,
     { defaultSchema: 'public' }
-  ).hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer('2') });
+  ).hydrate({ hydrationState: versionedHydrationState(2) });
 
   // by_project[n]
   const SYNC_RULES_DYNAMIC = SqlSyncRules.fromYaml(
@@ -49,7 +50,7 @@ bucket_definitions:
     data: []
     `,
     { defaultSchema: 'public' }
-  ).hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer('3') });
+  ).hydrate({ hydrationState: versionedHydrationState(3) });
 
   const syncContext = new SyncContext({
     maxBuckets: 100,
@@ -614,7 +615,7 @@ config:
 
       const rules = SqlSyncRules.fromYaml(source, {
         defaultSchema: 'public'
-      }).hydrate({ bucketIdTransformer: SqlSyncRules.versionedBucketIdTransformer('1') });
+      }).hydrate({ hydrationState: versionedHydrationState(1) });
 
       return new BucketChecksumState({
         syncContext,

packages/sync-rules/src/BucketSource.ts

@@ -1,24 +1,13 @@
 import { BucketParameterQuerier, ParameterLookup, PendingQueriers } from './BucketParameterQuerier.js';
 import { ColumnDefinition } from './ExpressionType.js';
-import { HydrationState, ParameterLookupScope } from './HydrationState.js';
+import { DEFAULT_HYDRATION_STATE, HydrationState, ParameterLookupScope } from './HydrationState.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { GetQuerierOptions } from './SqlSyncRules.js';
 import { TablePattern } from './TablePattern.js';
-import {
-  BucketIdTransformer,
-  EvaluatedParametersResult,
-  EvaluateRowOptions,
-  EvaluationResult,
-  SourceSchema,
-  SqliteRow
-} from './types.js';
+import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, SourceSchema, SqliteRow } from './types.js';
 
 export interface CreateSourceParams {
-  hydrationState?: HydrationState;
-  /**
-   * @deprecated Use hydrationState instead.
-   */
-  bucketIdTransformer?: BucketIdTransformer;
+  hydrationState: HydrationState;
 }
 
 /**
@@ -227,7 +216,8 @@ export function mergeParameterQuerierSources(sources: BucketParameterQuerierSour
  * it is useful to have a single merged source that can evaluate everything.
  */
 export function debugHydratedMergedSource(bucketSource: BucketSource, params?: CreateSourceParams): DebugMergedSource {
-  const resolvedParams = params ?? { bucketIdTransformer: (id: string) => id };
+  const hydrationState = params?.hydrationState ?? DEFAULT_HYDRATION_STATE;
+  const resolvedParams = { hydrationState };
   const dataSource = mergeDataSources(
     bucketSource.dataSources.map((source) => source.createDataSource(resolvedParams))
   );

packages/sync-rules/src/HydrationState.ts

@@ -1,5 +1,4 @@
 import { BucketDataSourceDefinition, BucketParameterLookupSourceDefinition } from './BucketSource.js';
-import { BucketIdTransformer, CreateSourceParams } from './index.js';
 
 export interface BucketSourceState {
   /** The prefix is the bucket name before the parameters. */
@@ -49,33 +48,29 @@ export const DEFAULT_HYDRATION_STATE: HydrationState = {
   }
 };
 
-export function versionedHydrationState(version: number) {
-  return new BucketIdTransformerHydrationState((bucketId: string) => {
-    return `${version}#${bucketId}`;
-  });
-}
-
-export class BucketIdTransformerHydrationState implements HydrationState {
-  constructor(private transformer: BucketIdTransformer) {}
-
-  getBucketSourceState(source: BucketDataSourceDefinition): BucketSourceState {
-    return {
-      bucketPrefix: this.transformer(source.defaultBucketPrefix)
-    };
-  }
-
-  getParameterLookupScope(source: BucketParameterLookupSourceDefinition): ParameterLookupScope {
-    // No transformations applied here
-    return source.defaultLookupScope;
-  }
-}
+/**
+ * Transforms bucket ids generated when evaluating the row by e.g. encoding version information.
+ *
+ * Because buckets are recreated on a sync rule redeploy, it makes sense to use different bucket ids (otherwise, clients
+ * may run into checksum errors causing a sync to take longer than necessary or breaking progress).
+ *
+ * So, this transformer receives the original bucket id as generated by defined sync rules, and can prepend a version
+ * identifier.
+ *
+ * Note that this transformation has not been present in older versions of the sync service. To preserve backwards
+ * compatibility, sync rules will not use this without an opt-in.
+ */
+export function versionedHydrationState(version: number): HydrationState {
+  return {
+    getBucketSourceState(source: BucketDataSourceDefinition): BucketSourceState {
+      return {
+        bucketPrefix: `${version}#${source.defaultBucketPrefix}`
+      };
+    },
 
-export function resolveHydrationState(params: CreateSourceParams): HydrationState {
-  if (params.hydrationState) {
-    return params.hydrationState;
-  } else if (params.bucketIdTransformer) {
-    return new BucketIdTransformerHydrationState(params.bucketIdTransformer);
-  } else {
-    return DEFAULT_HYDRATION_STATE;
-  }
+    getParameterLookupScope(source: BucketParameterLookupSourceDefinition): ParameterLookupScope {
+      // No transformations applied here
+      return source.defaultLookupScope;
+    }
+  };
 }

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -6,7 +6,6 @@ import {
   CreateSourceParams
 } from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
-import { resolveHydrationState } from './HydrationState.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { SqlDataQuery } from './SqlDataQuery.js';
@@ -158,7 +157,7 @@ export class BucketDefinitionDataSource implements BucketDataSourceDefinition {
   }
 
   createDataSource(params: CreateSourceParams): BucketDataSource {
-    const hydrationState = resolveHydrationState(params);
+    const hydrationState = params.hydrationState;
     const bucketPrefix = hydrationState.getBucketSourceState(this).bucketPrefix;
     return {
       evaluateRow: (options) => {

packages/sync-rules/src/SqlParameterQuery.ts

@@ -19,7 +19,7 @@ import {
   CreateSourceParams
 } from './BucketSource.js';
 import { SqlRuleError } from './errors.js';
-import { ParameterLookupScope, resolveHydrationState } from './HydrationState.js';
+import { ParameterLookupScope } from './HydrationState.js';
 import { BucketDataSourceDefinition, GetQuerierOptions } from './index.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { AvailableTable, SqlTools } from './sql_filters.js';
@@ -346,7 +346,7 @@ export class SqlParameterQuery
   }
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
-    const hydrationState = resolveHydrationState(params);
+    const hydrationState = params.hydrationState;
     const bucketPrefix = hydrationState.getBucketSourceState(this.querierDataSource).bucketPrefix;
     const lookupState = hydrationState.getParameterLookupScope(this);
 
@@ -359,7 +359,7 @@ export class SqlParameterQuery
   }
 
   createParameterLookupSource(params: CreateSourceParams): BucketParameterLookupSource {
-    const hydrationState = resolveHydrationState(params);
+    const hydrationState = params.hydrationState;
     const lookupState = hydrationState.getParameterLookupScope(this);
     return {
       evaluateParameterRow: (sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] => {

packages/sync-rules/src/SqlSyncRules.ts

@@ -5,19 +5,20 @@ import {
   BucketDataSourceDefinition,
   BucketParameterLookupSourceDefinition,
   BucketParameterQuerierSourceDefinition,
-  BucketSource
+  BucketSource,
+  CreateSourceParams
 } from './BucketSource.js';
 import { CompatibilityContext, CompatibilityEdition, CompatibilityOption } from './compatibility.js';
 import { SqlRuleError, SyncRulesErrors, YamlError } from './errors.js';
 import { SqlEventDescriptor } from './events/SqlEventDescriptor.js';
+import { DEFAULT_HYDRATION_STATE } from './HydrationState.js';
 import { validateSyncRulesSchema } from './json_schema.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { QueryParseResult, SqlBucketDescriptor } from './SqlBucketDescriptor.js';
 import { syncStreamFromSql } from './streams/from_sql.js';
 import { HydratedSyncRules } from './SyncRules.js';
 import { TablePattern } from './TablePattern.js';
 import {
-  BucketIdTransformer,
   QueryParseOptions,
   RequestParameters,
   SourceSchema,
@@ -388,21 +389,20 @@ export class SqlSyncRules {
   /**
    * Hydrate the sync rule definitions with persisted state into runnable sync rules.
    *
-   * Right now this is just the bucketIdTransformer, but this is expected to expand in the future to support
-   * incremental sync rule reprocessing.
-   *
-   * @param params.bucketIdTransformer A function that transforms bucket ids based on persisted state. May omit for tests.
+   * @param params.hydrationState Transforms bucket ids based on persisted state. May omit for tests.
    */
-  hydrate(params?: { bucketIdTransformer?: BucketIdTransformer }): HydratedSyncRules {
-    const bucketIdTransformer = this.compatibility.isEnabled(CompatibilityOption.versionedBucketIds)
-      ? (params?.bucketIdTransformer ?? ((id: string) => id))
-      : (id: string) => id;
+  hydrate(params?: CreateSourceParams): HydratedSyncRules {
+    let hydrationState = params?.hydrationState;
+    if (hydrationState == null || !this.compatibility.isEnabled(CompatibilityOption.versionedBucketIds)) {
+      hydrationState = DEFAULT_HYDRATION_STATE;
+    }
+    const resolvedParams = { hydrationState };
     return new HydratedSyncRules({
       definition: this,
-      createParams: { bucketIdTransformer },
-      bucketDataSources: this.bucketDataSources.map((d) => d.createDataSource({ bucketIdTransformer })),
+      createParams: resolvedParams,
+      bucketDataSources: this.bucketDataSources.map((d) => d.createDataSource(resolvedParams)),
       bucketParameterLookupSources: this.bucketParameterLookupSources.map((d) =>
-        d.createParameterLookupSource({ bucketIdTransformer })
+        d.createParameterLookupSource(resolvedParams)
       ),
       eventDescriptors: this.eventDescriptors,
       compatibility: this.compatibility
@@ -488,8 +488,4 @@ export class SqlSyncRules {
       }
     }
   }
-
-  static versionedBucketIdTransformer(version: string) {
-    return (bucketId: string) => `${version}#${bucketId}`;
-  }
 }

packages/sync-rules/src/StaticSqlParameterQuery.ts

@@ -7,7 +7,6 @@ import {
   CreateSourceParams
 } from './BucketSource.js';
 import { SqlRuleError } from './errors.js';
-import { resolveHydrationState } from './HydrationState.js';
 import { BucketDataSourceDefinition, GetQuerierOptions } from './index.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { AvailableTable, SqlTools } from './sql_filters.js';
@@ -180,7 +179,7 @@ export class StaticSqlParameterQuery implements BucketParameterQuerierSourceDefi
   }
 
   createParameterQuerierSource(params: CreateSourceParams): BucketParameterQuerierSource {
-    const hydrationState = resolveHydrationState(params);
+    const hydrationState = params.hydrationState;
     const bucketPrefix = hydrationState.getBucketSourceState(this.querierDataSource).bucketPrefix;
     return {
       pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {