diff --git a/packages/core/client/src/locale/zh_CN.ts b/packages/core/client/src/locale/zh_CN.ts index 796279ba4..d91de6e5e 100644 --- a/packages/core/client/src/locale/zh_CN.ts +++ b/packages/core/client/src/locale/zh_CN.ts @@ -553,6 +553,13 @@ export default { 'Continue after all branches succeeded': '全部分支都成功后才能继续', 'Continue after any branch succeeded': '任意分支成功后就继续', + 'Delay': '延时', + 'Duration': '时长', + 'End Status': '到时状态', + 'Select status': '选择状态', + 'Succeed and continue': '通过并继续', + 'Fail and exit': '失败并退出', + 'Create record': '新增数据', 'Update record': '更新数据', 'Query record': '查询数据', diff --git a/packages/core/client/src/workflow/calculators.tsx b/packages/core/client/src/workflow/calculators.tsx index 98b0e122e..6725600fe 100644 --- a/packages/core/client/src/workflow/calculators.tsx +++ b/packages/core/client/src/workflow/calculators.tsx @@ -270,14 +270,6 @@ export function Operand({ display: flex; gap: .5em; align-items: center; - - .ant-select, - .ant-cascader-picker, - .ant-picker, - .ant-input-number, - .ant-input-affix-wrapper{ - width: auto; - } `}> !(v % item.value)); +} + +export default function ({ value = 60000, onChange }) { + const { t } = useTranslation(); + const option = getNumberOption(value); + const quantity = Math.round(value / option.value); + + return ( +
+ onChange(Math.round(v * option.value))}/> + +
+ ); +} diff --git a/packages/core/client/src/workflow/nodes/delay.tsx b/packages/core/client/src/workflow/nodes/delay.tsx new file mode 100644 index 000000000..954a4d42b --- /dev/null +++ b/packages/core/client/src/workflow/nodes/delay.tsx @@ -0,0 +1,38 @@ +import Duration from "../components/Duration"; + +export default { + title: '{{t("Delay")}}', + type: 'delay', + group: 'control', + fieldset: { + 'config.duration': { + type: 'number', + name: 'config.duration', + title: '{{t("Duration")}}', + 'x-decorator': 'FormItem', + 'x-component': 'Duration', + }, + 'config.endStatus': { + type: 'number', + name: 'config.endStatus', + title: '{{t("End Status")}}', + 'x-decorator': 'FormItem', + 'x-component': 'Select', + 'x-component-props': { + placeholder: '{{t("Select status")}}', + }, + enum: [ + { label: '{{t("Succeed and continue")}}', value: 1 }, + { label: '{{t("Fail and exit")}}', value: -1 }, + ] + } + }, + view: { + + }, + scope: { + }, + components: { + Duration + } +}; diff --git a/packages/core/client/src/workflow/nodes/index.tsx b/packages/core/client/src/workflow/nodes/index.tsx index 8f817ff0d..f2e6c4fc7 100644 --- a/packages/core/client/src/workflow/nodes/index.tsx +++ b/packages/core/client/src/workflow/nodes/index.tsx @@ -11,11 +11,13 @@ import { AddButton, useFlowContext } from '../WorkflowCanvas'; import calculation from './calculation'; import condition from './condition'; -import create from './create'; -import destroy from './destroy'; import parallel from './parallel'; +import delay from './delay'; + import query from './query'; +import create from './create'; import update from './update'; +import destroy from './destroy'; export interface Instruction { title: string; @@ -33,13 +35,15 @@ export interface Instruction { export const instructions = new Registry(); +instructions.register('condition', condition); +instructions.register('parallel', parallel); +instructions.register('calculation', calculation); +instructions.register('delay', delay); + instructions.register('query', query); instructions.register('create', create); instructions.register('update', update); instructions.register('destroy', destroy); -instructions.register('condition', condition); -instructions.register('parallel', parallel); -instructions.register('calculation', calculation); function useUpdateAction() { const { t } = useTranslation(); @@ -220,7 +224,11 @@ export function NodeDefaultView(props) { 'x-component-props': { disabled: workflow.executed, className: css` - .ant-select{ + .ant-select, + .ant-cascader-picker, + .ant-picker, + .ant-input-number, + .ant-input-affix-wrapper{ width: auto; } ` diff --git a/packages/plugins/workflow/src/Plugin.ts b/packages/plugins/workflow/src/Plugin.ts index e98e5cef6..5b5d3dbd2 100644 --- a/packages/plugins/workflow/src/Plugin.ts +++ b/packages/plugins/workflow/src/Plugin.ts @@ -94,8 +94,17 @@ export default class WorkflowPlugin extends Plugin { this.toggle(workflow); }); }); - // [Life Cycle]: initialize all necessary seed data - // this.app.on('db.init', async () => {}); + + this.app.on('beforeStop', async () => { + const collection = db.getCollection('workflows'); + const workflows = await collection.repository.find({ + filter: { enabled: true }, + }); + + workflows.forEach((workflow: WorkflowModel) => { + this.toggle(workflow, false); + }); + }); } toggle(workflow: WorkflowModel, enable?: boolean) { diff --git a/packages/plugins/workflow/src/Processor.ts b/packages/plugins/workflow/src/Processor.ts index 6f103c3aa..5fa4c8e4e 100644 --- a/packages/plugins/workflow/src/Processor.ts +++ b/packages/plugins/workflow/src/Processor.ts @@ -148,7 +148,7 @@ export default class Processor { let job; try { // call instruction to get result and status - job = await instruction.call(node, prevJob, this); + job = await instruction(node, prevJob, this); if (!job) { return null; } @@ -233,13 +233,23 @@ export default class Processor { async saveJob(payload) { const { database } = this.execution.constructor; const { model } = database.getCollection('jobs'); - const [job] = (await model.upsert( - { + let job; + if (payload instanceof model) { + job = await payload.save({ transaction: this.transaction }); + } else if (payload.id) { + [job] = await model.update(payload, { + where: { id: payload.id }, + returning: true, + transaction: this.transaction + }); + } else { + job = await model.create({ ...payload, executionId: this.execution.id, - }, - { transaction: this.transaction }, - )) as unknown as [JobModel, boolean | null]; + }, { + transaction: this.transaction + }); + } this.jobsMap.set(job.id, job); this.jobsMapByNodeId[job.nodeId] = job.result; diff --git a/packages/plugins/workflow/src/__tests__/index.ts b/packages/plugins/workflow/src/__tests__/index.ts index 3e6265faa..05d75808c 100644 --- a/packages/plugins/workflow/src/__tests__/index.ts +++ b/packages/plugins/workflow/src/__tests__/index.ts @@ -4,6 +4,7 @@ import { MockServer, mockServer } from '@nocobase/test'; import Plugin from '..'; import { JOB_STATUS } from '../constants'; import calculators from '../calculators'; +import { ApplicationOptions } from '@nocobase/server'; export function sleep(ms: number) { return new Promise(resolve => { @@ -11,13 +12,17 @@ export function sleep(ms: number) { }); } -export async function getApp(options = {}): Promise { +interface MockAppOptions extends ApplicationOptions { + manual?: boolean; +} + +export async function getApp({ manual, ...options }: MockAppOptions = {}): Promise { const app = mockServer(options); app.plugin(Plugin, { instructions: { echo: { - run({ result }, execution) { + run(node, { result }, execution) { return { status: JOB_STATUS.RESOLVED, result @@ -26,18 +31,18 @@ export async function getApp(options = {}): Promise { }, error: { - run(input, execution) { + run(node, input, execution) { throw new Error('definite error'); } }, 'prompt->error': { - run(this, input, execution) { + run(node, input, execution) { return { status: JOB_STATUS.PENDING }; }, - resume(this, input, execution) { + resume(node, input, execution) { throw new Error('input failed'); } } @@ -60,7 +65,9 @@ export async function getApp(options = {}): Promise { console.error(error); } - await app.start(); + if (!manual) { + await app.start(); + } return app; } diff --git a/packages/plugins/workflow/src/__tests__/instructions/delay.test.ts b/packages/plugins/workflow/src/__tests__/instructions/delay.test.ts new file mode 100644 index 000000000..0fa83ea2e --- /dev/null +++ b/packages/plugins/workflow/src/__tests__/instructions/delay.test.ts @@ -0,0 +1,135 @@ +import { Application } from '@nocobase/server'; +import Database from '@nocobase/database'; +import { getApp, sleep } from '..'; +import { EXECUTION_STATUS, JOB_STATUS } from '../../constants'; + + + +describe('workflow > instructions > delay', () => { + let app: Application; + let db: Database; + let PostRepo; + let WorkflowModel; + let workflow; + + beforeEach(async () => { + app = await getApp(); + + db = app.db; + WorkflowModel = db.getCollection('workflows').model; + PostRepo = db.getCollection('posts').repository; + + workflow = await WorkflowModel.create({ + enabled: true, + type: 'collection', + config: { + mode: 1, + collection: 'posts' + } + }); + }); + + afterEach(() => app.stop()); + + describe('runtime', () => { + it('delay to resolved', async () => { + const n1 = await workflow.createNode({ + type: 'delay', + config: { + duration: 1000, + endStatus: JOB_STATUS.RESOLVED + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toEqual(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + await sleep(2000); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toEqual(EXECUTION_STATUS.RESOLVED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + }); + + it('delay to reject', async () => { + const n1 = await workflow.createNode({ + type: 'delay', + config: { + duration: 1000, + endStatus: JOB_STATUS.REJECTED + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toEqual(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + await sleep(2000); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toEqual(EXECUTION_STATUS.REJECTED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.REJECTED); + }); + }); + + describe('app lifecycle', () => { + beforeEach(async () => { + await workflow.createNode({ + type: 'delay', + config: { + duration: 2000, + endStatus: JOB_STATUS.RESOLVED + } + }); + }); + + it('restart app should trigger delayed job', async () => { + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toEqual(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + await app.stop(); + await sleep(500); + + await app.start(); + await sleep(2000); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toEqual(EXECUTION_STATUS.RESOLVED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + }); + + it('restart app should trigger missed delayed job', async () => { + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toEqual(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + await app.stop(); + await sleep(2000); + + await app.start(); + await sleep(500); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toEqual(EXECUTION_STATUS.RESOLVED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + }); + }); +}); diff --git a/packages/plugins/workflow/src/instructions/calculation.ts b/packages/plugins/workflow/src/instructions/calculation.ts index 21b5dfb19..47d7dc791 100644 --- a/packages/plugins/workflow/src/instructions/calculation.ts +++ b/packages/plugins/workflow/src/instructions/calculation.ts @@ -30,8 +30,8 @@ import { calculate } from "../calculators"; // } export default { - async run(this: FlowNodeModel, prevJob, processor) { - const { calculation } = this.config || {}; + async run(node: FlowNodeModel, prevJob, processor) { + const { calculation } = node.config || {}; const result = calculation ? calculate({ diff --git a/packages/plugins/workflow/src/instructions/condition.ts b/packages/plugins/workflow/src/instructions/condition.ts index d8fb4ad87..af1072ba9 100644 --- a/packages/plugins/workflow/src/instructions/condition.ts +++ b/packages/plugins/workflow/src/instructions/condition.ts @@ -73,10 +73,10 @@ function logicCalculate(calculation, input, processor) { export default { - async run(this, prevJob, processor) { + async run(node, prevJob, processor) { // TODO(optimize): loading of jobs could be reduced and turned into incrementally in processor // const jobs = await processor.getJobs(); - const { calculation, rejectOnFalse } = this.config || {}; + const { calculation, rejectOnFalse } = node.config || {}; const result = logicCalculate(calculation, prevJob, processor); if (!result && rejectOnFalse) { @@ -90,12 +90,12 @@ export default { status: JOB_STATUS.RESOLVED, result, // TODO(optimize): try unify the building of job - nodeId: this.id, + nodeId: node.id, upstreamId: prevJob && prevJob.id || null }; const branchNode = processor.nodes - .find(item => item.upstream === this && Boolean(item.branchIndex) === result); + .find(item => item.upstream === node && Boolean(item.branchIndex) === result); if (!branchNode) { return job; @@ -106,13 +106,13 @@ export default { return processor.run(branchNode, savedJob); }, - async resume(this, branchJob, processor) { + async resume(node, branchJob, processor) { if (branchJob.status === JOB_STATUS.RESOLVED) { - // return to continue this.downstream + // return to continue node.downstream return branchJob; } // pass control to upper scope by ending current scope - return processor.end(this, branchJob); + return processor.end(node, branchJob); } }; diff --git a/packages/plugins/workflow/src/instructions/create.ts b/packages/plugins/workflow/src/instructions/create.ts index c631e0669..c9a25dfec 100644 --- a/packages/plugins/workflow/src/instructions/create.ts +++ b/packages/plugins/workflow/src/instructions/create.ts @@ -2,13 +2,13 @@ import { JOB_STATUS } from "../constants"; import FlowNodeModel from "../models/FlowNode"; export default { - async run(this: FlowNodeModel, input, processor) { + async run(node: FlowNodeModel, input, processor) { const { collection, params = {} - } = this.config; + } = node.config; - const repo = (this.constructor).database.getRepository(collection); + const repo = (node.constructor).database.getRepository(collection); const options = processor.getParsedValue(params); const result = await repo.create({ ...options, diff --git a/packages/plugins/workflow/src/instructions/delay.ts b/packages/plugins/workflow/src/instructions/delay.ts new file mode 100644 index 000000000..3ac1239cb --- /dev/null +++ b/packages/plugins/workflow/src/instructions/delay.ts @@ -0,0 +1,92 @@ +import Plugin from '..'; +import { JOB_STATUS } from "../constants"; +import ExecutionModel from '../models/Execution'; +import JobModel from '../models/Job'; +import Processor from '../Processor'; + +type ValueOf = T[keyof T]; + +interface DelayConfig { + endStatus: ValueOf; + duration: number; +} + +export default class { + timers: Map = new Map(); + + constructor(protected plugin: Plugin) { + plugin.app.on('beforeStart', () => this.load()); + plugin.app.on('beforeStop', () => this.unload()) + } + + async load() { + const { model } = this.plugin.db.getCollection('jobs'); + const jobs = await model.findAll({ + where: { + status: JOB_STATUS.PENDING + }, + include: [ + { + association: 'execution' + }, + { + association: 'node', + where: { + type: 'delay' + }, + required: true + } + ] + }) as JobModel[]; + + jobs.forEach(job => { + this.schedule(job, job.node.config.duration); + }); + } + + unload() { + for (const timer of this.timers.values()) { + clearTimeout(timer); + } + + this.timers = new Map(); + } + + schedule(job, duration: number) { + const now = new Date(); + const createdAt = Date.parse(job.createdAt); + const delay = createdAt + duration - now.getTime(); + const trigger = this.trigger.bind(this, job); + this.timers.set(job.id, setTimeout(trigger, Math.max(0, delay))); + } + + async trigger(job) { + const { execution = await job.getExecution() as ExecutionModel } = job; + const processor = this.plugin.createProcessor(execution); + await processor.resume(job); + if (this.timers.get(job.id)) { + this.timers.delete(job.id); + } + } + + run = async (node, prevJob, processor: Processor) => { + const job = await processor.saveJob({ + status: JOB_STATUS.PENDING, + result: null, + nodeId: node.id, + upstreamId: prevJob?.id ?? null + }); + + const { duration } = node.config as DelayConfig; + // add to schedule + this.schedule(job, duration); + + return processor.end(node, job); + }; + + resume = async (node, prevJob, processor: Processor) => { + const { endStatus } = node.config as DelayConfig; + prevJob.set('status', endStatus); + return prevJob; + }; +} diff --git a/packages/plugins/workflow/src/instructions/destroy.ts b/packages/plugins/workflow/src/instructions/destroy.ts index 8d224566d..c94ddab9d 100644 --- a/packages/plugins/workflow/src/instructions/destroy.ts +++ b/packages/plugins/workflow/src/instructions/destroy.ts @@ -2,13 +2,13 @@ import { JOB_STATUS } from "../constants"; import FlowNodeModel from "../models/FlowNode"; export default { - async run(this: FlowNodeModel, input, processor) { + async run(node: FlowNodeModel, input, processor) { const { collection, params = {} - } = this.config; + } = node.config; - const repo = (this.constructor).database.getRepository(collection); + const repo = (node.constructor).database.getRepository(collection); const options = processor.getParsedValue(params); const result = await repo.destroy({ ...options, diff --git a/packages/plugins/workflow/src/instructions/index.ts b/packages/plugins/workflow/src/instructions/index.ts index 4f953c7ad..4d3e81f01 100644 --- a/packages/plugins/workflow/src/instructions/index.ts +++ b/packages/plugins/workflow/src/instructions/index.ts @@ -1,3 +1,7 @@ +import path from 'path'; + +import { requireModule } from '@nocobase/utils'; + import FlowNodeModel from '../models/FlowNode'; import Plugin from '..'; @@ -24,7 +28,7 @@ export type InstructionResult = Job | Promise; // - base on input and context, do any calculations or system call (io), and produce a result or pending. export interface Instruction { run( - this: FlowNodeModel, + node: FlowNodeModel, // what should input to be? // - just use previously output result for convenience? input: any, @@ -35,7 +39,7 @@ export interface Instruction { // for start node in main flow (or branch) to resume when manual sub branch triggered resume?( - this: FlowNodeModel, + node: FlowNodeModel, input: any, processor: Processor ): InstructionResult @@ -56,6 +60,8 @@ export default function( instructions.register('update', update); instructions.register('destroy', destroy); + instructions.register('delay', new (requireModule(path.join(__dirname, 'delay')))(plugin)); + for (const [name, instruction] of Object.entries(more)) { instructions.register(name, typeof instruction === 'function' ? new instruction(plugin) : instruction); } diff --git a/packages/plugins/workflow/src/instructions/parallel.ts b/packages/plugins/workflow/src/instructions/parallel.ts index 547ca330f..9bb41dd5b 100644 --- a/packages/plugins/workflow/src/instructions/parallel.ts +++ b/packages/plugins/workflow/src/instructions/parallel.ts @@ -40,15 +40,15 @@ const StatusGetters = { }; export default { - async run(this: FlowNodeModel, prevJob: JobModel, processor: Processor) { + async run(node: FlowNodeModel, prevJob: JobModel, processor: Processor) { const branches = processor.nodes - .filter(item => item.upstream === this && item.branchIndex !== null) + .filter(item => item.upstream === node && item.branchIndex !== null) .sort((a, b) => a.branchIndex - b.branchIndex); const job = await processor.saveJob({ status: JOB_STATUS.PENDING, result: Array(branches.length).fill(null), - nodeId: this.id, + nodeId: node.id, upstreamId: prevJob?.id ?? null }); @@ -59,11 +59,11 @@ export default { // another benifit of this is, it could handle sequenced branches in future. await branches.reduce((promise: Promise, branch) => promise.then(() => processor.run(branch, job)), Promise.resolve()); - return processor.end(this, job); + return processor.end(node, job); }, - async resume(this, branchJob, processor: Processor) { - const job = processor.findBranchParentJob(branchJob, this); + async resume(node: FlowNodeModel, branchJob, processor: Processor) { + const job = processor.findBranchParentJob(branchJob, node); const { result, status } = job; // if parallel has been done (resolved / rejected), do not care newly executed branch jobs. @@ -74,7 +74,7 @@ export default { // find the index of the node which start the branch const jobNode = processor.nodesMap.get(branchJob.nodeId); const { branchIndex } = processor.findBranchStartNode(jobNode); - const { mode = PARALLEL_MODE.ALL } = this.config || {}; + const { mode = PARALLEL_MODE.ALL } = node.config || {}; const newResult = [...result.slice(0, branchIndex), branchJob.get(), ...result.slice(branchIndex + 1)]; job.set({ @@ -84,7 +84,7 @@ export default { if (job.status === JOB_STATUS.PENDING) { await job.save({ transaction: processor.transaction }); - return processor.end(this, job); + return processor.end(node, job); } return job; diff --git a/packages/plugins/workflow/src/instructions/prompt.ts b/packages/plugins/workflow/src/instructions/prompt.ts index 005965fdf..80dc88f4b 100644 --- a/packages/plugins/workflow/src/instructions/prompt.ts +++ b/packages/plugins/workflow/src/instructions/prompt.ts @@ -1,13 +1,13 @@ import { JOB_STATUS } from "../constants"; export default { - run(this, input, processor) { + run(node, input, processor) { return { status: JOB_STATUS.PENDING }; }, - resume(this, job, processor) { + resume(node, job, processor) { job.set('status', JOB_STATUS.RESOLVED); return job; } diff --git a/packages/plugins/workflow/src/instructions/query.ts b/packages/plugins/workflow/src/instructions/query.ts index 37d63cee9..9001427b8 100644 --- a/packages/plugins/workflow/src/instructions/query.ts +++ b/packages/plugins/workflow/src/instructions/query.ts @@ -3,14 +3,14 @@ import Processor from "../Processor"; import { JOB_STATUS } from "../constants"; export default { - async run(this: FlowNodeModel, input, processor: Processor) { + async run(node: FlowNodeModel, input, processor: Processor) { const { collection, multiple, params = {} - } = this.config; + } = node.config; - const repo = (this.constructor).database.getRepository(collection); + const repo = (node.constructor).database.getRepository(collection); const options = processor.getParsedValue(params); const result = await (multiple ? repo.find : repo.findOne).call(repo, { ...options, diff --git a/packages/plugins/workflow/src/instructions/update.ts b/packages/plugins/workflow/src/instructions/update.ts index 4aed59232..e0d2fede6 100644 --- a/packages/plugins/workflow/src/instructions/update.ts +++ b/packages/plugins/workflow/src/instructions/update.ts @@ -3,14 +3,14 @@ import Processor from "../Processor"; import { JOB_STATUS } from "../constants"; export default { - async run(this: FlowNodeModel, input, processor: Processor) { + async run(node: FlowNodeModel, input, processor: Processor) { const { collection, multiple = false, params = {} - } = this.config; + } = node.config; - const repo = (this.constructor).database.getRepository(collection); + const repo = (node.constructor).database.getRepository(collection); const options = processor.getParsedValue(params); const result = await repo.update({ ...options,