Claude/async typed object factory cz yhz

cockpit-core/src/main/java/com/orbitalhq/cockpit/core/security/authentication/oidc/PropelAuthApiKeyValidator.kt

@@ -19,6 +19,7 @@ import org.springframework.stereotype.Component
 import org.springframework.web.client.RestClient
 import org.springframework.web.client.toEntity
 import reactor.core.publisher.Mono
+import reactor.core.scheduler.Schedulers
 import java.time.Instant
 
 @ConditionalOnProperty("vyne.security.openIdp.roles.format", havingValue = CloudPropelAuthClaimsExtractor.PropelAuthJwtKind, matchIfMissing = false)
@@ -83,7 +84,7 @@ class PropelAuthApiKeyValidator(
 //         sink.success(JwtAuthenticationToken(
 //            jwt
 //         ))
-      }
+      }.subscribeOn(Schedulers.boundedElastic()) as Mono<Authentication>
 
 
    }

function-loading/functions-binding/src/main/java/com/orbitalhq/functions/scanner/BoundFunction.kt

@@ -44,7 +44,7 @@ class BoundFunction(
 
    override val functionName: QualifiedName = QualifiedName(namespace, name)
 
-   override fun invoke(
+   override suspend fun invoke(
       inputValues: List<TypedInstance>,
       schema: Schema,
       returnType: Type,

pipelines/stream-engine/src/main/java/com/orbitalhq/pipelines/jet/streams/ResultStreamAuthorizationDecorator.kt

@@ -47,7 +47,7 @@ class ResultStreamAuthorizationDecorator(
       return stream.mapNotNull { value ->
          // first, parse back to a typed instance
          val valueAsTypedInstance = TypedInstance.from(instanceType, value, querySchema, source = Provided)
-         val evaluatedTypedInstance = policyEvaluator.evaluate(valueAsTypedInstance, queryContext, executionScope)
+         val evaluatedTypedInstance = kotlinx.coroutines.runBlocking { policyEvaluator.evaluate(valueAsTypedInstance, queryContext, executionScope) }
          // Convert back to a raw object, since that's what we started with
          evaluatedTypedInstance.toRawObject()
       }

taxiql-query-engine/src/main/java/com/orbitalhq/Vyne.kt

@@ -230,7 +230,7 @@ class Vyne(
    }
 
    @VisibleForTesting
-   internal fun buildContextAndExpression(
+   internal suspend fun buildContextAndExpression(
       taxiQl: TaxiQlQuery,
       queryId: String,
       clientQueryId: String?,
@@ -277,7 +277,7 @@ class Vyne(
       )
    }
 
-   private fun convertTaxiQlFactToInstances(
+   private suspend fun convertTaxiQlFactToInstances(
       taxiQl: TaxiQlQuery,
       arguments: Map<String, Any?>,
       executionContextFacts: Set<Fact> = emptySet()
@@ -314,27 +314,26 @@ class Vyne(
       // to us are available in the expressions we're evaluating.
       // eg:
       // given { name : String = 'foo' , upper = upperCase(name) }
-      val evaluatedExpressions = taxiQl.facts
-         .filter { it.value is FactValue.Expression }
-         .fold(constants) { previousParams, parameter ->
-            val facts = CopyOnWriteFactBag(emptyList(), schema, previousParams)
-            val valueSupplier = FactBagValueSupplier(facts, schema)
-            val accessorReader = AccessorReader(
-               valueSupplier,
-               schema.functionRegistry,
-               schema
-            )
+      var evaluatedExpressions = constants
+      for (parameter in taxiQl.facts.filter { it.value is FactValue.Expression }) {
+         val facts = CopyOnWriteFactBag(emptyList(), schema, evaluatedExpressions)
+         val valueSupplier = FactBagValueSupplier(facts, schema)
+         val accessorReader = AccessorReader(
+            valueSupplier,
+            schema.functionRegistry,
+            schema
+         )
 
-            val expression = parameter.value as FactValue.Expression
-            val evaluationResult = accessorReader.evaluate(
-               value = facts,
-               returnType = schema.type(parameter.type),
-               expression = expression.expression,
-               format = null,
-               dataSource = Provided
-            )
-            previousParams + ScopedFact(ProjectionFunctionScope(parameter.name, parameter.type), evaluationResult)
-         }
+         val expression = parameter.value as FactValue.Expression
+         val evaluationResult = accessorReader.evaluate(
+            value = facts,
+            returnType = schema.type(parameter.type),
+            expression = expression.expression,
+            format = null,
+            dataSource = Provided
+         )
+         evaluatedExpressions = evaluatedExpressions + ScopedFact(ProjectionFunctionScope(parameter.name, parameter.type), evaluationResult)
+      }
       return evaluatedExpressions.map { it.scope.name to it.fact }
          .toMap()
 
@@ -423,14 +422,16 @@ class Vyne(
       // to our predicate.
       // That's wrong, as generally the collection will be the input, especially if our predciate / expression
       // is a contains(...)
-      val buildResult = TypedObjectFactory(
-         expressionType,
-         queryContext.facts,
-         schemaWithType,
-         source = Provided,
-         inPlaceQueryEngine = queryContext,
-         functionResultCache = queryContext.functionResultCache
-      ).build()
+      val buildResult = kotlinx.coroutines.runBlocking {
+         TypedObjectFactory(
+            expressionType,
+            queryContext.facts,
+            schemaWithType,
+            source = Provided,
+            inPlaceQueryEngine = queryContext,
+            functionResultCache = queryContext.functionResultCache
+         ).build()
+      }
       return buildResult
    }
 

taxiql-query-engine/src/main/java/com/orbitalhq/query/DirectServiceInvocationStrategy.kt

@@ -68,7 +68,7 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
       return invokeOperations(operations, context, target)
    }
 
-   private fun lookForCandidateServices(
+   private suspend fun lookForCandidateServices(
       context: QueryContext,
       target: Set<QuerySpecTypeNode>
    ): Map<QuerySpecTypeNode, Map<RemoteOperation, Map<Parameter, TypedInstance>>> {
@@ -84,13 +84,16 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
    /**
     * Returns the operations that we can invoke, grouped by target query node.
     */
-   private fun getCandidateOperations(
+   private suspend fun getCandidateOperations(
       schema: Schema,
       target: Set<QuerySpecTypeNode>,
       context: QueryContext
    ): Map<QuerySpecTypeNode, Map<RemoteOperation, Map<Parameter, TypedInstance>>> {
-      val grouped = target.map { it to getCandidateOperations(schema, it, context) }
-         .groupBy({ it.first }, { it.second })
+      val grouped = mutableMapOf<QuerySpecTypeNode, MutableList<Map<RemoteOperation, Map<Parameter, TypedInstance>>>>()
+      for (item in target) {
+         val candidates = getCandidateOperations(schema, item, context)
+         grouped.getOrPut(item) { mutableListOf() }.add(candidates)
+      }
 
       val result = grouped.mapValues { (_, operationParameterMaps) ->
          operationParameterMaps.reduce { acc, map -> acc + map }
@@ -103,7 +106,7 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
     * (either because they have no parameters, or because all their parameters are populated by constraints)
     * and the set of parameters that we have identified values for
     */
-   private fun getCandidateOperations(
+   private suspend fun getCandidateOperations(
       schema: Schema,
       target: QuerySpecTypeNode,
       context: QueryContext
@@ -114,34 +117,28 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
             it.returnType.isAssignableTo(target.type) && it.operationType == OperationScope.READ_ONLY
          }
       }
-      val operations = operationsForType
-         .mapNotNull { operation ->
-            val (satisfiesConstraints, operationParameters) = compareOperationContractToDataRequirementsAndFetchSearchParams(
-               operation,
-               target,
-               schema,
-               context
-            )
-            if (!satisfiesConstraints) {
-               null
-            } else {
-               operation to operationParameters
-            }
-         }
-         .map { (operation, parameters) ->
-            populateParamsFromContextValues(operation, parameters, context)
+      val result = mutableMapOf<RemoteOperation, Map<Parameter, TypedInstance>>()
+      for (operation in operationsForType) {
+         val (satisfiesConstraints, operationParameters) = compareOperationContractToDataRequirementsAndFetchSearchParams(
+            operation,
+            target,
+            schema,
+            context
+         )
+         if (!satisfiesConstraints) continue
+
+         val (op1, params1) = populateParamsFromContextValues(operation, operationParameters, context)
+         val (op2, params2) = provideUnpopulatedParametersWithDefaults(op1, params1, context)
+
+         // Check to see if there are any outstanding parameters that haven't been populated
+         val unpopulatedParams = op2.parameters.filter { parameter ->
+            !params2.containsKey(parameter) && !parameter.nullable
          }
-         .map { (operation, parameters) ->
-            provideUnpopulatedParametersWithDefaults(operation, parameters, context)
+         if (unpopulatedParams.isEmpty()) {
+            result[op2] = params2
          }
-         .filter { (operation, populatedOperationParameters) ->
-            // Check to see if there are any outstanding parameters that haven't been populated
-            val unpopulatedParams = operation.parameters.filter { parameter ->
-               !populatedOperationParameters.containsKey(parameter) && !parameter.nullable
-            }
-            unpopulatedParams.isEmpty()
-         }
-      return operations.toMap()
+      }
+      return result
    }
 
    /**
@@ -166,17 +163,17 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
     * Adds any parameters that are so far unpopulated, but
     * have a default expression that can be used to populate them
     */
-   private fun provideUnpopulatedParametersWithDefaults(
+   private suspend fun provideUnpopulatedParametersWithDefaults(
       operation: RemoteOperation,
       parameters: Map<Parameter, TypedInstance>,
       context: QueryContext
    ): Pair<RemoteOperation, Map<Parameter, TypedInstance>> {
-      val defaultValues = operation.parameters
-         .filter { it.defaultValue != null }
-         .filter { !parameters.containsKey(it) }
-         .associateWith { parameter ->
-            context.evaluate(parameter.defaultValue!!)
+      val defaultValues = mutableMapOf<Parameter, TypedInstance>()
+      for (param in operation.parameters) {
+         if (param.defaultValue != null && !parameters.containsKey(param)) {
+            defaultValues[param] = context.evaluate(param.defaultValue!!)
          }
+      }
       return operation to (parameters + defaultValues)
    }
 
@@ -185,7 +182,7 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
     * If a contract exists on the target which provides input params, and the operation
     * can satisfy the contract, then the parameters to search inputs are returned mapped.
     */
-   private fun compareOperationContractToDataRequirementsAndFetchSearchParams(
+   private suspend fun compareOperationContractToDataRequirementsAndFetchSearchParams(
       remoteOperation: RemoteOperation,
       target: QuerySpecTypeNode,
       schema: Schema,
@@ -208,12 +205,11 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
       // when evaluted against a contract of
       // find { Film[](PublicationDate >= 2020-10-02)
       // would provide a value of `date`
-      val satisfiedConstraints = targetDataConstraints.mapNotNull { requiredConstraint ->
+      val satisfiedConstraints = mutableListOf<Pair<lang.taxi.services.operations.constraints.Constraint, List<Pair<Parameter, TypedInstance>>>>()
+      for (requiredConstraint in targetDataConstraints) {
          val constraintComparison = remoteOperation.contract.satisfies(requiredConstraint)
          if (constraintComparison.satisfiesRequestedConstraint) {
-            requiredConstraint to getProvidedParameterValues(remoteOperation, constraintComparison.providedValues, context, schema)
-         } else {
-            null
+            satisfiedConstraints.add(requiredConstraint to getProvidedParameterValues(remoteOperation, constraintComparison.providedValues, context, schema))
          }
       }
 
@@ -224,19 +220,20 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
       return allOperationConstraintsSatisfied to operationConstraintParameterValues
    }
 
-   private fun getProvidedParameterValues(
+   private suspend fun getProvidedParameterValues(
       remoteOperation: Operation,
       providedValues: List<Pair<Expression, Expression>>,
       context: QueryContext,
       schema: Schema
    ): List<Pair<Parameter, TypedInstance>> {
-      return providedValues.mapNotNull { (paramExpression, providedValueExpression) ->
+      val result = mutableListOf<Pair<Parameter, TypedInstance>>()
+      for ((paramExpression, providedValueExpression) in providedValues) {
          when (paramExpression) {
             is ArgumentSelector -> {
                val parameter = remoteOperation.parameter(paramExpression.scopeWithPath)
                if (parameter == null) {
                   logger.warn { "An expression was found to provide a value for parameter ${paramExpression.path}, but no such parameter exists on operation ${remoteOperation.name}" }
-                  return@mapNotNull null
+                  continue
                }
                // Short circut - if it's a literal (which is the most common case), then
                // provide the value
@@ -250,15 +247,15 @@ class DirectServiceInvocationStrategy(invocationService: OperationInvocationServ
 
                   else -> context.evaluate(expression = providedValueExpression)
                }
-               parameter to expressionResult
+               result.add(parameter to expressionResult)
             }
 
             else -> {
                logger.warn { "Not implemented: Mapping parameterExpression of type ${paramExpression::class.simpleName}" }
-               null
             }
          }
       }
+      return result
    }
 }
 

taxiql-query-engine/src/main/java/com/orbitalhq/query/ObjectBuilder.kt

@@ -192,7 +192,7 @@ class ObjectBuilder(
          }
    }
 
-   private fun evaluateExpressionType(targetType: Type): TypedInstance {
+   private suspend fun evaluateExpressionType(targetType: Type): TypedInstance {
       return TypedObjectFactory(
          targetType,
          // Note: This used to be an empty list,
@@ -232,7 +232,7 @@ class ObjectBuilder(
     *   items: Thing[] by [ThingToIterate[]   with { CustomerName }]
     * }[]
     */
-   private fun buildCollectionWithProjectionExpression(targetType: Type): TypedInstance {
+   private suspend fun buildCollectionWithProjectionExpression(targetType: Type): TypedInstance {
       val collectionProjectionBuilder = accessorReaders.filterIsInstance<CollectionProjectionBuilder>().firstOrNull()
          ?: error("No CollectionProjectionBuilder was present in the acessor readers")
       return collectionProjectionBuilder.process(

taxiql-query-engine/src/main/java/com/orbitalhq/query/QueryContext.kt

@@ -596,7 +596,7 @@ data class QueryContext(
 
    }
 
-   override fun evaluate(expression: Expression, facts: FactBag, source: DataSource): TypedInstance {
+   override suspend fun evaluate(expression: Expression, facts: FactBag, source: DataSource): TypedInstance {
       return TypedObjectFactory(
          schema.type(expression.returnType),
          facts.withAdditionalScopedFacts(this.scopedFacts, schema),
@@ -607,7 +607,7 @@ data class QueryContext(
       ).evaluateExpression(expression)
    }
 
-   override fun evaluate(expression: Expression, value: TypedInstance, source: DataSource): TypedInstance {
+   override suspend fun evaluate(expression: Expression, value: TypedInstance, source: DataSource): TypedInstance {
       return TypedObjectFactory(
          schema.type(expression.returnType),
          value,