@flowlab/core API 文档 本文档详细介绍了 @flowlab/core 工作流引擎核心库的公共 API。建议使用子路径导入来按需加载模块。 1. 核心类型 (@flowlab/core/types) 此模块定义了工作流引擎使用的核心接口和枚举。 枚举 (Enums) NodeStatus 表示工作流节点或整个工作流实例的状态。 PENDING: 等待执行 RUNNING: 正在执行 COMPLETED: 执行成功 FAILED: 执行失败 TIMEOUT: 执行超时 SKIPPED: 被跳过 (例如，条件不满足) CANCELLED: 已取消 COMPENSATING: 正在补偿 COMPENSATED: 已补偿 StepType 表示工作流中步骤的类型。 TASK: 普通任务节点 CONDITION: 条件分支 PARALLEL: 并行步骤 SUB_WORKFLOW: 子工作流 EVENT_TRIGGER: 事件触发器 EVENT_LISTENER: 事件监听器 接口 (Interfaces) IWorkflowContext 代表一个工作流实例的完整执行上下文。 workflowId: string: 工作流实例的唯一 ID。 workflowDefinitionId: string: 所属工作流定义的 ID。 input: Record: 工作流启动时的初始输入数据。 output?: Record: 工作流执行成功后的最终输出数据。 status: NodeStatus: 工作流实例的当前整体状态。 variables: Record: 工作流执行期间可读写的变量存储对象。节点可以通过输入/输出映射与此对象交互。 startTime?: Date: 工作流实例开始执行的时间。 endTime?: Date: 工作流实例结束执行的时间。 tenantId?: string: (可选) 租户 ID，用于多租户隔离。 userId?: string: (可选) 触发工作流的用户 ID。 userRole?: string: (可选) 触发工作流的用户角色，用于权限检查。 history: IExecutionRecord[]: 按时间顺序记录的每个步骤的执行详情。 error?: Error: 如果工作流失败，记录遇到的错误对象。 [key: string]: any: 允许附加其他自定义属性。 INodeContext 代表一个任务节点在某次执行尝试中的上下文。 nodeId: string: 当前执行节点的类型 ID (来自 INode.id)。 stepId: string: 当前执行步骤在工作流定义中的 ID (来自 IStepConfig.id)。 input: Record: 传递给当前节点执行的输入数据 (已解析输入映射)。 output?: Record: 节点执行成功后产生的输出数据。 status: NodeStatus: 当前节点本次执行尝试的状态。 startTime?: Date: 本次节点执行尝试的开始时间。 endTime?: Date: 本次节点执行尝试的结束时间。 retries: number: 当前是第几次尝试执行 (0-based index)。 error?: Error: 如果本次尝试失败或超时，记录遇到的错误对象。 logs: string[]: 节点在本次执行期间记录的日志信息。 workflowContext: IWorkflowContext: 对所属工作流实例上下文 (IWorkflowContext) 的引用。 [key: string]: any: 允许附加其他自定义属性。 INodeOutput INode 的 execute 方法需要返回的执行结果结构。 status: NodeStatus: 节点执行后的状态 (COMPLETED, FAILED, SKIPPED)。注意： 不应返回 RUNNING 或 PENDING。 output?: Record: (可选) 节点执行成功时产生的输出数据。 error?: Error: (可选) 节点执行失败时关联的错误对象。 INode 定义自定义任务节点必须实现的接口。 id: string: 节点的唯一类型标识符。NodeRegistry 使用此 ID 来查找节点实现。 execute(context: INodeContext): Promise: (必需) 节点的核心执行逻辑。接收节点上下文 INodeContext，必须返回一个包含执行结果 INodeOutput 的 Promise。 compensate?(context: INodeContext): Promise: (可选) 补偿逻辑。当工作流需要回滚或补偿此节点之前的操作时，执行器会调用此方法。通常在节点执行产生副作用（如数据库写入、API 调用）时实现。 validateInput?(input: Record): boolean: (可选) 输入验证逻辑。在 execute 方法执行前，执行器可以调用此方法验证 context.input 是否有效。返回 false 将导致节点执行失败。 requiredRoles?: string[]: (可选) 定义执行此节点所需的角色列表。执行器会根据 IWorkflowContext.userRole 进行检查。 IStepConfig 定义工作流中一个步骤的配置。 id: string: 步骤在工作流内的唯一 ID。 type: StepType: 步骤的类型。 nodeId?: string: (当 type 为 TASK 时必需) 关联的 INode 的类型 ID。 name?: string: (可选) 步骤的可读名称，用于日志和监控。 input?: Record: (可选) 定义节点的输入。可以是静态值，也可以是从 workflow.variables 或 workflow.input 映射的路径字符串 (例如 {'nodeVar': 'workflow.variables.someData'})。 outputMapping?: Record: (可选) 定义如何将节点成功执行后的 output 映射回 workflow.variables。键是目标 workflow.variables 路径，值是源 nodeOutput 路径 (例如 {'workflow.variables.result': 'nodeOutput.data'})。 nextStepId?: string: (可选) 对于顺序步骤，指定下一个要执行的步骤的 ID。 timeout?: number: (可选) 此步骤执行的超时时间 (毫秒)。超时会导致步骤状态变为 TIMEOUT。 maxRetries?: number: (可选) 当步骤执行失败时，最大允许的重试次数 (默认 0，表示不重试)。总执行次数为 maxRetries + 1。 retryDelay?: number: (可选) 每次重试之间的延迟时间 (毫秒，默认 0)。 condition?: (context: IWorkflowContext) => boolean | string: (当 type 为 CONDITION 时必需) 条件评估函数。接收工作流上下文，返回 boolean (决定是否走 nextStepId 或跳过) 或 string (作为 branches 的键)。 branches?: { [key: string]: string }: (当 type 为 CONDITION 且 condition 返回 string 时使用) 定义条件分支。键是 condition 函数可能返回的字符串，值是对应的下一个步骤 ID。 parallelSteps?: IStepConfig[]: (当 type 为 PARALLEL 时必需) 定义需要并行执行的步骤配置数组。 subWorkflowId?: string: (当 type 为 SUB_WORKFLOW 时必需) 要调用的子工作流的定义 ID。 subWorkflowInput?: Record: (可选，用于 SUB_WORKFLOW) 定义传递给子工作流的输入，支持映射。 event?: string: (当 type 为 EVENT_TRIGGER 或 EVENT_LISTENER 时必需) 关联的事件名称。 sla?: number: (可选) 服务水平协议 (SLA) 阈值 (毫秒)。用于监控步骤执行是否超时或超过预期时间。 compensateOnFailure?: boolean | string: (可选) 定义此步骤失败时是否需要触发补偿。true 表示需要回滚之前的步骤；string 表示跳转到指定的补偿步骤 ID (功能待定)。 requiredRoles?: string[]: (可选) 执行此步骤所需的用户角色列表。 IExecutionRecord 记录在 IWorkflowContext.history 中的单次步骤执行详情。 stepId: string: 执行的步骤 ID。 nodeId?: string: 执行的节点类型 ID (如果是 TASK 步骤)。 status: NodeStatus: 本次执行尝试的结果状态。 startTime: Date: 本次尝试的开始时间。 endTime?: Date: 本次尝试的结束时间。 input: Record: 本次尝试使用的输入数据。 output?: Record: 本次尝试产生的输出数据 (如果成功)。 error?: string: 本次尝试遇到的错误消息 (如果失败或超时)。 logs: string[]: 本次尝试期间节点记录的日志。 attempt: number: 本次尝试是第几次 (0-based index)。 IEventPayload 事件管理器中传递的事件数据结构。 eventName: string: 事件名称。 data?: Record: 事件关联的数据。 timestamp: Date: 事件发生的时间戳。 source?: string: (可选) 事件来源标识 (例如触发事件的工作流实例 ID)。 2. 核心 API (@flowlab/core/core) 此模块包含构建、管理和执行工作流的核心类和函数。 WorkflowDefinition 用于定义工作流结构和步骤的类。 constructor(id: string, name?: string, description?: string): 创建一个新的工作流定义。 id: 工作流定义的唯一 ID。 name: (可选) 工作流的可读名称。 description: (可选) 工作流的描述。 addStep(config: Omit & { nodeId: string }): this: 添加一个顺序执行的任务节点步骤。 addConditionStep(config: Omit & { condition: Function; branches: object }): this: 添加一个条件分支步骤。 addParallelStep(config: Omit & { parallelSteps: IStepConfig[] }): this: 添加一个并行执行步骤。 addSubWorkflowStep(config: Omit & { subWorkflowId: string }): this: 添加一个子工作流步骤。 addEventTriggerStep(config: Omit & { event: string }): this: 添加一个事件触发步骤。 addEventListenerStep(config: Omit & { event: string }): this: 添加一个事件监听步骤。 setStartStep(stepId: string): this: 显式设置工作流的起始步骤 ID。如果未调用，则第一个添加的非监听器步骤将作为起始步骤。 getStartStepId(): string: 获取起始步骤 ID。 getStep(stepId: string): Step | undefined: 根据 ID 获取步骤实例。 getSteps(): Map: 获取所有步骤的 Map。 validate(): boolean: 验证工作流定义的有效性（例如，检查起始步骤、步骤链接等）。执行器在运行前会调用此方法。 Step 代表工作流中的一个步骤实例。通常由 WorkflowDefinition 内部创建和管理。其属性主要来自 IStepConfig。 BaseNode 所有自定义任务节点的抽象基类。提供了 INode 接口的基础实现（compensate 和 validateInput 的默认行为）。自定义节点应继承此类并实现 id 属性和 execute 方法。 NodeRegistry 用于注册和查找 INode 实现的类。 register(node: INode): void: 注册一个节点实例。如果 ID 已存在，会覆盖旧的并打印警告。 get(nodeId: string): INode | undefined: 根据节点类型 ID 获取已注册的节点实例。 has(nodeId: string): boolean: 检查指定 ID 的节点是否已注册。 listNodeIds(): string[]: 返回所有已注册节点的 ID 列表。 nodeRegistry (导出实例) 一个预创建的 NodeRegistry 单例实例，可以直接导入和使用。 import { nodeRegistry } from '@flowlab/core/core'; nodeRegistry.register(new MyCustomNode()); createWorkflowContext(...) 工厂函数，用于创建新的 IWorkflowContext 实例。 function createWorkflowContext( definitionId: string, input: Record, tenantId?: string, userId?: string, userRole?: string ): IWorkflowContext; createNodeContext(...) 工厂函数，用于创建新的 INodeContext 实例（主要由 WorkflowExecutor 内部使用）。 function createNodeContext( workflowContext: IWorkflowContext, stepId: string, nodeId: string | undefined, input: Record ): INodeContext; WorkflowExecutor 负责执行 WorkflowDefinition 实例的类。 constructor(workflowDefinition: WorkflowDefinition, context: IWorkflowContext, options: IWorkflowExecutorOptions): 创建执行器实例。 workflowDefinition: 要执行的工作流定义。构造函数会先调用其 validate() 方法。 context: 使用 createWorkflowContext 创建的工作流实例上下文。 options: 执行器选项 IWorkflowExecutorOptions。 nodeRegistry: NodeRegistry: (必需) 包含所有可用节点的注册表实例。 maxLoopIterations?: number: (可选) 防止无限循环的最大步骤执行次数 (默认 1000)。 logger?: ILogger: (可选, 待集成) 用于记录执行日志的记录器实例。 eventManager?: IEventManager: (可选, 待集成) 用于触发事件的事件管理器实例。 scheduler?: IScheduler: (可选, 待集成) 用于调度任务的调度器实例。 workflowStore?: IWorkflowStore: (可选, 待集成) 用于加载子工作流定义的存储实例。 run(): Promise: 启动工作流执行。这是一个异步方法，返回一个 Promise，该 Promise 在工作流执行完成（成功或失败）时解析为最终的 IWorkflowContext。 注意事项 (WorkflowExecutor): run() 方法会处理步骤的顺序执行、条件判断、并行（并发 Promise）、子工作流调用、超时、重试和基本补偿逻辑。 执行过程中会更新传入的 IWorkflowContext 对象的状态、变量、历史记录等。 错误处理：run() 方法会捕获执行期间的内部错误，并将工作流状态设置为 FAILED，错误信息记录在 context.error 中。如果节点执行失败且达到重试次数，也会导致工作流失败。 补偿：当步骤配置了 compensateOnFailure: true 且执行失败时，执行器会尝试调用失败节点（如果有 compensate 方法）和之前成功执行且需要补偿的节点的 compensate 方法（按补偿栈逆序）。 状态管理：IWorkflowContext 是可变的，执行器会直接修改它。如果需要持久化状态，需要在外部处理。 并行限制：当前的并行步骤实现使用 Promise.all，是并发执行，受 Node.js 事件循环限制，并非真正的多线程/多进程并行。 3. 错误类型 (@flowlab/core/error) 定义了库中使用的特定错误类，都继承自 WorkflowError。 WorkflowError: 所有工作流相关错误的基类。 NodeExecutionError: 节点执行期间发生的错误。包含 nodeId 和 stepId 属性。 TimeoutError: 节点执行超时的特定错误，继承自 NodeExecutionError。 AuthorizationError: 因权限不足导致执行失败的错误。 ConfigurationError: 工作流定义或节点配置无效时抛出的错误。 4. 事件管理 (@flowlab/core/event) 定义了事件触发和监听的接口及基本实现。 IEventManager 事件管理器接口。 emit(eventName: string, data?: Record, triggerContext?: IWorkflowContext): Promise: 触发一个事件。 bindWorkflowToEvent(eventName: string, workflowDefinitionId: string, options?: any): void: 将一个工作流定义绑定到特定事件。当该事件被 emit 时，应启动对应的工作流（具体启动逻辑需实现）。 on(eventName: string, listener: (payload: IEventPayload) => void): void: 注册一个通用的事件监听器。 off(eventName: string, listener: (payload: IEventPayload) => void): void: 移除事件监听器。 InMemoryEventManager 一个基于内存的简单 IEventManager 实现，用于测试或单体应用。注意： 它本身不包含启动工作流的逻辑，需要在使用时注入相应的启动函数。 5. 调度器 (@flowlab/core/scheduler) 定义了任务调度（延迟执行、分布式执行）的接口和实现。 IScheduler 任务调度器通用接口。 scheduleWorkflow(definitionId: string, input: Record, options?: IScheduleOptions): Promise: 调度一个工作流的执行。 scheduleNode(nodeId: string, context: Partial & { workflowContext: Pick }, options?: IScheduleOptions): Promise: 调度单个任务节点的执行（用于分布式 Worker）。 cancel(taskId: string): Promise: 尝试取消一个已调度的任务。 IScheduleOptions 调度选项接口。 delay?: number: 延迟执行时间 (毫秒)。 priority?: number: 任务优先级。 tenantId?: string, userId?: string: 覆盖上下文中的租户/用户信息。 parentWorkflowId?: string: 用于跟踪的父工作流 ID。 queueName?: string: 目标消息队列名称。 kubernetesJobConfig?: any: Kubernetes Job 相关配置。 LocalScheduler 一个本地调度器的简单实现。注意： 它实际上是立即执行（或使用 setTimeout 模拟延迟），并不做真正的后台调度。 IDistributedScheduler 分布式调度器接口，继承自 IScheduler。 DistributedSchedulerPlaceholder 一个分布式调度器的占位符实现。演示了如何将任务信息（工作流或节点）发送到假想的外部系统（如 RabbitMQ, Kafka, Kubernetes）。注意： 需要根据实际使用的分布式系统进行具体实现。 6. 监控与日志 (@flowlab/core/monitoring) 定义了日志记录接口和基本实现。 ILogger 日志记录器接口。 debug(message: string, meta?: any): void info(message: string, meta?: any): void warn(message: string, meta?: any): void error(message: string, error?: Error, meta?: any): void ConsoleLogger 一个将日志输出到控制台的简单 ILogger 实现。 7. 主入口 (@flowlab/core) 包的主入口点 (import { ... } from '@flowlab/core';)。 它重新导出了来自各个子模块的核心和常用 API，例如 WorkflowDefinition, WorkflowExecutor, BaseNode, nodeRegistry, NodeStatus, StepType, WorkflowError, ConsoleLogger 等。 建议: 为了更好的代码组织和按需加载，推荐优先使用子路径导入 (e.g., import { WorkflowDefinition } from '@flowlab/core/core';)。 通用注意事项: 异步性: 工作流执行和节点执行都是异步的 (async/await, Promise)。 状态管理: @flowlab/core 本身不负责持久化工作流实例的状态。IWorkflowContext 对象在内存中被修改。如果需要持久化、恢复或在分布式环境中使用，需要在外部实现状态的保存和加载逻辑。 错误处理: 库提供了基本的错误类型和重试机制。复杂的错误处理策略（如业务异常处理、特定补偿流程）需要在自定义节点或工作流设计中实现。 扩展性: 可以通过实现 INode 接口创建任意复杂的自定义任务节点。事件管理器和调度器接口也允许替换为自定义或第三方实现。 依赖: 核心库目前仅依赖 uuid。如果使用分布式调度器或特定事件管理器，可能需要引入额外的依赖（如 amqplib, @kubernetes/client-node 等）。 ES 模块: 包被设计为 ES 模块 ("type": "module")，并使用 package.json 的 "exports" 字段进行解析。请确保您的 Node.js 版本和 tsconfig.json 配置支持这一点。