diff --git a/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts b/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts new file mode 100644 index 0000000..ae5f827 --- /dev/null +++ b/ts/examples/simple-example-ts/src/functions/forEachRuntimeFunction.ts @@ -0,0 +1,33 @@ +import { + Consumer, + DisplayMessage, + FunctionContext, + Identifier, + List, + Name, + Parameter, + Signature, +} from "@code0-tech/hercules"; + +@Identifier("for_each_runtime") +@Signature("(list: LIST, consumer: CONSUMER): void") +@Name({code: "en-US", content: "For Each"}) +@DisplayMessage({code: "en-US", content: "For each element of ${list} do ${consumer}"}) +@Parameter({ + runtimeName: "list", + name: [{code: "en-US", content: "List"}], + description: [{code: "en-US", content: "The list whose elements are iterated over"}], +}) +@Parameter({ + runtimeName: "consumer", + name: [{code: "en-US", content: "Consumer"}], + description: [{code: "en-US", content: "The sub flow (item) => void executed once per element"}], +}) +export class ForEachRuntimeFunction { + async run(_context: FunctionContext, list: List, consumer: Consumer): Promise { + for (const element of list) { + const result = await consumer(element); + console.log(`[for_each] sub flow result:`, result); + } + } +} diff --git a/ts/examples/simple-example-ts/src/index.ts b/ts/examples/simple-example-ts/src/index.ts index 27af5fe..07a2698 100644 --- a/ts/examples/simple-example-ts/src/index.ts +++ b/ts/examples/simple-example-ts/src/index.ts @@ -1,11 +1,12 @@ import {Action, CodeZeroEvent} from "@code0-tech/hercules"; import {FibonacciRuntimeFunction} from "./functions/fibonacciRuntimeFunction.js"; import {FibonacciFunction} from "./functions/fibonacciFunction.js"; +import {ForEachRuntimeFunction} from "./functions/forEachRuntimeFunction.js"; import {UserCreatedRuntimeEvent} from "./events/userCreatedRuntimeEvent.js"; import {EmailDataType} from "./data_types/emailDataType.js"; const action = new Action( - process.env.ACTION_ID ?? "example-action", + process.env.ACTION_ID ?? "testing-action", process.env.VERSION ?? "0.0.0", process.env.AQUILA_URL ?? "127.0.0.1:8081", "code0-tech", @@ -25,6 +26,9 @@ action.registerRuntimeFunction(FibonacciRuntimeFunction); // Function: named public variant that extends the runtime function action.registerFunction(FibonacciFunction); +// Runtime function that executes a sub flow parameter for each element of a list +action.registerRuntimeFunction(ForEachRuntimeFunction); + // Data type: derived from Zod schema action.registerDataTypeClass(EmailDataType); @@ -36,17 +40,23 @@ action.on(CodeZeroEvent.connected, () => { console.log("Connected to aquila"); }); -action.on(CodeZeroEvent.streamMessageReceived, (message: any) => { - console.log(message); -}); - action.on(CodeZeroEvent.error, (error: Error) => { console.error("Stream error:", error.message); - process.exit(0); + console.log("Attempting to reconnect in 5s..."); + setTimeout(() => { + action.connect(process.env.AUTH_TOKEN ?? "your_auth_token_here").catch((err: Error) => { + action.emit(CodeZeroEvent.error, err); + }) + }, 5000); }); - -action.connect(process.env.AUTH_TOKEN ?? "token").catch((err: unknown) => { - console.error("Failed to connect:", err); - process.exit(1); +action.connect(process.env.AUTH_TOKEN ?? "your_auth_token_here").catch((err: Error) => { + action.emit(CodeZeroEvent.error, err); }); + +action.on(CodeZeroEvent.moduleUpdated, (message: any) => { + console.dir(message, {depth: null}); + console.dir(action.configs.values(), {depth: null}) +}) + +export {action}; diff --git a/ts/package-lock.json b/ts/package-lock.json index 3fd2fff..60ad0c9 100644 --- a/ts/package-lock.json +++ b/ts/package-lock.json @@ -8,8 +8,11 @@ "name": "@code0-tech/hercules", "version": "0.0.0", "license": "ISC", + "bin": { + "hercules": "bin/hercules.js" + }, "devDependencies": { - "@code0-tech/tucana": "0.0.74", + "@code0-tech/tucana": "0.0.80", "@emnapi/core": "^1.11.1", "@emnapi/runtime": "^1.11.1", "@grpc/grpc-js": "^1.14.3", @@ -116,9 +119,9 @@ } }, "node_modules/@code0-tech/tucana": { - "version": "0.0.74", - "resolved": "https://registry.npmjs.org/@code0-tech/tucana/-/tucana-0.0.74.tgz", - "integrity": "sha512-AgCl812Sn/MyY1+cRG0pdN0Uj7z93hDa4E9bclaf9hS5wc1PrlQSutDgePvJFR/VBWoeYQczne7MNP7RDu9NRg==", + "version": "0.0.80", + "resolved": "https://registry.npmjs.org/@code0-tech/tucana/-/tucana-0.0.80.tgz", + "integrity": "sha512-ocpG/Mg3F5HoVrx8UERwton+3dsFocj50HLPRIbb6XYnqD/wv6Pj2w9rjEnx9CHnIWUkayVgHCLrr4zmKa/cBw==", "dev": true, "license": "MIT" }, diff --git a/ts/package.json b/ts/package.json index 70dd763..96b88d6 100644 --- a/ts/package.json +++ b/ts/package.json @@ -39,7 +39,7 @@ "devDependencies": { "@emnapi/core": "^1.11.1", "@emnapi/runtime": "^1.11.1", - "@code0-tech/tucana": "0.0.74", + "@code0-tech/tucana": "0.0.80", "@grpc/grpc-js": "^1.14.3", "@protobuf-ts/grpc-backend": "^2.11.1", "@protobuf-ts/grpc-transport": "^2.11.1", @@ -57,7 +57,7 @@ "vitest": "^4.1.2" }, "peerDependencies": { - "@code0-tech/tucana": "0.0.74", + "@code0-tech/tucana": "0.0.80", "@grpc/grpc-js": "^1.14.3", "@protobuf-ts/grpc-backend": "^2.11.1", "@protobuf-ts/grpc-transport": "^2.11.1", diff --git a/ts/src/action.ts b/ts/src/action.ts index 46e66eb..4a82571 100644 --- a/ts/src/action.ts +++ b/ts/src/action.ts @@ -1,7 +1,10 @@ import {EventEmitter} from "node:events"; +import {randomUUID} from "node:crypto"; import type {GrpcOptions, GrpcTransport} from "@protobuf-ts/grpc-transport"; -import {ActionTransferRequest, type ActionTransferResponse} from "@code0-tech/tucana/aquila"; -import {constructValue, type PlainValue} from "@code0-tech/tucana/helpers"; +import {ActionTransferRequest, type ActionNodeSubFlowValue, type ActionTransferResponse} from "@code0-tech/tucana/aquila"; +import {constructValue, toAllowedValue, type PlainValue} from "@code0-tech/tucana/helpers"; +import type {Value, Error as ProtoError} from "@code0-tech/tucana/shared"; +import {RuntimeError} from "./types"; import type {DuplexStreamingCall} from "@protobuf-ts/runtime-rpc"; import type {FunctionClass} from "./models/function.model"; import type {RuntimeFunctionClass} from "./models/runtime_function.model"; @@ -18,6 +21,7 @@ import {CodeZeroEvent, type CodeZeroEventMap} from "./events"; import {createConnection} from "./internal/connection"; import {buildModule} from "./internal/module-builder"; import {ConfigManager} from "./manager/config-manager"; +import {FlowManager} from "./manager/FlowManager"; import {FunctionManager} from "./manager/FunctionManager"; import {RuntimeFunctionManager} from "./manager/RuntimeFunctionManager"; import {DataTypeManager} from "./manager/DataTypeManager"; @@ -41,8 +45,16 @@ export class Action extends EventEmitter { private _transport?: GrpcTransport; private _stream?: DuplexStreamingCall; private readonly _actions = new Map(actions.map(a => [a.packetType, a.handle])); + // Pending sub flow / flow execution requests awaiting a response, keyed by + // execution identifier. A queue is used because a sub flow can be executed + // repeatedly under the same execution identifier; responses are matched FIFO. + private readonly _pendingExecutions = new Map void; + reject: (error: unknown) => void; + }[]>(); readonly configs = new ConfigManager(); + readonly flows = new FlowManager(); readonly functions = new FunctionManager(); readonly runtimeFunctions = new RuntimeFunctionManager(); readonly dataTypes = new DataTypeManager(); @@ -130,6 +142,88 @@ export class Action extends EventEmitter { this.emit(CodeZeroEvent.streamMessageSent, request); } + /** + * Execute the sub flow referenced by a {@link ActionNodeSubFlowValue} parameter + * with the given parameters and resolve with its result. May be called + * repeatedly for the same sub flow (e.g. once per iteration). + */ + async executeSubFlow(subFlow: ActionNodeSubFlowValue, ...params: PlainValue[]): Promise { + if (!this._stream) throw new Error("Not connected. Call connect() first."); + const {executionIdentifier} = subFlow; + const result = this._awaitExecutionResponse(executionIdentifier); + const request = ActionTransferRequest.create({ + data: { + oneofKind: "subFlowExecution", + subFlowExecution: { + executionIdentifier, + parameters: params.map(p => constructValue(p ?? null)), + }, + }, + }); + await this._stream.requests.send(request); + this.emit(CodeZeroEvent.streamMessageSent, request); + return result; + } + + /** + * Execute one of the action's own flows by id and resolve with its result. + */ + async executeFlow(flowId: string | bigint, payload?: PlainValue): Promise { + if (!this._stream) throw new Error("Not connected. Call connect() first."); + const executionIdentifier = randomUUID(); + const result = this._awaitExecutionResponse(executionIdentifier); + const request = ActionTransferRequest.create({ + data: { + oneofKind: "flowExecution", + flowExecution: { + executionIdentifier, + flowId: String(flowId), + payload: constructValue(payload ?? null), + }, + }, + }); + await this._stream.requests.send(request); + this.emit(CodeZeroEvent.streamMessageSent, request); + return result; + } + + private _awaitExecutionResponse(executionIdentifier: string): Promise { + return new Promise((resolve, reject) => { + const queue = this._pendingExecutions.get(executionIdentifier) ?? []; + queue.push({resolve, reject}); + this._pendingExecutions.set(executionIdentifier, queue); + }); + } + + /** + * Resolve or reject a pending sub flow / flow execution request. Invoked by + * the response handlers when Aquila reports an execution's outcome. + */ + resolveExecutionResponse( + executionIdentifier: string, + result: + | {oneofKind: "success"; success: Value} + | {oneofKind: "failure"; failure: ProtoError} + | {oneofKind: undefined}, + ): void { + const queue = this._pendingExecutions.get(executionIdentifier); + const pending = queue?.shift(); + if (queue && queue.length === 0) this._pendingExecutions.delete(executionIdentifier); + if (!pending) { + this.emit(CodeZeroEvent.error, new Error( + `Received execution response for unknown execution identifier: ${executionIdentifier}`, + )); + return; + } + if (result.oneofKind === "success") { + pending.resolve(toAllowedValue(result.success)); + } else if (result.oneofKind === "failure") { + pending.reject(new RuntimeError(result.failure.code, result.failure.message)); + } else { + pending.reject(new Error("Received execution response with no result")); + } + } + async connect(authToken: string, aquilaUrl?: string, grpcOptions?: GrpcOptions) { if (isExportMode()) return; const url = aquilaUrl ?? this._aquilaUrl; diff --git a/ts/src/actions/Execution.ts b/ts/src/actions/Execution.ts index 7d79bfc..740da1d 100644 --- a/ts/src/actions/Execution.ts +++ b/ts/src/actions/Execution.ts @@ -1,7 +1,7 @@ import {constructValue, PlainValue, toAllowedValue} from "@code0-tech/tucana/helpers"; import {ActionExecutionRequest, ActionExecutionResponse, ActionTransferRequest} from "@code0-tech/tucana/aquila"; import {NodeExecutionResult, Error as ProtoError} from "@code0-tech/tucana/shared"; -import {FunctionContext, RuntimeError} from "../types"; +import {FunctionContext, RuntimeError, SubFlow} from "../types"; import {RuntimeFunctionProps} from "../models/runtime_function.model"; import {CodeZeroEvent} from "../events"; import type {Action} from "../action"; @@ -12,10 +12,15 @@ function nowMicros(): bigint { return BigInt(Math.floor((performance.timeOrigin + performance.now()) * 1000)); } -function buildParams(execution: ActionExecutionRequest, func: RuntimeFunctionProps): (PlainValue | undefined)[] { - return (func.parameters || []).map(param => { - const field = execution.parameters?.fields?.[param.runtimeName]; - return field ? toAllowedValue(field) : undefined; +function buildParams(action: Action, execution: ActionExecutionRequest, func: RuntimeFunctionProps): (PlainValue | SubFlow | undefined)[] { + return (func.parameters || []).map((param, index) => { + const field = execution.parameters?.[index]; + if (field?.value.oneofKind === "literalValue") return toAllowedValue(field.value.literalValue); + if (field?.value.oneofKind === "subFlow") { + const subFlow = field.value.subFlow; + return (...args: PlainValue[]) => action.executeSubFlow(subFlow, ...args); + } + return undefined; }); } @@ -28,7 +33,7 @@ export function handle(action: Action, execution: ActionExecutionRequest): void return; } - const params = buildParams(execution, func); + const params = buildParams(action, execution, func); const conf = action.configs.get(execution.projectId) ?? { projectId: 0n, @@ -40,6 +45,7 @@ export function handle(action: Action, execution: ActionExecutionRequest): void projectId: execution.projectId, executionId: execution.executionIdentifier, matchedConfig: conf, + executeFlow: (flowId, payload) => action.executeFlow(flowId, payload), }; const startedAt = nowMicros(); @@ -59,7 +65,7 @@ export function handle(action: Action, execution: ActionExecutionRequest): void startedAt, finishedAt, parameterResults: [], - result: {oneofKind: "success", success: constructValue(value)}, + result: {oneofKind: "success", success: constructValue(value ?? null)}, }), }), }, diff --git a/ts/src/actions/FlowExecution.ts b/ts/src/actions/FlowExecution.ts new file mode 100644 index 0000000..7970f47 --- /dev/null +++ b/ts/src/actions/FlowExecution.ts @@ -0,0 +1,10 @@ +import type {ActionFlowExecutionResponse} from "@code0-tech/tucana/aquila"; +import {CodeZeroEvent} from "../events"; +import type {Action} from "../action"; + +export const packetType = "flowExecutionResponse"; + +export function handle(action: Action, response: ActionFlowExecutionResponse): void { + action.emit(CodeZeroEvent.flowExecutionResponseReceived, response); + action.resolveExecutionResponse(response.executionIdentifier, response.result); +} diff --git a/ts/src/actions/FlowUpdate.ts b/ts/src/actions/FlowUpdate.ts new file mode 100644 index 0000000..4632754 --- /dev/null +++ b/ts/src/actions/FlowUpdate.ts @@ -0,0 +1,17 @@ +import type {ActionFlowUpdate} from "@code0-tech/tucana/aquila"; +import {CodeZeroEvent} from "../events"; +import type {Action} from "../action"; + +export const packetType = "flowUpdate"; + +export function handle(action: Action, update: ActionFlowUpdate): void { + if (update.data.oneofKind === "updatedFlow") { + const flow = update.data.updatedFlow; + action.flows.set(flow.flowId, flow); + action.emit(CodeZeroEvent.flowUpdated, flow); + } else if (update.data.oneofKind === "deletedFlow") { + const flowId = update.data.deletedFlow; + action.flows.delete(flowId); + action.emit(CodeZeroEvent.flowDeleted, flowId); + } +} diff --git a/ts/src/actions/SubFlowExecution.ts b/ts/src/actions/SubFlowExecution.ts new file mode 100644 index 0000000..bc06646 --- /dev/null +++ b/ts/src/actions/SubFlowExecution.ts @@ -0,0 +1,10 @@ +import type {ActionSubFlowExecutionResponse} from "@code0-tech/tucana/aquila"; +import {CodeZeroEvent} from "../events"; +import type {Action} from "../action"; + +export const packetType = "subFlowExecutionResponse"; + +export function handle(action: Action, response: ActionSubFlowExecutionResponse): void { + action.emit(CodeZeroEvent.subFlowExecutionResponseReceived, response); + action.resolveExecutionResponse(response.executionIdentifier, response.result); +} diff --git a/ts/src/actions/index.ts b/ts/src/actions/index.ts index d531f3b..162f6c0 100644 --- a/ts/src/actions/index.ts +++ b/ts/src/actions/index.ts @@ -1,6 +1,9 @@ import type {Action} from "../action"; import * as ModuleConfigurations from "./ModuleConfigurations"; import * as Execution from "./Execution"; +import * as SubFlowExecution from "./SubFlowExecution"; +import * as FlowExecution from "./FlowExecution"; +import * as FlowUpdate from "./FlowUpdate"; export interface ActionHandler { packetType: string; @@ -10,4 +13,7 @@ export interface ActionHandler { export const actions: ActionHandler[] = [ ModuleConfigurations, Execution, + SubFlowExecution, + FlowExecution, + FlowUpdate, ]; diff --git a/ts/src/events.ts b/ts/src/events.ts index 8c06e82..eee64e2 100644 --- a/ts/src/events.ts +++ b/ts/src/events.ts @@ -1,4 +1,11 @@ -import type {ActionExecutionRequest, ActionTransferRequest, ActionTransferResponse} from "@code0-tech/tucana/aquila"; +import type { + ActionExecutionRequest, + ActionFlow, + ActionFlowExecutionResponse, + ActionSubFlowExecutionResponse, + ActionTransferRequest, + ActionTransferResponse +} from "@code0-tech/tucana/aquila"; import type {ModuleConfigurations} from "@code0-tech/tucana/shared"; import type {Action} from "./action.ts"; @@ -9,6 +16,10 @@ export enum CodeZeroEvent { streamMessageSent = "streamMessageSent", moduleUpdated = "moduleUpdated", executionRequestReceived = "executionRequestReceived", + subFlowExecutionResponseReceived = "subFlowExecutionResponseReceived", + flowExecutionResponseReceived = "flowExecutionResponseReceived", + flowUpdated = "flowUpdated", + flowDeleted = "flowDeleted", } export interface CodeZeroEventMap { @@ -18,5 +29,9 @@ export interface CodeZeroEventMap { [CodeZeroEvent.streamMessageSent]: [ActionTransferRequest] [CodeZeroEvent.moduleUpdated]: [ModuleConfigurations] [CodeZeroEvent.executionRequestReceived]: [ActionExecutionRequest] + [CodeZeroEvent.subFlowExecutionResponseReceived]: [ActionSubFlowExecutionResponse] + [CodeZeroEvent.flowExecutionResponseReceived]: [ActionFlowExecutionResponse] + [CodeZeroEvent.flowUpdated]: [ActionFlow] + [CodeZeroEvent.flowDeleted]: [bigint] [key: string]: unknown[] } diff --git a/ts/src/index.ts b/ts/src/index.ts index 42e524f..4545b74 100644 --- a/ts/src/index.ts +++ b/ts/src/index.ts @@ -9,6 +9,7 @@ export * from "./decorators/runtime_function.dec" export * from "./definitions" export * from "./manager/BaseManager" export * from "./manager/config-manager" +export * from "./manager/FlowManager" export * from "./manager/FunctionManager" export * from "./manager/RuntimeFunctionManager" export * from "./manager/DataTypeManager" diff --git a/ts/src/internal/connection.ts b/ts/src/internal/connection.ts index aa93f33..bca0825 100644 --- a/ts/src/internal/connection.ts +++ b/ts/src/internal/connection.ts @@ -1,9 +1,14 @@ import {GrpcOptions, GrpcTransport} from "@protobuf-ts/grpc-transport"; -import {ActionTransferRequest, ActionTransferResponse, ActionTransferServiceClient} from "@code0-tech/tucana/aquila"; +import { + ActionLogon_ScalingOption, + ActionTransferRequest, + ActionTransferResponse, + ActionTransferServiceClient +} from "@code0-tech/tucana/aquila"; import type {Module} from "@code0-tech/tucana/shared"; import {ChannelCredentials} from "@grpc/grpc-js"; -import {type RpcOptions} from "@protobuf-ts/runtime-rpc"; import type {DuplexStreamingCall} from "@protobuf-ts/runtime-rpc"; +import {type RpcOptions} from "@protobuf-ts/runtime-rpc"; export interface Connection { transport: GrpcTransport; @@ -29,7 +34,7 @@ export async function createConnection( await stream.requests.send(ActionTransferRequest.create({ data: { oneofKind: "logon", - logon: {module}, + logon: {module, scalingOption: ActionLogon_ScalingOption.SPLIT}, }, })); diff --git a/ts/src/internal/module-builder.ts b/ts/src/internal/module-builder.ts index 31bad86..0d988a3 100644 --- a/ts/src/internal/module-builder.ts +++ b/ts/src/internal/module-builder.ts @@ -43,6 +43,7 @@ export function buildModule(data: ModuleBuildData): Module { documentation: data.documentation, name: data.name, description: [], + definitionSource: "action", configurations: data.configurationDefinitions.map(def => ({ identifier: def.identifier, name: def.name ?? [], diff --git a/ts/src/manager/FlowManager.ts b/ts/src/manager/FlowManager.ts new file mode 100644 index 0000000..7729ce5 --- /dev/null +++ b/ts/src/manager/FlowManager.ts @@ -0,0 +1,8 @@ +import type {ActionFlow} from "@code0-tech/tucana/aquila"; +import {BaseManager} from "./BaseManager"; + +/** + * Registry of the action's own flows, kept in sync via ActionFlowUpdate messages. + * Keyed by flow id. Use {@link Action.executeFlow} to run one of these flows. + */ +export class FlowManager extends BaseManager {} diff --git a/ts/src/types.ts b/ts/src/types.ts index 2f2453f..76d2ce5 100644 --- a/ts/src/types.ts +++ b/ts/src/types.ts @@ -1,18 +1,23 @@ import {FlowTypeSetting_UniquenessScope, RuntimeFlowTypeSetting_UniquenessScope} from "@code0-tech/tucana/shared"; +import type {ActionNodeSubFlowValue} from "@code0-tech/tucana/aquila"; import {PlainValue} from "@code0-tech/tucana/helpers"; import 'reflect-metadata'; export {FlowTypeSetting_UniquenessScope, RuntimeFlowTypeSetting_UniquenessScope}; +export type {ActionNodeSubFlowValue, PlainValue}; export interface Translation { code: "en-US" | "de-DE" | string, content: string } +export type SubFlow = (...args: PlainValue[]) => Promise; + export interface FunctionContext { projectId: number | bigint, executionId: string, - matchedConfig: ProjectConfiguration + matchedConfig: ProjectConfiguration, + executeFlow: (flowId: string | bigint, payload?: PlainValue) => Promise, } export interface ProjectConfiguration {