Implement new parameter syntax

simolus3 · simolus3 · commit b5fd3f0b3942 · 2025-08-11T14:12:01.000+02:00
diff --git a/packages/sync-rules/src/request_functions.ts b/packages/sync-rules/src/request_functions.ts
@@ -1,10 +1,12 @@
 import { ExpressionType } from './ExpressionType.js';
+import { jsonExtract, jsonExtractFromRecord } from './sql_functions.js';
 import { ParameterValueSet, SqliteValue } from './types.js';
 
 export interface SqlParameterFunction {
   readonly debugName: string;
-  call: (parameters: ParameterValueSet) => SqliteValue;
+  call: (parameters: ParameterValueSet, ...args: SqliteValue[]) => SqliteValue;
   getReturnType(): ExpressionType;
+  parameterCount: number;
   /** request.user_id(), request.jwt(), token_parameters.* */
   usesAuthenticatedRequestParameters: boolean;
   /** request.parameters(), user_parameters.* */
@@ -13,23 +15,83 @@ export interface SqlParameterFunction {
   documentation: string;
 }
 
-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
-};
+/**
+ * Defines a `parameters` function and a `parameter` function.
+ *
+ * The `parameters` function extracts a JSON object from the {@link ParameterValueSet} while the `parameter` function
+ * takes a second argument (a JSON path or a single key) to extract.
+ */
+export function parameterFunctions(options: {
+  schema: string;
+  extractJsonString: (v: ParameterValueSet) => string;
+  extractJsonParsed: (v: ParameterValueSet) => any;
+  sourceDescription: string;
+  sourceDocumentation: string;
+  usesAuthenticatedRequestParameters: boolean;
+  usesUnauthenticatedRequestParameters: boolean;
+}) {
+  const allParameters: SqlParameterFunction = {
+    debugName: `${options.schema}.parameters`,
+    parameterCount: 0,
+    call(parameters: ParameterValueSet) {
+      return options.extractJsonString(parameters);
+    },
+    getReturnType() {
+      return ExpressionType.TEXT;
+    },
+    detail: options.sourceDescription,
+    documentation: `Returns ${options.sourceDocumentation}`,
+    usesAuthenticatedRequestParameters: options.usesAuthenticatedRequestParameters,
+    usesUnauthenticatedRequestParameters: options.usesUnauthenticatedRequestParameters
+  };
+
+  const extractParameter: SqlParameterFunction = {
+    debugName: `${options.schema}.parameter`,
+    parameterCount: 1,
+    call(parameters: ParameterValueSet, path) {
+      const parsed = options.extractJsonParsed(parameters);
+      if (typeof path == 'string') {
+        if (path.startsWith('$.')) {
+          return jsonExtractFromRecord(parsed, path, '->>');
+        } else {
+          return parsed[path];
+        }
+      }
+
+      return null;
+    },
+    getReturnType() {
+      return ExpressionType.ANY;
+    },
+    detail: `Extract value from ${options.sourceDescription}`,
+    documentation: `Returns an extracted value (via the key as the second argument) from ${options.sourceDocumentation}`,
+    usesAuthenticatedRequestParameters: options.usesAuthenticatedRequestParameters,
+    usesUnauthenticatedRequestParameters: options.usesUnauthenticatedRequestParameters
+  };
+
+  return { parameters: allParameters, parameter: extractParameter };
+}
+
+export function globalRequestParameterFunctions(schema: string) {
+  return parameterFunctions({
+    schema,
+    extractJsonString: function (v: ParameterValueSet): string {
+      return v.rawUserParameters;
+    },
+    extractJsonParsed: function (v: ParameterValueSet) {
+      return v.userParameters;
+    },
+    sourceDescription: 'Unauthenticated request parameters as JSON',
+    sourceDocumentation:
+      '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
+  });
+}
 
 export const request_jwt: SqlParameterFunction = {
   debugName: 'request.jwt',
+  parameterCount: 0,
   call(parameters: ParameterValueSet) {
     return parameters.rawTokenPayload;
   },
@@ -44,6 +106,7 @@ export const request_jwt: SqlParameterFunction = {
 
 export const request_user_id: SqlParameterFunction = {
   debugName: 'request.user_id',
+  parameterCount: 0,
   call(parameters: ParameterValueSet) {
     return parameters.userId;
   },
@@ -56,8 +119,8 @@ export const request_user_id: SqlParameterFunction = {
   usesUnauthenticatedRequestParameters: false
 };
 
-export const REQUEST_FUNCTIONS_NAMED = {
-  parameters: request_parameters,
+const REQUEST_FUNCTIONS_NAMED = {
+  ...globalRequestParameterFunctions('request'),
   jwt: request_jwt,
   user_id: request_user_id
 };
diff --git a/packages/sync-rules/src/sql_filters.ts b/packages/sync-rules/src/sql_filters.ts
@@ -4,7 +4,7 @@ import { nil } from 'pgsql-ast-parser/src/utils.js';
 import { BucketPriority, isValidPriority } from './BucketDescription.js';
 import { ExpressionType } from './ExpressionType.js';
 import { SqlRuleError } from './errors.js';
-import { REQUEST_FUNCTIONS } from './request_functions.js';
+import { REQUEST_FUNCTIONS, SqlParameterFunction } from './request_functions.js';
 import {
   BASIC_OPERATORS,
   OPERATOR_IN,
@@ -96,7 +96,10 @@ export interface SqlToolsOptions {
    */
   supportsParameterExpressions?: boolean;
 
-  isStream?: boolean;
+  /**
+   * For each schema, all available parameter functions.
+   */
+  parameterFunctions?: Record<string, Record<string, SqlParameterFunction>>;
 
   /**
    * Schema for validations.
@@ -117,7 +120,7 @@ export class SqlTools {
 
   readonly supportsExpandingParameters: boolean;
   readonly supportsParameterExpressions: boolean;
-  readonly isSyncStream: boolean;
+  readonly parameterFunctions: Record<string, Record<string, SqlParameterFunction>>;
 
   schema?: QuerySchema;
 
@@ -136,7 +139,7 @@ export class SqlTools {
     this.sql = options.sql;
     this.supportsExpandingParameters = options.supportsExpandingParameters ?? false;
     this.supportsParameterExpressions = options.supportsParameterExpressions ?? false;
-    this.isSyncStream = options.isStream ?? false;
+    this.parameterFunctions = options.parameterFunctions ?? { request: REQUEST_FUNCTIONS };
   }
 
   error(message: string, expr: NodeLocation | Expr | undefined): ClauseError {
@@ -309,6 +312,7 @@ export class SqlTools {
     } else if (expr.type == 'call' && expr.function?.name != null) {
       const schema = expr.function.schema; // schema.function()
       const fn = expr.function.name;
+
       if (schema == null) {
         // Just fn()
         const fnImpl = SQL_FUNCTIONS[fn];
@@ -319,36 +323,41 @@ export class SqlTools {
         const argClauses = expr.args.map((arg) => this.compileClause(arg));
         const composed = this.composeFunction(fnImpl, argClauses, expr.args);
         return composed;
-      } else if (schema == 'request' && !this.isSyncStream) {
-        // Special function
+      } else if (schema in this.parameterFunctions) {
         if (!this.supportsParameterExpressions) {
           return this.error(`${schema} schema is not available in data queries`, expr);
         }
 
-        if (expr.args.length > 0) {
-          return this.error(`Function '${schema}.${fn}' does not take arguments`, expr);
-        }
+        const impl = this.parameterFunctions[schema][fn];
 
-        if (fn in REQUEST_FUNCTIONS) {
-          const fnImpl = REQUEST_FUNCTIONS[fn];
-          return {
-            key: 'request.parameters()',
-            lookupParameterValue(parameters) {
-              return fnImpl.call(parameters);
-            },
-            usesAuthenticatedRequestParameters: fnImpl.usesAuthenticatedRequestParameters,
-            usesUnauthenticatedRequestParameters: fnImpl.usesUnauthenticatedRequestParameters
-          } satisfies ParameterValueClause;
-        } else {
-          return this.error(`Function '${schema}.${fn}' is not defined`, expr);
-        }
-      } else if (this.isSyncStream && schema in STREAM_FUNCTIONS) {
-        const impl = STREAM_FUNCTIONS[schema][fn];
         if (impl) {
+          if (expr.args.length != impl.parameterCount) {
+            return this.error(`Function '${schema}.${fn}' takes ${impl.parameterCount} arguments.`, expr);
+          }
+
+          const compiledArguments = expr.args.map(this.compileClause);
+          let hasInvalidArgument = false;
+          for (let i = 0; i < expr.args.length; i++) {
+            const argument = compiledArguments[i];
+
+            if (!isParameterValueClause(argument)) {
+              hasInvalidArgument = true;
+              if (!isClauseError(argument)) {
+                this.error('Must only depend on data derived from request.', expr.args[i]);
+              }
+            }
+          }
+
+          if (hasInvalidArgument) {
+            return { error: true };
+          }
+
+          const parameterArguments = compiledArguments as ParameterValueClause[];
           return {
-            key: `${schema}.${fn}()`,
+            key: `${schema}.${fn}(${parameterArguments.map((p) => p.key).join(',')})`,
             lookupParameterValue(parameters) {
-              return impl.call(parameters);
+              const evaluatedArgs = parameterArguments.map((p) => p.lookupParameterValue(parameters));
+              return impl.call(parameters, ...evaluatedArgs);
             },
             usesAuthenticatedRequestParameters: impl.usesAuthenticatedRequestParameters,
             usesUnauthenticatedRequestParameters: impl.usesUnauthenticatedRequestParameters
diff --git a/packages/sync-rules/src/sql_functions.ts b/packages/sync-rules/src/sql_functions.ts
@@ -1,6 +1,6 @@
 import { JSONBig } from '@powersync/service-jsonbig';
 import { SQLITE_FALSE, SQLITE_TRUE, sqliteBool, sqliteNot } from './sql_support.js';
-import { SqliteValue } from './types.js';
+import { SqliteRow, SqliteValue } from './types.js';
 import { jsonValueToSqlite } from './utils.js';
 // Declares @syncpoint/wkx module
 // This allows for consumers of this lib to resolve types correctly
@@ -870,8 +870,17 @@ function concat(a: SqliteValue, b: SqliteValue): string | null {
 
 export function jsonExtract(sourceValue: SqliteValue, path: SqliteValue, operator: string) {
   const valueText = castAsText(sourceValue);
+  if (valueText == null || path == null) {
+    return null;
+  }
+
+  let value = JSONBig.parse(valueText) as any;
+  return jsonExtractFromRecord(value, path, operator);
+}
+
+export function jsonExtractFromRecord(value: any, path: SqliteValue, operator: string) {
   const pathText = castAsText(path);
-  if (valueText == null || pathText == null) {
+  if (value == null || pathText == null) {
     return null;
   }
 
@@ -882,7 +891,6 @@ export function jsonExtract(sourceValue: SqliteValue, path: SqliteValue, operato
     throw new Error(`JSON path must start with $.`);
   }
 
-  let value = JSONBig.parse(valueText) as any;
   for (let c of components) {
     if (value == null) {
       break;
diff --git a/packages/sync-rules/src/streams/from_sql.ts b/packages/sync-rules/src/streams/from_sql.ts
@@ -40,6 +40,7 @@ import {
   SelectStatement,
   Statement
 } from 'pgsql-ast-parser';
+import { STREAM_FUNCTIONS } from './functions.js';
 
 export function syncStreamFromSql(
   descriptorName: string,
@@ -80,9 +81,9 @@ class SyncStreamCompiler {
       valueTables: [alias],
       sql: this.sql,
       schema: querySchema,
+      parameterFunctions: STREAM_FUNCTIONS,
       supportsParameterExpressions: true,
-      supportsExpandingParameters: true, // needed for table.column IN (subscription.parameters() -> ...)
-      isStream: true
+      supportsExpandingParameters: true // needed for table.column IN (subscription.parameters() -> ...)
     });
     tools.checkSpecificNameCase(tableRef);
     let filter = this.whereClauseToFilters(tools, query.where);
@@ -372,7 +373,7 @@ class SyncStreamCompiler {
       sql: this.sql,
       schema: querySchema,
       supportsParameterExpressions: true,
-      isStream: true
+      parameterFunctions: STREAM_FUNCTIONS
     });
     tools.checkSpecificNameCase(tableRef);
 
diff --git a/packages/sync-rules/src/streams/functions.ts b/packages/sync-rules/src/streams/functions.ts
@@ -1,46 +1,45 @@
-import { ExpressionType } from '../ExpressionType.js';
-import { request_jwt, request_user_id, SqlParameterFunction } from '../request_functions.js';
+import {
+  globalRequestParameterFunctions,
+  parameterFunctions,
+  request_user_id,
+  SqlParameterFunction
+} from '../request_functions.js';
 import { ParameterValueSet } from '../types.js';
 
-const subscription_parameters: SqlParameterFunction = {
-  debugName: 'subscription.parameters',
-  call(parameters: ParameterValueSet) {
-    return parameters.rawStreamParameters;
-  },
-  getReturnType() {
-    return ExpressionType.TEXT;
-  },
-  detail: 'Unauthenticated subscription parameters as JSON',
-  documentation:
-    'Returns parameters passed by the client for this stream as a JSON string. These parameters are not authenticated - any value can be passed in by the client.',
-  usesAuthenticatedRequestParameters: false,
-  usesUnauthenticatedRequestParameters: true
-};
-
-const connection_parameters: SqlParameterFunction = {
-  debugName: 'connection.parameters',
-  call(parameters: ParameterValueSet) {
-    return parameters.rawUserParameters;
-  },
-  getReturnType() {
-    return ExpressionType.TEXT;
-  },
-  detail: 'Unauthenticated connection 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
-};
-
 export const STREAM_FUNCTIONS: Record<string, Record<string, SqlParameterFunction>> = {
   subscription: {
-    parameters: subscription_parameters
+    ...parameterFunctions({
+      schema: 'subscription',
+      extractJsonString: function (v: ParameterValueSet): string {
+        return v.rawStreamParameters ?? '{}';
+      },
+      extractJsonParsed: function (v: ParameterValueSet) {
+        return v.streamParameters ?? {};
+      },
+      sourceDescription: 'Unauthenticated subscription parameters as JSON',
+      sourceDocumentation:
+        'parameters passed by the client for this stream as a JSON string. These parameters are not authenticated - any value can be passed in by the client.',
+      usesAuthenticatedRequestParameters: false,
+      usesUnauthenticatedRequestParameters: true
+    })
   },
   connection: {
-    parameters: connection_parameters
+    ...globalRequestParameterFunctions('connection')
   },
-  token: {
+  auth: {
     user_id: request_user_id,
-    jwt: request_jwt
+    ...parameterFunctions({
+      schema: 'auth',
+      extractJsonString: function (v: ParameterValueSet): string {
+        return v.rawTokenPayload;
+      },
+      extractJsonParsed: function (v: ParameterValueSet) {
+        return v.tokenParameters;
+      },
+      sourceDescription: 'JWT payload as JSON',
+      sourceDocumentation: 'JWT payload as a JSON string. This is always validated against trusted keys',
+      usesAuthenticatedRequestParameters: true,
+      usesUnauthenticatedRequestParameters: false
+    })
   }
 };
diff --git a/packages/sync-rules/src/types.ts b/packages/sync-rules/src/types.ts
diff --git a/packages/sync-rules/test/src/streams.test.ts b/packages/sync-rules/test/src/streams.test.ts
