Bring back BucketSource.

Author: rkistner

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

@@ -2,6 +2,7 @@ import {
   BucketDataSourceDefinition,
   BucketDescription,
   BucketPriority,
+  BucketSource,
   HydratedSyncRules,
   RequestedStream,
   RequestJwtPayload,
@@ -248,15 +249,15 @@ export class BucketChecksumState {
       const streamNameToIndex = new Map<string, number>();
       this.streamNameToIndex = streamNameToIndex;
 
-      for (const source of this.parameterState.syncRules.bucketDataSources) {
-        if (this.parameterState.isSubscribedToStream(source.definition)) {
-          streamNameToIndex.set(source.definition.name, subscriptions.length);
+      for (const source of this.parameterState.syncRules.definition.bucketSources) {
+        if (this.parameterState.isSubscribedToStream(source)) {
+          streamNameToIndex.set(source.name, subscriptions.length);
 
           subscriptions.push({
-            name: source.definition.name,
-            is_default: source.definition.subscribedToByDefault,
+            name: source.name,
+            is_default: source.subscribedToByDefault,
             errors:
-              this.parameterState.streamErrors[source.definition.name]?.map((e) => ({
+              this.parameterState.streamErrors[source.name]?.map((e) => ({
                 subscription: e.subscription?.opaque_id ?? 'default',
                 message: e.message
               })) ?? []
@@ -484,7 +485,7 @@ export class BucketParameterState {
     };
   }
 
-  isSubscribedToStream(desc: BucketDataSourceDefinition): boolean {
+  isSubscribedToStream(desc: BucketSource): boolean {
     return (desc.subscribedToByDefault && this.includeDefaultStreams) || this.subscribedStreamNames.has(desc.name);
   }
 

packages/sync-rules/src/BucketSource.ts

@@ -16,13 +16,43 @@ export interface CreateSourceParams {
   bucketIdTransformer: BucketIdTransformer;
 }
 
+export interface BucketSource {
+  readonly name: string;
+  readonly type: BucketSourceType;
+  readonly subscribedToByDefault: boolean;
+
+  /**
+   * BucketDataSource describing the data in this bucket/stream definition.
+   *
+   * The same data source could in theory be present in multiple stream definitions.
+   */
+  readonly dataSource: BucketDataSourceDefinition;
+
+  /**
+   * BucketParameterQuerierSource describing the parameter queries / stream subqueries in this bucket/stream definition.
+   *
+   * The same source could in theory be present in multiple stream definitions.
+   */
+  readonly parameterQuerierSources: BucketParameterQuerierSourceDefinition[];
+
+  /**
+   * BucketParameterLookupSource describing the parameter tables used in this bucket/stream definition.
+   *
+   * The same source could in theory be present in multiple stream definitions.
+   */
+  readonly parameterLookupSources: BucketParameterLookupSourceDefinition[];
+}
+
+export interface HydratedBucketSource {
+  readonly definition: BucketSource;
+
+  readonly parameterQuerierSources: BucketParameterQuerierSource[];
+}
+
 /**
  * Encodes a static definition of a bucket source, as parsed from sync rules or stream definitions.
  */
 export interface BucketDataSourceDefinition {
-  readonly name: string;
-  readonly type: BucketSourceType;
-  readonly subscribedToByDefault: boolean;
   /**
    * For debug use only.
    */
@@ -50,8 +80,6 @@ export interface BucketDataSourceDefinition {
  * This is only relevant for parameter queries that query tables.
  */
 export interface BucketParameterLookupSourceDefinition {
-  readonly name: string;
-  readonly type: BucketSourceType;
   /**
    * For debug use only.
    */
@@ -70,9 +98,6 @@ export interface BucketParameterLookupSourceDefinition {
  * This may use request data only, or it may use parameter lookup data persisted by a BucketParameterLookupSourceDefinition.
  */
 export interface BucketParameterQuerierSourceDefinition {
-  readonly name: string;
-  readonly type: BucketSourceType;
-  readonly subscribedToByDefault: boolean;
   /**
    * For debug use only.
    */

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -4,7 +4,8 @@ import {
   BucketParameterLookupSourceDefinition,
   BucketParameterQuerierSourceDefinition,
   BucketSourceType,
-  CreateSourceParams
+  CreateSourceParams,
+  BucketSource
 } from './BucketSource.js';
 import { ColumnDefinition } from './ExpressionType.js';
 import { IdSequence } from './IdSequence.js';
@@ -28,10 +29,21 @@ export interface QueryParseResult {
   errors: SqlRuleError[];
 }
 
-export class SqlBucketDescriptor implements BucketDataSourceDefinition {
+export class SqlBucketDescriptor implements BucketDataSourceDefinition, BucketSource {
   name: string;
   private bucketParametersInternal: string[] | null = null;
 
+  public readonly subscribedToByDefault: boolean = true;
+
+  /**
+   * source table -> queries
+   */
+  dataQueries: SqlDataQuery[] = [];
+  parameterQueries: SqlParameterQuery[] = [];
+  globalParameterQueries: (StaticSqlParameterQuery | TableValuedFunctionSqlParameterQuery)[] = [];
+
+  parameterIdSequence = new IdSequence();
+
   constructor(name: string) {
     this.name = name;
   }
@@ -40,22 +52,21 @@ export class SqlBucketDescriptor implements BucketDataSourceDefinition {
     return BucketSourceType.SYNC_RULE;
   }
 
-  public get subscribedToByDefault(): boolean {
-    return true;
-  }
-
   public get bucketParameters(): string[] {
     return this.bucketParametersInternal ?? [];
   }
 
-  /**
-   * source table -> queries
-   */
-  dataQueries: SqlDataQuery[] = [];
-  parameterQueries: SqlParameterQuery[] = [];
-  globalParameterQueries: (StaticSqlParameterQuery | TableValuedFunctionSqlParameterQuery)[] = [];
+  get dataSource() {
+    return this;
+  }
 
-  parameterIdSequence = new IdSequence();
+  get parameterLookupSources() {
+    return this.parameterQueries;
+  }
+
+  get parameterQuerierSources() {
+    return [...this.parameterQueries, ...this.globalParameterQueries];
+  }
 
   addDataQuery(sql: string, options: SyncRulesOptions, compatibility: CompatibilityContext): QueryParseResult {
     if (this.bucketParametersInternal == null) {
@@ -112,14 +123,6 @@ export class SqlBucketDescriptor implements BucketDataSourceDefinition {
     };
   }
 
-  getParameterQuerierSourceDefinitions(): BucketParameterQuerierSourceDefinition[] {
-    return [...this.parameterQueries, ...this.globalParameterQueries];
-  }
-
-  getParameterLookupSourceDefinitions(): BucketParameterLookupSourceDefinition[] {
-    return [...this.parameterQueries];
-  }
-
   getSourceTables(): Set<TablePattern> {
     let result = new Set<TablePattern>();
     for (let query of this.parameterQueries) {

packages/sync-rules/src/SqlSyncRules.ts

@@ -4,7 +4,8 @@ import { BucketParameterQuerier, QuerierError } from './BucketParameterQuerier.j
 import {
   BucketDataSourceDefinition,
   BucketParameterLookupSourceDefinition,
-  BucketParameterQuerierSourceDefinition
+  BucketParameterQuerierSourceDefinition,
+  BucketSource
 } from './BucketSource.js';
 import { CompatibilityContext, CompatibilityEdition, CompatibilityOption } from './compatibility.js';
 import { SqlRuleError, SyncRulesErrors, YamlError } from './errors.js';
@@ -85,6 +86,7 @@ export class SqlSyncRules {
   bucketDataSources: BucketDataSourceDefinition[] = [];
   bucketParameterLookupSources: BucketParameterLookupSourceDefinition[] = [];
   bucketParameterQuerierSources: BucketParameterQuerierSourceDefinition[] = [];
+  bucketSources: BucketSource[] = [];
 
   eventDescriptors: SqlEventDescriptor[] = [];
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
@@ -236,9 +238,11 @@ export class SqlSyncRules {
           return descriptor.addDataQuery(q, queryOptions, compatibility);
         });
       }
-      rules.bucketDataSources.push(descriptor);
-      rules.bucketParameterLookupSources.push(...descriptor.getParameterLookupSourceDefinitions());
-      rules.bucketParameterQuerierSources.push(...descriptor.getParameterQuerierSourceDefinitions());
+
+      rules.bucketSources.push(descriptor);
+      rules.bucketDataSources.push(descriptor.dataSource);
+      rules.bucketParameterLookupSources.push(...descriptor.parameterLookupSources);
+      rules.bucketParameterQuerierSources.push(...descriptor.parameterQuerierSources);
     }
 
     for (const entry of streamMap?.items ?? []) {
@@ -263,9 +267,10 @@ export class SqlSyncRules {
       if (data instanceof Scalar) {
         rules.withScalar(data, (q) => {
           const [parsed, errors] = syncStreamFromSql(key, q, queryOptions);
-          rules.bucketDataSources.push(parsed);
-          rules.bucketParameterLookupSources.push(parsed);
-          rules.bucketParameterQuerierSources.push(parsed);
+          rules.bucketSources.push(parsed);
+          rules.bucketDataSources.push(parsed.dataSource);
+          rules.bucketParameterLookupSources.push(...parsed.parameterLookupSources);
+          rules.bucketParameterQuerierSources.push(...parsed.parameterQuerierSources);
           return {
             parsed: true,
             errors

packages/sync-rules/src/SyncRules.ts

@@ -1,4 +1,10 @@
-import { BucketDataSource, BucketParameterLookupSource, BucketParameterQuerierSource } from './BucketSource.js';
+import {
+  BucketDataSource,
+  BucketParameterLookupSource,
+  BucketParameterQuerierSource,
+  BucketParameterQuerierSourceDefinition,
+  HydratedBucketSource
+} from './BucketSource.js';
 import {
   BucketParameterQuerier,
   CompatibilityContext,
@@ -25,14 +31,15 @@ import { EvaluatedParametersResult, EvaluateRowOptions, EvaluationResult, Sqlite
  * specifically affects bucket names.
  */
 export class HydratedSyncRules {
+  bucketSources: HydratedBucketSource[] = [];
   bucketDataSources: BucketDataSource[];
   bucketParameterQuerierSources: BucketParameterQuerierSource[];
   bucketParameterLookupSources: BucketParameterLookupSource[];
 
   eventDescriptors: SqlEventDescriptor[] = [];
   compatibility: CompatibilityContext = CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY;
 
-  private definition: SqlSyncRules;
+  readonly definition: SqlSyncRules;
 
   constructor(params: {
     definition: SqlSyncRules;
@@ -46,12 +53,29 @@ export class HydratedSyncRules {
     this.bucketParameterQuerierSources = params.bucketParameterQuerierSources;
     this.bucketParameterLookupSources = params.bucketParameterLookupSources;
     this.definition = params.definition;
+
     if (params.eventDescriptors) {
       this.eventDescriptors = params.eventDescriptors;
     }
     if (params.compatibility) {
       this.compatibility = params.compatibility;
     }
+
+    let querierMap = new Map<BucketParameterQuerierSourceDefinition, HydratedBucketSource>();
+    for (let definition of this.definition.bucketSources) {
+      const hydratedBucketSource: HydratedBucketSource = { definition: definition, parameterQuerierSources: [] };
+      this.bucketSources.push(hydratedBucketSource);
+      for (let querier of definition.parameterQuerierSources) {
+        querierMap.set(querier, hydratedBucketSource);
+      }
+    }
+    for (let querier of params.bucketParameterQuerierSources) {
+      const bucketSource = querierMap.get(querier.definition);
+      if (bucketSource == null) {
+        throw new Error('Cannot find BucketSource for BucketParameterQuerierSource');
+      }
+      bucketSource.parameterQuerierSources.push(querier);
+    }
   }
 
   // These methods do not depend on hydration, so we can just forward them to the definition.
@@ -131,12 +155,14 @@ export class HydratedSyncRules {
     const errors: QuerierError[] = [];
     const pending = { queriers, errors };
 
-    for (const source of this.bucketParameterQuerierSources) {
+    for (const source of this.bucketSources) {
       if (
         (source.definition.subscribedToByDefault && options.hasDefaultStreams) ||
         source.definition.name in options.streams
       ) {
-        source.pushBucketParameterQueriers(pending, options);
+        for (let querier of source.parameterQuerierSources) {
+          querier.pushBucketParameterQueriers(pending, options);
+        }
       }
     }
 

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

@@ -8,6 +8,7 @@ import {
   BucketParameterLookupSourceDefinition,
   BucketParameterQuerierSource,
   BucketParameterQuerierSourceDefinition,
+  BucketSource,
   BucketSourceType,
   CreateSourceParams
 } from '../BucketSource.js';
@@ -27,7 +28,11 @@ import {
 import { StreamVariant } from './variant.js';
 
 export class SyncStream
-  implements BucketDataSourceDefinition, BucketParameterLookupSourceDefinition, BucketParameterQuerierSourceDefinition
+  implements
+    BucketDataSourceDefinition,
+    BucketParameterLookupSourceDefinition,
+    BucketParameterQuerierSourceDefinition,
+    BucketSource
 {
   name: string;
   subscribedToByDefault: boolean;
@@ -53,6 +58,18 @@ export class SyncStream
     return this.data.bucketParameters;
   }
 
+  public get dataSource() {
+    return this;
+  }
+
+  public get parameterLookupSources() {
+    return [this];
+  }
+
+  public get parameterQuerierSources() {
+    return [this];
+  }
+
   createDataSource(params: CreateSourceParams): BucketDataSource {
     return {
       definition: this,