shishantbiswas/bknd
1
0
Fork 0
mirror of https://github.com/bknd-io/bknd/ synced 2026-08-03 08:36:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

merged origin/release/0.12

Browse Source
This commit is contained in:
dswbx
2025-04-11 13:10:25 +02:00
parent a55eca75d0 e5d6ed536c
commit 95a8289afa
79 changed files with 177 additions and 544 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/flows/AppFlows.ts
-1
View File
@@ -25,7 +25,6 @@ export class AppFlows extends Module<typeof flowsConfigSchema> {
}
override async build() {
//console.log("building flows", this.config);
const flows = transformObject(this.config.flows, (flowConfig, name) => {
return Flow.fromObject(name, flowConfig as any, TASKS);
});
app/src/flows/flows/Execution.ts
+2 -2
View File
@@ -2,6 +2,7 @@ import { Event, EventManager, type ListenerHandler } from "core/events";
import type { EmitsEvents } from "core/events";
import type { Task, TaskResult } from "../tasks/Task";
import type { Flow } from "./Flow";
import { $console } from "core";
export type TaskLog = TaskResult & {
task: Task;
@@ -185,10 +186,9 @@ export class Execution implements EmitsEvents {
await Promise.all(promises);
return this.run();
} catch (e) {
console.log("RuntimeExecutor: error", e);
$console.error("RuntimeExecutor: error", e);
// for now just throw
// biome-ignore lint/complexity/noUselessCatch: @todo: add error task on flow
throw e;
}
}
app/src/flows/flows/Flow.ts
+2 -12
View File
@@ -5,6 +5,7 @@ import { Condition, TaskConnection } from "../tasks/TaskConnection";
import { Execution } from "./Execution";
import { FlowTaskConnector } from "./FlowTaskConnector";
import { Trigger } from "./triggers/Trigger";
import { $console } from "core";
type Jsoned<T extends { toJSON: () => object }> = ReturnType<T["toJSON"]>;
@@ -53,8 +54,6 @@ export class Flow {
}
getSequence(sequence: Task[][] = []): Task[][] {
//console.log("queue", queue.map((step) => step.map((t) => t.name)));
// start task
if (sequence.length === 0) {
sequence.push([this.startTask]);
@@ -69,7 +68,6 @@ export class Flow {
// check if task already in one of queue steps
// this is when we have a circle back
if (sequence.some((step) => step.includes(outTask))) {
//console.log("Task already in queue", outTask.name);
return;
}
nextStep.push(outTask);
@@ -110,14 +108,6 @@ export class Flow {
return this;
}
/*getResponse() {
if (!this.respondingTask) {
return;
}
return this.respondingTask.log.output;
}*/
// @todo: check for existence
addConnection(connection: TaskConnection) {
// check if connection already exists
@@ -179,7 +169,7 @@ export class Flow {
// @ts-ignore
return new cls(name, obj.params);
} catch (e: any) {
console.log("Error creating task", name, obj.type, obj, taskClass);
$console.error("Error creating task", name, obj.type, obj, taskClass);
throw new Error(`Error creating task ${obj.type}: ${e.message}`);
}
});
app/src/flows/flows/FlowTaskConnector.ts
-34
View File
@@ -31,37 +31,11 @@ export class FlowTaskConnector {
}
}
/*const targetDepth = this.task(target).getDepth();
console.log("depth", ownDepth, targetDepth);
// if target has a lower depth
if (targetDepth > 0 && ownDepth >= targetDepth) {
// check for unique out conditions
console.log(
"out conditions",
this.source.name,
this.getOutConnections().map((c) => [c.target.name, c.condition])
);
if (
this.getOutConnections().some(
(c) =>
c.condition[0] === condition[0] &&
c.condition[1] === condition[1]
)
) {
throw new Error(
"Task cannot be connected to a deeper task with the same condition"
);
}
}*/
this.flow.addConnection(new TaskConnection(this.source, target, { condition, max_retries }));
}
asOutputFor(target: Task, condition?: Condition) {
this.task(target).asInputFor(this.source, condition);
//new FlowTaskConnector(this.flow, target).asInputFor(this.source);
//this.flow.addConnection(new TaskConnection(target, this.source));
}
getNext() {
@@ -107,12 +81,4 @@ export class FlowTaskConnector {
getOutTasks(result?: TaskResult): Task[] {
return this.getOutConnections(result).map((c) => c.target);
}
/*getNextRunnableConnections() {
return this.getOutConnections().filter((c) => c.source.log.success);
}
getNextRunnableTasks() {
return this.getNextRunnableConnections().map((c) => c.target);
}*/
}
app/src/flows/flows/executors/RuntimeExecutor.ts
+2 -2
View File
@@ -1,4 +1,5 @@
import type { Task } from "../../tasks/Task";
import { $console } from "core";
export class RuntimeExecutor {
async run(
@@ -10,7 +11,6 @@ export class RuntimeExecutor {
return;
}
//const promises = tasks.map((t) => t.run());
const promises = tasks.map(async (t) => {
const result = await t.run();
onDone?.(t, result);
@@ -20,7 +20,7 @@ export class RuntimeExecutor {
try {
await Promise.all(promises);
} catch (e) {
console.log("RuntimeExecutor: error", e);
$console.error("RuntimeExecutor: error", e);
}
return this.run(nextTasks, onDone);
app/src/flows/flows/triggers/EventTrigger.ts
+2 -5
View File
@@ -1,6 +1,7 @@
import type { EventManager } from "core/events";
import type { Flow } from "../Flow";
import { Trigger } from "./Trigger";
import { $console } from "core";
import * as tbbox from "@sinclair/typebox";
const { Type } = tbbox;
@@ -23,17 +24,13 @@ export class EventTrigger extends Trigger<typeof EventTrigger.schema> {
emgr.on(
this.config.event,
async (event) => {
console.log("event", event);
/*if (!this.match(event)) {
return;
}*/
const execution = flow.createExecution();
this.executions.push(execution);
try {
await execution.start(event.params);
} catch (e) {
console.error(e);
$console.error(e);
}
},
this.config.mode,
app/src/flows/flows/triggers/HttpTrigger.ts
+5 -10
View File
@@ -12,14 +12,11 @@ export class HttpTrigger extends Trigger<typeof HttpTrigger.schema> {
static override schema = Type.Composite([
Trigger.schema,
Type.Object(
{
path: Type.String({ pattern: "^/.*$" }),
method: StringEnum(httpMethods, { default: "GET" }),
response_type: StringEnum(["json", "text", "html"], { default: "json" }),
},
//{ additionalProperties: false }
),
Type.Object({
path: Type.String({ pattern: "^/.*$" }),
method: StringEnum(httpMethods, { default: "GET" }),
response_type: StringEnum(["json", "text", "html"], { default: "json" }),
}),
]);
override async register(flow: Flow, hono: Hono<any>) {
@@ -45,7 +42,5 @@ export class HttpTrigger extends Trigger<typeof HttpTrigger.schema> {
execution.start(params);
return c.json({ success: true });
});
//console.log("--registered flow", flow.name, "on", method, this.config.path);
}
}
app/src/flows/flows/triggers/Trigger.ts
+3 -6
View File
@@ -10,12 +10,9 @@ export class Trigger<Schema extends typeof Trigger.schema = typeof Trigger.schem
type = "manual";
config: Static<Schema>;
static schema = Type.Object(
{
mode: StringEnum(["sync", "async"], { default: "async" }),
},
//{ additionalProperties: false }
);
static schema = Type.Object({
mode: StringEnum(["sync", "async"], { default: "async" }),
});
constructor(config?: Partial<Static<Schema>>) {
const schema = (this.constructor as typeof Trigger).schema;
app/src/flows/flows/triggers/index.ts
-1
View File
@@ -4,7 +4,6 @@ import { Trigger } from "./Trigger";
export { Trigger, EventTrigger, HttpTrigger };
//export type TriggerMapType = { [key: string]: { cls: typeof Trigger } };
export const TriggerMap = {
manual: { cls: Trigger },
event: { cls: EventTrigger },
app/src/flows/index.ts
+1 -5
View File
@@ -23,13 +23,9 @@ export {
} from "./flows/triggers";
import { Task } from "./tasks/Task";
export { type TaskResult, type TaskRenderProps } from "./tasks/Task";
export type { TaskResult, TaskRenderProps } from "./tasks/Task";
export { TaskConnection, Condition } from "./tasks/TaskConnection";
// test
//export { simpleFetch } from "./examples/simple-fetch";
//export type TaskMapType = { [key: string]: { cls: typeof Task<any> } };
export const TaskMap = {
fetch: { cls: FetchTask },
log: { cls: LogTask },
app/src/flows/tasks/Task.tsx
+1 -32
View File
@@ -14,10 +14,6 @@ export type TaskResult<Output = any> = {
params: any;
};
/*export type TaskRenderProps<T extends Task = Task> = NodeProps<{
task: T;
state: { i: number; isStartTask: boolean; isRespondingTask; event: ExecutionEvent | undefined };
}>;*/
export type TaskRenderProps<T extends Task = Task> = any;
export function dynamic<Type extends TSchema>(
@@ -94,21 +90,6 @@ export abstract class Task<Params extends TObject = TObject, Output = unknown> {
// @todo: string enums fail to validate
this._params = parse(schema, params || {});
/*const validator = new Validator(schema as any);
const _params = Default(schema, params || {});
const result = validator.validate(_params);
if (!result.valid) {
//console.log("---errors", result, { params, _params });
const error = result.errors[0]!;
throw new Error(
`Invalid params for task "${name}.${error.keyword}": "${
error.error
}". Params given: ${JSON.stringify(params)}`
);
}
this._params = _params as Static<Params>;*/
}
get params() {
@@ -127,11 +108,8 @@ export abstract class Task<Params extends TObject = TObject, Output = unknown> {
const newParams: any = {};
const renderer = new SimpleRenderer(inputs, { strictVariables: true, renderKeys: true });
//console.log("--resolveParams", params);
for (const [key, value] of Object.entries(params)) {
if (value && SimpleRenderer.hasMarkup(value)) {
//console.log("--- has markup", value);
try {
newParams[key] = await renderer.render(value as string);
} catch (e: any) {
@@ -151,29 +129,21 @@ export abstract class Task<Params extends TObject = TObject, Output = unknown> {
throw e;
}
continue;
} else {
//console.log("-- no markup", key, value);
}
newParams[key] = value;
}
//console.log("--beforeDecode", newParams);
const v = Value.Decode(schema, newParams);
//console.log("--afterDecode", v);
//process.exit();
return v;
return Value.Decode(schema, newParams);
}
private async cloneWithResolvedParams(_inputs: Map<string, any>) {
const inputs = Object.fromEntries(_inputs.entries());
//console.log("--clone:inputs", inputs, this.params);
const newParams = await Task.resolveParams(
(this.constructor as any).schema,
this._params,
inputs,
);
//console.log("--clone:newParams", this.name, newParams);
return this.clone(this.name, newParams as any);
}
@@ -201,7 +171,6 @@ export abstract class Task<Params extends TObject = TObject, Output = unknown> {
success = true;
} catch (e: any) {
success = false;
//status.output = undefined;
if (e instanceof BkndError) {
error = e.toJSON();
app/src/flows/tasks/TaskConnection.ts
-1
View File
@@ -80,7 +80,6 @@ export class Condition {
return result.success === false;
case "matches":
return get(result.output, this.path) === this.value;
//return this.value === output[this.path];
}
}
app/src/flows/tasks/presets/FetchTask.ts
-12
View File
@@ -15,8 +15,6 @@ export class FetchTask<Output extends Record<string, any>> extends Task<
url: Type.String({
pattern: "^(http|https)://",
}),
//method: Type.Optional(Type.Enum(FetchMethodsEnum)),
//method: Type.Optional(dynamic(Type.String({ enum: FetchMethods, default: "GET" }))),
method: Type.Optional(dynamic(StringEnum(FetchMethods, { default: "GET" }))),
headers: Type.Optional(
dynamic(
@@ -43,7 +41,6 @@ export class FetchTask<Output extends Record<string, any>> extends Task<
}
async execute() {
//console.log(`method: (${this.params.method})`);
if (!FetchMethods.includes(this.params.method ?? "GET")) {
throw this.error("Invalid method", {
given: this.params.method,
@@ -54,19 +51,12 @@ export class FetchTask<Output extends Record<string, any>> extends Task<
const body = this.getBody();
const headers = new Headers(this.params.headers?.map((h) => [h.key, h.value]));
/*console.log("[FETCH]", {
url: this.params.url,
method: this.params.method ?? "GET",
headers,
body
});*/
const result = await fetch(this.params.url, {
method: this.params.method ?? "GET",
headers,
body,
});
//console.log("fetch:response", result);
if (!result.ok) {
throw this.error("Failed to fetch", {
status: result.status,
@@ -75,8 +65,6 @@ export class FetchTask<Output extends Record<string, any>> extends Task<
}
const data = (await result.json()) as Output;
//console.log("fetch:response:data", data);
return data;
}
}
app/src/flows/tasks/presets/LogTask.ts
+2 -1
View File
@@ -1,4 +1,5 @@
import { Task } from "../Task";
import { $console } from "core";
import * as tbbox from "@sinclair/typebox";
const { Type } = tbbox;
@@ -11,7 +12,7 @@ export class LogTask extends Task<typeof LogTask.schema> {
async execute() {
await new Promise((resolve) => setTimeout(resolve, this.params.delay));
console.log(`[DONE] LogTask: ${this.name}`);
$console.log(`[DONE] LogTask: ${this.name}`);
return true;
}
}