Support rowvalue IN subquery syntax

Author: simolus3

packages/sync-rules/src/BaseSqlDataQuery.ts

@@ -36,7 +36,6 @@ export interface BaseSqlDataQueryOptions {
   descriptorName: string;
   bucketParameters: string[];
   tools: SqlTools;
-
   errors?: SqlRuleError[];
 }
 

packages/sync-rules/src/SqlBucketDescriptor.ts

@@ -1,4 +1,4 @@
-import { BucketDescription, BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
+import { BucketInclusionReason, ResolvedBucket } from './BucketDescription.js';
 import { BucketParameterQuerier, mergeBucketParameterQueriers } from './BucketParameterQuerier.js';
 import { BucketSource, ResultSetDescription } from './BucketSource.js';
 import { IdSequence } from './IdSequence.js';
@@ -7,7 +7,6 @@ import { SqlDataQuery } from './SqlDataQuery.js';
 import { SqlParameterQuery } from './SqlParameterQuery.js';
 import { GetQuerierOptions, SyncRulesOptions } from './SqlSyncRules.js';
 import { StaticSqlParameterQuery } from './StaticSqlParameterQuery.js';
-import { StreamQuery } from './StreamQuery.js';
 import { TablePattern } from './TablePattern.js';
 import { TableValuedFunctionSqlParameterQuery } from './TableValuedFunctionSqlParameterQuery.js';
 import { SqlRuleError } from './errors.js';
@@ -18,8 +17,7 @@ import {
   QueryParseOptions,
   RequestParameters,
   SourceSchema,
-  SqliteRow,
-  StreamParseOptions
+  SqliteRow
 } from './types.js';
 
 export interface QueryParseResult {
@@ -96,24 +94,6 @@ export class SqlBucketDescriptor implements BucketSource {
     };
   }
 
-  addUnifiedStreamQuery(sql: string, options: StreamParseOptions): QueryParseResult {
-    const [query, errors] = StreamQuery.fromSql(this.name, sql, options);
-    for (const parameterQuery of query.inferredParameters) {
-      if (parameterQuery instanceof StaticSqlParameterQuery) {
-        this.globalParameterQueries.push(parameterQuery);
-      } else {
-        this.parameterQueries.push(parameterQuery);
-      }
-    }
-    this.dataQueries.push(query.data);
-    this.subscribedToByDefault = options.default ?? false;
-
-    return {
-      parsed: true,
-      errors
-    };
-  }
-
   evaluateRow(options: EvaluateRowOptions): EvaluationResult[] {
     let results: EvaluationResult[] = [];
     for (let query of this.dataQueries) {

packages/sync-rules/src/SqlSyncRules.ts

@@ -26,6 +26,8 @@ import {
   SyncRules
 } from './types.js';
 import { BucketSource } from './BucketSource.js';
+import { SyncStream } from './streams/stream.js';
+import { syncStreamFromSql } from './streams/from_sql.js';
 
 const ACCEPT_POTENTIALLY_DANGEROUS_QUERIES = Symbol('ACCEPT_POTENTIALLY_DANGEROUS_QUERIES');
 
@@ -214,8 +216,6 @@ export class SqlSyncRules implements SyncRules {
         continue;
       }
 
-      const descriptor = new SqlBucketDescriptor(key, SqlBucketDescriptorType.STREAM);
-
       const accept_potentially_dangerous_queries =
         value.get('accept_potentially_dangerous_queries', true)?.value == true;
 
@@ -229,14 +229,17 @@ export class SqlSyncRules implements SyncRules {
       const data = value.get('query', true) as unknown;
       if (data instanceof Scalar) {
         rules.withScalar(data, (q) => {
-          return descriptor.addUnifiedStreamQuery(q, queryOptions);
+          const [parsed, errors] = syncStreamFromSql(key, q, options);
+          rules.bucketSources.push(parsed);
+          return {
+            parsed: true,
+            errors
+          };
         });
       } else {
         rules.errors.push(this.tokenError(data, 'Must be a string.'));
         continue;
       }
-
-      rules.bucketSources.push(descriptor);
     }
 
     const eventMap = parsed.get('event_definitions') as YAMLMap;

packages/sync-rules/src/StreamQuery.ts

@@ -18,6 +18,7 @@ export * from './SqlParameterQuery.js';
 export * from './SqlSyncRules.js';
 export * from './StaticSchema.js';
 export { SyncStream } from './streams/stream.js';
+export { syncStreamFromSql } from './streams/from_sql.js';
 export * from './TablePattern.js';
 export * from './types.js';
 export * from './utils.js';

packages/sync-rules/src/index.ts

@@ -28,7 +28,7 @@ const request_parameters: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: true
 };
 
-const request_jwt: SqlParameterFunction = {
+export const request_jwt: SqlParameterFunction = {
   debugName: 'request.jwt',
   call(parameters: ParameterValueSet) {
     return parameters.rawTokenPayload;
@@ -42,7 +42,7 @@ const request_jwt: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: false
 };
 
-const request_user_id: SqlParameterFunction = {
+export const request_user_id: SqlParameterFunction = {
   debugName: 'request.user_id',
   call(parameters: ParameterValueSet) {
     return parameters.userId;
@@ -61,20 +61,3 @@ export const REQUEST_FUNCTIONS: Record<string, SqlParameterFunction> = {
   jwt: request_jwt,
   user_id: request_user_id
 };
-
-export const QUERY_FUNCTIONS: Record<string, SqlParameterFunction> = {
-  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
-  }
-};