乾渡（noomiflow）

乾渡是面向 Node.js / TypeScript 的轻量工作流引擎：用 JSON 描述流程，用 Relaen 持久化定义、实例与任务，适合审批、编排与业务自动化。

包名：noomiflow　·　原项目演进自 noomiflow

为何叫乾渡

「乾」为天、为行；「渡」为抵达。乾渡希望用简洁的 API，把业务从起点安全地渡到终点——分支、汇聚、等待、子流程皆可。

特性一览

| 能力 | 说明 | |------|------| | 流程控制 | 顺序流；排他 / 并行 / 包容网关 | | 人工任务 | 指派、候选人、候选组、认领、改派 | | 自动任务 | 本地 JS 模块任务（moduletask） | | 等待与协作 | 定时 / 消息 / 信号等待；子流程调用 | | 变量与恢复 | 流程变量、网关汇聚计数、等待态持久化与恢复 | | 版本发布 | 定义版本自增；draft / published | | 事件钩子 | NFEventBus 监听流程与节点生命周期 | | 历史轨迹 | 按实例查询节点办理记录 |

环境要求

• Node.js 18+（推荐 20+）
• TypeScript 5+（开发）
• MySQL 等 Relaen 支持的数据库（运行时持久化）

安装

npm install qiandu

依赖需自行初始化 Relaen，并将 entities 指向本包内实体路径（构建后一般为 node_modules/qiandu/dist/core/entity/**/*.js）。

快速开始

import { RelaenManager } from 'relaen';
import { NFEngine, NFUserManager, NFEventBus } from 'qiandu';

await RelaenManager.init({
  dialect: 'mysql',
  host: '127.0.0.1',
  port: 3306,
  username: 'root',
  password: '***',
  database: 'qiandu',
  entities: ['node_modules/qiandu/dist/core/entity/**/*.js'],
});

NFEventBus.on('process.end', ({ process }) => {
  console.log('流程结束', process.entity.processId);
});

// 1. 定义并发布流程
await NFEngine.defineProcess({
  name: '人工审核流程',
  status: 'published',
  nodes: [ /* 见下方 */ ],
});

// 2. 创建实例并启动（发起人 userId = 1）
const proc = await NFEngine.createProcess('人工审核流程', '订单-001', 1);
await proc.start();

// 3. 写流程变量 + 办理（userId 必填；agree 只进任务记录）
await proc.next({
  userId: 1,
  variables: { data: 1 },
});

// 或：认领后完成
const ts = NFEngine.getTaskService();
const todo = await NFUserManager.getUnHandleNodes(1);
await ts.claim(todo.rows[0].nodeId, 1);
await ts.finish(todo.rows[0].nodeId, {
  userId: 1,
  agree: 1,
  reason: '通过',
  variables: { approved: true },
});

流程定义

定义对象必须包含 name 和 nodes：{ name, nodes, dueTime?, version?, status? }。
节点之间用 sequence 连接；条件写在顺序流的 cond 上。

{
  "name": "人工审核流程",
  "nodes": [
    { "type": "start", "id": "start" },
    { "type": "sequence", "src": "start", "dst": "task1" },
    { "type": "usertask", "id": "task1", "name": "提交审核" },
    { "type": "sequence", "src": "task1", "dst": "exgate1" },
    { "type": "exclusive", "id": "exgate1" },
    { "type": "sequence", "src": "exgate1", "dst": "end", "cond": "data==1" },
    { "type": "sequence", "src": "exgate1", "dst": "task2", "cond": "data==2" },
    {
      "type": "usertask",
      "id": "task2",
      "name": "审核",
      "candidateUsers": "2"
    },
    { "type": "sequence", "src": "task2", "dst": "end" },
    { "type": "end", "id": "end" }
  ]
}

节点类型

| type | 说明 | |------|------| | start / end | 开始 / 结束 | | sequence | 顺序流：src / dst；可选 cond、default | | usertask | 人工任务：assignee / candidateUsers / candidateGroups（均为用户或组 id） | | moduletask | 本地模块：必填 path | | exclusive | 排他：第一条满足条件的出口，否则 default | | parallel | 并行：分叉全出 / 汇聚等全部入边 | | inclusive | 包容：多选出边；汇聚按激活可达数等待 | | wait | 等待：waitKind = timer | message | signal | | subprocess | 子流程：callProcess 为已发布定义名 |

