Split out sync rule definitions from compiled ones.

Author: rkistner

packages/service-core/src/routes/endpoints/sync-rules.ts

@@ -202,7 +202,7 @@ async function debugSyncRules(apiHandler: RouteAPI, sync_rules: string) {
 
     return {
       valid: true,
-      bucket_definitions: rules.bucketSources.map((source) => source.debugRepresentation()),
+      bucket_definitions: rules.debugRepresentation(),
       source_tables: resolved_tables,
       data_tables: rules.debugGetOutputTables()
     };

packages/sync-rules/src/BucketParameterQuerier.ts

@@ -1,4 +1,4 @@
-import { BucketDescription, ResolvedBucket } from './BucketDescription.js';
+import { ResolvedBucket } from './BucketDescription.js';
 import { RequestedStream } from './SqlSyncRules.js';
 import { RequestParameters, SqliteJsonRow, SqliteJsonValue } from './types.js';
 import { normalizeParameterValue } from './utils.js';

packages/sync-rules/src/BucketSource.ts

@@ -3,41 +3,32 @@ import { ColumnDefinition } from './ExpressionType.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
 import { GetQuerierOptions } from './SqlSyncRules.js';
 import { TablePattern } from './TablePattern.js';
-import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, SourceSchema, SqliteRow } from './types.js';
+import {
+  BucketIdTransformer,
+  EvaluatedParametersResult,
+  EvaluateRowOptions,
+  EvaluationResult,
+  SourceSchema,
+  SqliteRow
+} from './types.js';
+
+export interface CreateSourceParams {
+  bucketIdTransformer: BucketIdTransformer;
+}
 
 /**
- * An interface declaring
- *
- *  - which buckets the sync service should create when processing change streams from the database.
- *  - how data in source tables maps to data in buckets (e.g. when we're not selecting all columns).
- *  - which buckets a given connection has access to.
- *
- * There are two ways to define bucket sources: Via sync rules made up of parameter and data queries, and via stream
- * definitions that only consist of a single query.
+ * Encodes a static definition of a bucket source, as parsed from sync rules or stream definitions.
  */
-export interface BucketDataSource {
+export interface BucketDataSourceDefinition {
   readonly name: string;
   readonly type: BucketSourceType;
-
   readonly subscribedToByDefault: boolean;
-
-  /**
-   * Given a row as it appears in a table that affects sync data, return buckets, logical table names and transformed
-   * data for rows to add to buckets.
-   */
-  evaluateRow(options: EvaluateRowOptions): EvaluationResult[];
-
   /**
-   * Reports {@link BucketParameterQuerier}s resolving buckets that a specific stream request should have access to.
-   *
-   * @param result The target array to insert queriers and errors into.
-   * @param options Options, including parameters that may affect the buckets loaded by this source.
+   * For debug use only.
    */
-  pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions): void;
-
+  readonly bucketParameters: string[];
   getSourceTables(): Set<TablePattern>;
-
-  /** Whether the table possibly affects the contents of buckets resolved by this source. */
+  createDataSource(params: CreateSourceParams): BucketDataSource;
   tableSyncsData(table: SourceTableInterface): boolean;
 
   /**
@@ -53,6 +44,30 @@ export interface BucketDataSource {
   debugRepresentation(): any;
 }
 
+export interface BucketParameterSourceDefinition {
+  readonly name: string;
+  readonly type: BucketSourceType;
+  readonly subscribedToByDefault: boolean;
+
+  getSourceTables(): Set<TablePattern>;
+  createParameterSource(params: CreateSourceParams): BucketParameterSource;
+
+  /**
+   * Whether {@link pushBucketParameterQueriers} may include a querier where
+   * {@link BucketParameterQuerier.hasDynamicBuckets} is true.
+   *
+   * This is mostly used for testing.
+   */
+  hasDynamicBucketQueries(): boolean;
+
+  getSourceTables(): Set<TablePattern>;
+
+  /** Whether the table possibly affects the buckets resolved by this source. */
+  tableSyncsParameters(table: SourceTableInterface): boolean;
+
+  debugRepresentation(): any;
+}
+
 /**
  * An interface declaring
  *
@@ -63,12 +78,28 @@ export interface BucketDataSource {
  * There are two ways to define bucket sources: Via sync rules made up of parameter and data queries, and via stream
  * definitions that only consist of a single query.
  */
