Rename more stream subscription references

Author: simolus3

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

@@ -388,8 +388,8 @@ export class BucketParameterState {
 
     this.querier = syncRules.getBucketParameterQuerier({
       globalParameters: this.syncParams,
-      hasDefaultSubscriptions: this.includeDefaultStreams,
-      resolveSubscription(name) {
+      hasDefaultStreams: this.includeDefaultStreams,
+      resolveOpenedStream(name) {
         const subscription = explicitlyOpenedStreams[name];
         if (subscription) {
           return subscription.parameters ?? {};

packages/sync-rules/src/SqlSyncRules.ts

@@ -42,17 +42,19 @@ export interface SyncRulesOptions {
 export interface GetQuerierOptions {
   globalParameters: RequestParameters;
   /**
-   * Whether the client is subscribing to default subscriptions (the default).
+   * Whether the client is subscribing to default query streams.
+   *
+   * Client do this by default, but can disable the behavior if needed.
    */
-  hasDefaultSubscriptions: boolean;
+  hasDefaultStreams: boolean;
   /**
-   * For streams, this is invoked to check whether the client has requested a subscription to
-   * the stream.
+   * For streams, this is invoked to check whether the client has opened the relevant stream.
    *
    * @param name The name of the stream as it appears in the sync rule definitions.
-   * @returns If a subscription is active, the stream parameters for that particular stream. Otherwise null.
+   * @returns If the strema has been opened by the client, the stream parameters for that particular stream. Otherwise
+   * null.
    */
-  resolveSubscription: (name: string) => Record<string, any> | null;
+  resolveOpenedStream: (name: string) => Record<string, any> | null;
 }
 
 export class SqlSyncRules implements SyncRules {
@@ -376,15 +378,15 @@ export class SqlSyncRules implements SyncRules {
       let params = options.globalParameters;
 
       if (descriptor.type == SqlBucketDescriptorType.STREAM) {
-        const subscription = options.resolveSubscription(descriptor.name);
+        const opened = options.resolveOpenedStream(descriptor.name);
 
-        if (!descriptor.subscribedToByDefault && subscription == null) {
+        if (!descriptor.subscribedToByDefault && opened == null) {
           // The client is not subscribing to this stream, so don't query buckets related to it.
           continue;
         }
 
-        if (subscription != null) {
-          params = params.withAddedParameters(subscription);
+        if (opened != null) {
+          params = params.withAddedStreamParameters(opened);
         }
 
         queriers.push(descriptor.getBucketParameterQuerier(params));

packages/sync-rules/src/TableValuedFunctionSqlParameterQuery.ts

@@ -222,9 +222,7 @@ export class TableValuedFunctionSqlParameterQuery {
 
   private getIndividualBucketDescription(row: SqliteRow, parameters: RequestParameters): BucketDescription | null {
     const mergedParams: ParameterValueSet = {
-      rawTokenPayload: parameters.rawTokenPayload,
-      rawUserParameters: parameters.rawUserParameters,
-      userId: parameters.userId,
+      ...parameters,
       lookup: (table, column) => {
         if (table == this.callTableName) {
           return row[column]!;

packages/sync-rules/src/request_functions.ts

@@ -13,21 +13,19 @@ export interface SqlParameterFunction {
   documentation: string;
 }
 
-const parametersFunction = (name: string): SqlParameterFunction => {
-  return {
-    debugName: name,
-    call(parameters: ParameterValueSet) {
-      return parameters.rawUserParameters;
-    },
-    getReturnType() {
-      return ExpressionType.TEXT;
-    },
-    detail: 'Unauthenticated request parameters as JSON',
-    documentation:
-      'Returns parameters passed by the client as a JSON string. These parameters are not authenticated - any value can be passed in by the client.',
-    usesAuthenticatedRequestParameters: false,
-    usesUnauthenticatedRequestParameters: true
-  };
+const request_parameters: SqlParameterFunction = {
+  debugName: 'request.parameters',
+  call(parameters: ParameterValueSet) {
+    return parameters.rawUserParameters;
+  },
+  getReturnType() {
+    return ExpressionType.TEXT;
+  },
+  detail: 'Unauthenticated request parameters as JSON',
+  documentation:
+    'Returns parameters passed by the client as a JSON string. These parameters are not authenticated - any value can be passed in by the client.',
+  usesAuthenticatedRequestParameters: false,
+  usesUnauthenticatedRequestParameters: true
 };
 
 const request_jwt: SqlParameterFunction = {
@@ -58,16 +56,25 @@ const request_user_id: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: false
 };
 
-export const REQUEST_FUNCTIONS_WITHOUT_PARAMETERS: Record<string, SqlParameterFunction> = {
+export const REQUEST_FUNCTIONS: Record<string, SqlParameterFunction> = {
+  parameters: request_parameters,
   jwt: request_jwt,
   user_id: request_user_id
 };
 
-export const REQUEST_FUNCTIONS: Record<string, SqlParameterFunction> = {
-  parameters: parametersFunction('request.parameters'),
-  ...REQUEST_FUNCTIONS_WITHOUT_PARAMETERS
-};
-
 export const QUERY_FUNCTIONS: Record<string, SqlParameterFunction> = {
-  params: parametersFunction('query.params')
+  params: {
+    debugName: 'stream.params',
+    call(parameters: ParameterValueSet) {
+      return parameters.rawUserParameters;
+    },
+    getReturnType() {
+      return ExpressionType.TEXT;
+    },
+    detail: 'Unauthenticated stream parameters as JSON',
+    documentation:
+      'Returns stream passed by the client when opening the stream. These parameters are not authenticated - any value can be passed in by the client.',
+    usesAuthenticatedRequestParameters: false,
+    usesUnauthenticatedRequestParameters: true
+  }
 };

packages/sync-rules/src/sql_filters.ts

@@ -427,18 +427,14 @@ export class SqlTools {
           return this.error(`${schema} schema is not available in data queries`, expr);
         }
 
-        if (fn == 'parameters' && this.supportsStreamInputs) {
-          return this.error(`'request.parameters()' is unavailable on streams - use 'stream.params()' instead.`, expr);
-        }
-
         if (expr.args.length > 0) {
           return this.error(`Function '${schema}.${fn}' does not take arguments`, expr);
         }
 
         if (fn in REQUEST_FUNCTIONS) {
           const fnImpl = REQUEST_FUNCTIONS[fn];
           return {
-            key: 'request.parameters()',
+            key: `stream.${fn}()`,
             lookupParameterValue(parameters) {
               return fnImpl.call(parameters);
             },
@@ -456,7 +452,7 @@ export class SqlTools {
         if (fn in QUERY_FUNCTIONS) {
           const fnImpl = QUERY_FUNCTIONS[fn];
           return {
-            key: 'stream.params()',
+            key: `stream.${fn}()`,
             lookupParameterValue(parameters) {
               return fnImpl.call(parameters);
             },

packages/sync-rules/src/types.ts

@@ -85,6 +85,11 @@ export interface ParameterValueSet {
    */
   rawUserParameters: string;
 
+  /**
+   * For streams, the raw JSON string of stream parameters.
+   */
+  rawStreamParameters: string | null;
+
   /**
    * JSON string of raw request parameters.
    */
@@ -102,6 +107,8 @@ export class RequestParameters implements ParameterValueSet {
    */
   rawUserParameters: string;
 
+  rawStreamParameters: string | null;
+
   /**
    * JSON string of raw request parameters.
    */
@@ -125,6 +132,7 @@ export class RequestParameters implements ParameterValueSet {
 
     this.rawUserParameters = JSONBig.stringify(clientParameters);
     this.userParameters = toSyncRulesParameters(clientParameters);
+    this.rawStreamParameters = null;
   }
 
   lookup(table: string, column: string): SqliteJsonValue {
@@ -136,10 +144,9 @@ export class RequestParameters implements ParameterValueSet {
     throw new Error(`Unknown table: ${table}`);
   }
 
-  withAddedParameters(params: Record<string, any>): RequestParameters {
+  withAddedStreamParameters(params: Record<string, any>): RequestParameters {
     const clone = structuredClone(this);
-    clone.rawUserParameters = JSONBig.stringify(params);
-    clone.userParameters = toSyncRulesParameters(params);
+    clone.rawStreamParameters = JSONBig.stringify(params);
 
     return clone;
   }