等待 / 子流程示例

{ "type": "wait", "id": "w1", "waitKind": "timer", "delay": 60000 }
{ "type": "wait", "id": "w2", "waitKind": "message", "messageName": "paid" }
{ "type": "wait", "id": "w3", "waitKind": "signal", "signalName": "go" }
{ "type": "subprocess", "id": "sub1", "callProcess": "子流程名" }

await proc.message('paid', { amount: 100 });
await proc.signal('go');
await proc.resumeWaits(); // 到期定时器、已结束子流程

核心 API

NFEngine

| 方法 | 说明 | |------|------| | defineProcess(cfg) | 保存定义（版本自动 +1，默认 published） | | publishProcess(name, ver?) | 发布草稿 | | listDefinitions(name?) | 列出定义 | | findDefinition(name, ver?) | 查找定义（默认最新已发布） | | createProcess(name, instName, userId?, ver?) | 创建实例 | | getInstance(id) | 加载并恢复运行态 | | getHistory(id) | 办理轨迹 | | closeProcess(id, reason?) | 强制关闭并清理待办 | | getTaskService() | 任务服务 |

NFProcess

| 方法 | 说明 | |------|------| | start / next / end / close | 生命周期 | | setParam / setParams / getParam | 流程变量 | | message / signal / resumeWaits | 等待唤醒 | | getHistory / getWaits | 历史与等待态 |

next 约定（重要）

• userId：仅当前节点为 人工任务 时必填（办理权限校验）；等待 / 模块 / 网关等节点可省略
• agree / reason：只写入任务办理记录（NfNode），不会自动进入流程变量
• 网关条件等业务变量：使用 setParams 或 next({ variables: { ... } })

await proc.next({
  userId: 2,
  agree: 1,
  reason: '审核通过',
  variables: { approved: true }, // 供后续网关使用
});

NFTaskService

认领 / 取消认领 / 指派 / 候选人 / 候选组 / 查询 / finish(taskId, cfg?)。
taskId 为任务实例 id（NfNode.nodeId），不是定义里的节点 id。

NFUserManager

用户与组维护；getUnHandleNodes / getHandledNodes / getCreatedProcess。

NFEventBus

import { NFEventBus } from 'qiandu';

NFEventBus.on('process.start', ({ process }) => {});
NFEventBus.on('task.complete', ({ node, cfg }) => {});
NFEventBus.on('wait.resume', ({ nodeId }) => {});

| 事件 | 时机 | |------|------| | process.start / end / close / error | 流程生命周期 | | node.enter / leave / error | 节点进出 | | task.create / complete | 人工任务 | | wait.resume | 等待被唤醒 |

目录结构

dianqudev/
├── core/           # 引擎、节点、实体、任务服务
├── examples/       # 可运行示例与流程 JSON
├── test/           # 不依赖数据库的单元测试
├── index.ts        # 包导出
└── package.json    # name: qiandu

示例与本地开发

git clone <repo>
cd qiandu/diandu/dianqudev
npm install
npm run build
npm test                 # 单元测试（无需数据库）

带库示例（先改 examples/relaenconfig.ts）：

node dist/examples/usertest.js          # STEP=init 初始化用户组
STEP=demo node dist/examples/usertasktest.js
STEP=demo node dist/examples/moduletest.js

| 示例 | 说明 | |------|------| | usertest.ts | 用户 / 组 | | usertasktest.ts | 人工审核（flow1） | | moduletest.ts | 模块求和（flow2） | | flows/*.json | 流程定义 |

版本与发布

// 草稿
await NFEngine.defineProcess({ name: '订单', nodes, status: 'draft' });
await NFEngine.publishProcess('订单');

// 或直接发布（默认）
await NFEngine.defineProcess({ name: '订单', nodes });

// 按版本创建实例
await NFEngine.createProcess('订单', 'inst-1', 1, 2);

旧数据若无 defType，createProcess 会按定义名回退查找。

设计约定速查

1. 变量与办理分离：网关读 getParam；审批意见用 agree 落在任务行。
2. 人工任务会停住：需 next 或 TaskService.finish 推进。
3. 等待节点会挂起：用 message / signal / resumeWaits / next()（无需 userId）继续。
4. 关闭流程会清待办：已关闭实例不再出现在未办列表。

许可证

MIT

乾渡 qiandu — 把流程从起点渡到终点。