-export interface BucketParameterSource {
-  readonly name: string;
-  readonly type: BucketSourceType;
+export interface BucketDataSource {
+  readonly definition: BucketDataSourceDefinition;
 
-  readonly subscribedToByDefault: boolean;
+  /**
+   * Given a row as it appears in a table that affects sync data, return buckets, logical table names and transformed
+   * data for rows to add to buckets.
+   */
+  evaluateRow(options: EvaluateRowOptions): EvaluationResult[];
+}
 
+/**
+ * An interface declaring
+ *
+ *  - which buckets the sync service should create when processing change streams from the database.
+ *  - how data in source tables maps to data in buckets (e.g. when we're not selecting all columns).
+ *  - which buckets a given connection has access to.
+ *
+ * There are two ways to define bucket sources: Via sync rules made up of parameter and data queries, and via stream
+ * definitions that only consist of a single query.
+ */
+export interface BucketParameterSource {
+  readonly definition: BucketParameterSourceDefinition;
   /**
    * Given a row in a source table that affects sync parameters, returns a structure to index which buckets rows should
    * be associated with.
@@ -87,21 +118,9 @@ export interface BucketParameterSource {
   pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions): void;
 
   /**
-   * Whether {@link pushBucketParameterQueriers} may include a querier where
-   * {@link BucketParameterQuerier.hasDynamicBuckets} is true.
-   *
-   * This is mostly used for testing.
+   * @deprecated Use `pushBucketParameterQueriers` instead and merge at the top-level.
    */
-  hasDynamicBucketQueries(): boolean;
-
-  getSourceTables(): Set<TablePattern>;
-
-  /** Whether the table possibly affects the buckets resolved by this source. */
-  tableSyncsParameters(table: SourceTableInterface): boolean;
-
-  debugWriteOutputTables(result: Record<string, { query: string }[]>): void;
-
-  debugRepresentation(): any;
+  getBucketParameterQuerier(options: GetQuerierOptions): BucketParameterQuerier;
 }
 
 export enum BucketSourceType {

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -1,6 +1,14 @@
 import { BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
 import { BucketParameterQuerier, mergeBucketParameterQueriers, PendingQueriers } from './BucketParameterQuerier.js';
-import { BucketDataSource, BucketParameterSource, BucketSourceType, ResultSetDescription } from './BucketSource.js';
+import {
+  BucketDataSource,
+  BucketDataSourceDefinition,
+  BucketParameterSource,
+  BucketParameterSourceDefinition,
+  BucketSourceType,
+  CreateSourceParams,
+  ResultSetDescription
+} from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
 import { IdSequence } from './IdSequence.js';
 import { SourceTableInterface } from './SourceTableInterface.js';
@@ -32,9 +40,9 @@ export interface QueryParseResult {
   errors: SqlRuleError[];
 }
 
-export class SqlBucketDescriptor implements BucketDataSource, BucketParameterSource {
+export class SqlBucketDescriptor implements BucketDataSourceDefinition, BucketParameterSourceDefinition {
   name: string;
-  bucketParameters?: string[];
+  private bucketParametersInternal: string[] | null = null;
 
   constructor(name: string) {
     this.name = name;
@@ -48,6 +56,10 @@ export class SqlBucketDescriptor implements BucketDataSource, BucketParameterSou
     return true;
   }
 
+  public get bucketParameters(): string[] {
+    return this.bucketParametersInternal ?? [];
+  }
+
   /**
    * source table -> queries
    */
@@ -58,10 +70,10 @@ export class SqlBucketDescriptor implements BucketDataSource, BucketParameterSou
   parameterIdSequence = new IdSequence();
 
   addDataQuery(sql: string, options: SyncRulesOptions, compatibility: CompatibilityContext): QueryParseResult {
-    if (this.bucketParameters == null) {
+    if (this.bucketParametersInternal == null) {
       throw new Error('Bucket parameters must be defined');
     }
-    const dataRows = SqlDataQuery.fromSql(this.name, this.bucketParameters, sql, options, compatibility);
+    const dataRows = SqlDataQuery.fromSql(this.name, this.bucketParametersInternal, sql, options, compatibility);
 
     this.dataQueries.push(dataRows);
 
@@ -73,11 +85,12 @@ export class SqlBucketDescriptor implements BucketDataSource, BucketParameterSou
 
   addParameterQuery(sql: string, options: QueryParseOptions): QueryParseResult {
     const parameterQuery = SqlParameterQuery.fromSql(this.name, sql, options, this.parameterIdSequence.nextId());
-    if (this.bucketParameters == null) {
-      this.bucketParameters = parameterQuery.bucketParameters;
+    if (this.bucketParametersInternal == null) {
+      this.bucketParametersInternal = parameterQuery.bucketParameters;
     } else {
       if (
-        new Set([...parameterQuery.bucketParameters!, ...this.bucketParameters]).size != this.bucketParameters.length
+        new Set([...parameterQuery.bucketParameters!, ...this.bucketParametersInternal]).size !=
+        this.bucketParametersInternal.length
       ) {
         throw new Error('Bucket parameters must match for each parameter query within a bucket');
       }
@@ -94,61 +107,71 @@ export class SqlBucketDescriptor implements BucketDataSource, BucketParameterSou
     };
   }
 
-  evaluateRow(options: EvaluateRowOptions): EvaluationResult[] {
-    let results: EvaluationResult[] = [];
-    for (let query of this.dataQueries) {
-      if (!query.applies(options.sourceTable)) {
-        continue;
+  createDataSource(params: CreateSourceParams): BucketDataSource {
+    return {
+      definition: this,
+      evaluateRow: (options) => {
+        let results: EvaluationResult[] = [];
+        for (let query of this.dataQueries) {
+          if (!query.applies(options.sourceTable)) {
+            continue;
+          }
+
+          results.push(...query.evaluateRow(options.sourceTable, options.record, params.bucketIdTransformer));
+        }
+        return results;
       }
-
-      results.push(...query.evaluateRow(options.sourceTable, options.record, options.bucketIdTransformer));
-    }
-    return results;
+    };
   }
 
-  evaluateParameterRow(sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] {
-    let results: EvaluatedParametersResult[] = [];
-    for (let query of this.parameterQueries) {
-      if (query.applies(sourceTable)) {
-        results.push(...query.evaluateParameterRow(row));
+  createParameterSource(params: CreateSourceParams): BucketParameterSource {
+    return {
+      definition: this,
+
+      evaluateParameterRow: (sourceTable: SourceTableInterface, row: SqliteRow): EvaluatedParametersResult[] => {
+        let results: EvaluatedParametersResult[] = [];
+        for (let query of this.parameterQueries) {
+          if (query.applies(sourceTable)) {
+            results.push(...query.evaluateParameterRow(row));
+          }
+        }
+        return results;
+      },
+      pushBucketParameterQueriers: (result: PendingQueriers, options: GetQuerierOptions) => {
+        const reasons = [this.bucketInclusionReason()];
+        const staticBuckets = this.getStaticBucketDescriptions(
+          options.globalParameters,
+          reasons,
+          params.bucketIdTransformer
+        );
+        const staticQuerier = {
+          staticBuckets,
+          hasDynamicBuckets: false,
+          parameterQueryLookups: [],
+          queryDynamicBucketDescriptions: async () => []
+        } satisfies BucketParameterQuerier;
+        result.queriers.push(staticQuerier);
+
+        if (this.parameterQueries.length == 0) {
+          return;
+        }
+
+        const dynamicQueriers = this.parameterQueries.map((query) =>
+          query.getBucketParameterQuerier(options.globalParameters, reasons, params.bucketIdTransformer)
+        );
+        result.queriers.push(...dynamicQueriers);
+      },
+
+      /**
+       * @deprecated Use `pushBucketParameterQueriers` instead and merge at the top-level.
+       */
+      getBucketParameterQuerier(options: GetQuerierOptions): BucketParameterQuerier {
+        const queriers: BucketParameterQuerier[] = [];
+        this.pushBucketParameterQueriers({ queriers, errors: [] }, options);
+
+        return mergeBucketParameterQueriers(queriers);
       }
-    }
-    return results;
-  }
-
-  /**
-   * @deprecated Use `pushBucketParameterQueriers` instead and merge at the top-level.
-   */
-  getBucketParameterQuerier(options: GetQuerierOptions): BucketParameterQuerier {
-    const queriers: BucketParameterQuerier[] = [];
-    this.pushBucketParameterQueriers({ queriers, errors: [] }, options);
-
-    return mergeBucketParameterQueriers(queriers);
-  }
-
-  pushBucketParameterQueriers(result: PendingQueriers, options: GetQuerierOptions) {
-    const reasons = [this.bucketInclusionReason()];
-    const staticBuckets = this.getStaticBucketDescriptions(
-      options.globalParameters,
-      reasons,
-      options.bucketIdTransformer
-    );
-    const staticQuerier = {
-      staticBuckets,
-      hasDynamicBuckets: false,
-      parameterQueryLookups: [],
-      queryDynamicBucketDescriptions: async () => []
-    } satisfies BucketParameterQuerier;
-    result.queriers.push(staticQuerier);
-
-    if (this.parameterQueries.length == 0) {
-      return;
-    }
-
-    const dynamicQueriers = this.parameterQueries.map((query) =>
-      query.getBucketParameterQuerier(options.globalParameters, reasons, options.bucketIdTransformer)
-    );
-    result.queriers.push(...dynamicQueriers);
+    };
   }
 
   getStaticBucketDescriptions(