From a73c37c9271b85c7e8e0afaac30fe50f19e6e0cb Mon Sep 17 00:00:00 2001 From: Junyi Date: Wed, 27 Jul 2022 18:23:16 +0800 Subject: [PATCH] feat(plugin-workflow): add assignees config for prompt instruction (#690) --- .../plugins/workflow/src/server/Plugin.ts | 12 +- .../plugins/workflow/src/server/Processor.ts | 4 +- .../__tests__/instructions/prompt.test.ts | 573 ++++++++++++++++++ .../workflow/src/server/actions/index.ts | 4 +- .../workflow/src/server/actions/jobs.ts | 28 + .../workflow/src/server/actions/workflows.ts | 2 +- .../workflow/src/server/calculators/index.ts | 12 +- .../plugins/workflow/src/server/constants.ts | 4 +- .../server/extensions/assignees/actions.ts | 58 ++ .../extensions/assignees/collections/jobs.ts | 18 + .../extensions/assignees/collections/users.ts | 17 + .../assignees/collections/users_jobs.ts | 42 ++ .../src/server/extensions/assignees/index.ts | 210 +++++++ .../workflow/src/server/extensions/index.ts | 5 + .../workflow/src/server/instructions/delay.ts | 5 +- .../workflow/src/server/instructions/index.ts | 34 +- .../src/server/instructions/prompt.ts | 76 ++- 17 files changed, 1063 insertions(+), 41 deletions(-) create mode 100644 packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts create mode 100644 packages/plugins/workflow/src/server/actions/jobs.ts create mode 100644 packages/plugins/workflow/src/server/extensions/assignees/actions.ts create mode 100644 packages/plugins/workflow/src/server/extensions/assignees/collections/jobs.ts create mode 100644 packages/plugins/workflow/src/server/extensions/assignees/collections/users.ts create mode 100644 packages/plugins/workflow/src/server/extensions/assignees/collections/users_jobs.ts create mode 100644 packages/plugins/workflow/src/server/extensions/assignees/index.ts create mode 100644 packages/plugins/workflow/src/server/extensions/index.ts diff --git a/packages/plugins/workflow/src/server/Plugin.ts b/packages/plugins/workflow/src/server/Plugin.ts index 5b5d3dbd2..1335a5fad 100644 --- a/packages/plugins/workflow/src/server/Plugin.ts +++ b/packages/plugins/workflow/src/server/Plugin.ts @@ -8,6 +8,8 @@ import initActions from './actions'; import initTriggers, { Trigger } from './triggers'; import initInstructions, { Instruction } from './instructions'; import Processor from './Processor'; +import calculators from './calculators'; +import extensions from './extensions'; import WorkflowModel from './models/Workflow'; import ExecutionModel from './models/Execution'; import { EXECUTION_STATUS } from './constants'; @@ -17,6 +19,8 @@ import { EXECUTION_STATUS } from './constants'; export default class WorkflowPlugin extends Plugin { instructions: Registry = new Registry(); triggers: Registry = new Registry(); + calculators = calculators; + extensions = extensions; onBeforeSave = async (instance: WorkflowModel, options) => { const Model = instance.constructor; @@ -80,6 +84,8 @@ export default class WorkflowPlugin extends Plugin { db.on('workflows.afterSave', (model: WorkflowModel) => this.toggle(model)); db.on('workflows.afterDestroy', (model: WorkflowModel) => this.toggle(model, false)); + this.app.on('afterLoadAll', async () => this.extensions.reduce((promise, extend) => promise.then(() => extend(this)), Promise.resolve())); + // [Life Cycle]: // * load all workflows in db // * add all hooks for enabled workflows @@ -122,10 +128,10 @@ export default class WorkflowPlugin extends Plugin { } } - async trigger(workflow, context: Object, options: Transactionable = {}) { + async trigger(workflow: WorkflowModel, context: Object, options: Transactionable = {}): Promise { // `null` means not to trigger if (context === null) { - return; + return null; } let transaction = null; @@ -145,7 +151,7 @@ export default class WorkflowPlugin extends Plugin { if (existed) { console.warn(`workflow ${workflow.id} has already been triggered in same execution (${transaction.id}), and newly triggering will be skipped.`); - return; + return null; } } diff --git a/packages/plugins/workflow/src/server/Processor.ts b/packages/plugins/workflow/src/server/Processor.ts index 93c9ccf42..a64008069 100644 --- a/packages/plugins/workflow/src/server/Processor.ts +++ b/packages/plugins/workflow/src/server/Processor.ts @@ -23,7 +23,7 @@ export default class Processor { [JOB_STATUS.PENDING]: EXECUTION_STATUS.STARTED, [JOB_STATUS.RESOLVED]: EXECUTION_STATUS.RESOLVED, [JOB_STATUS.REJECTED]: EXECUTION_STATUS.REJECTED, - [JOB_STATUS.CANCELLED]: EXECUTION_STATUS.CANCELLED, + [JOB_STATUS.CANCELED]: EXECUTION_STATUS.CANCELED, }; transaction: Transaction; @@ -168,8 +168,6 @@ export default class Processor { } let savedJob; - // TODO(optimize): many checking of resuming or new could be improved - // could be implemented separately in exec() / resume() if (job instanceof Model) { savedJob = (await job.save({ transaction: this.transaction })) as unknown as JobModel; } else { diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts new file mode 100644 index 000000000..f471deac2 --- /dev/null +++ b/packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts @@ -0,0 +1,573 @@ +import { MockServer } from '@nocobase/test'; +import Database from '@nocobase/database'; +import UserPlugin from '@nocobase/plugin-users'; +import { getApp, sleep } from '..'; +import { EXECUTION_STATUS, JOB_STATUS } from '../../constants'; + + + +describe('workflow > instructions > prompt', () => { + describe('base', () => { + let app: MockServer; + let agent; + let db: Database; + let PostRepo; + let WorkflowModel; + let workflow; + + beforeEach(async () => { + app = await getApp(); + agent = app.agent(); + 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(() => db.close()); + + it('resume to resolve', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + actions: { + [JOB_STATUS.RESOLVED]: 'submit' + } + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [pending] = await workflow.getExecutions(); + expect(pending.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await pending.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + const { status } = await agent.resource('jobs').submit({ + filterByTk: j1.id, + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 1 } + } + }); + expect(status).toBe(202); + + // NOTE: wait for no await execution + await sleep(500); + + const [resolved] = await workflow.getExecutions(); + expect(resolved.status).toBe(EXECUTION_STATUS.RESOLVED); + const [j2] = await resolved.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + expect(j2.result).toEqual({ a: 1 }); + }); + }); + + describe('assignees', () => { + let app: MockServer; + let agent; + let userAgents; + let db: Database; + let PostRepo; + let WorkflowModel; + let workflow; + let UserModel; + let users; + let UserJobModel; + + beforeEach(async () => { + app = await getApp({ + plugins: [ + '@nocobase/plugin-users' + ] + }); + agent = app.agent(); + db = app.db; + WorkflowModel = db.getCollection('workflows').model; + PostRepo = db.getCollection('posts').repository; + UserModel = db.getCollection('users').model; + UserJobModel = db.getModel('users_jobs'); + + users = await UserModel.bulkCreate([ + { id: 1, nickname: 'a' }, + { id: 2, nickname: 'b' } + ]); + + const userPlugin = app.getPlugin('@nocobase/plugin-users') as UserPlugin; + userAgents = users.map((user) => app.agent().auth(userPlugin.jwtService.sign({ + userId: user.id, + }), { type: 'bearer' })); + + workflow = await WorkflowModel.create({ + enabled: true, + type: 'collection', + config: { + mode: 1, + collection: 'posts' + } + }); + }); + + afterEach(() => db.close()); + + describe('mode: 0 (single record)', () => { + it('the only user assigned could submit', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id] + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [pending] = await workflow.getExecutions(); + expect(pending.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await pending.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + const usersJobs = await UserJobModel.findAll(); + expect(usersJobs.length).toBe(1); + expect(usersJobs[0].status).toBe(JOB_STATUS.PENDING); + expect(usersJobs[0].userId).toBe(users[0].id); + expect(usersJobs[0].jobId).toBe(j1.id); + + const res1 = await agent.resource('jobs').submit({ + filterByTk: j1.id + }); + expect(res1.status).toBe(401); + + const res2 = await userAgents[1].resource('jobs').submit({ + filterByTk: j1.id, + values: { + status: JOB_STATUS.RESOLVED + } + }); + expect(res2.status).toBe(404); + + const res3 = await userAgents[0].resource('jobs').submit({ + filterByTk: j1.id, + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 1 } + } + }); + expect(res3.status).toBe(202); + + await sleep(500); + + const [j2] = await pending.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + expect(j2.result).toEqual({ a: 1 }); + + const usersJobsAfter = await UserJobModel.findAll(); + expect(usersJobsAfter.length).toBe(1); + expect(usersJobsAfter[0].status).toBe(JOB_STATUS.RESOLVED); + expect(usersJobsAfter[0].result).toEqual({ a: 1 }); + + const res4 = await userAgents[0].resource('jobs').submit({ + filterByTk: j1.id, + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 2 } + } + }); + expect(res4.status).toBe(400); + }); + + it('any user assigned could submit', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id] + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const [pending] = await workflow.getExecutions(); + expect(pending.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await pending.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + + const res1 = await userAgents[1].resource('jobs').submit({ + filterByTk: j1.id, + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 1 } + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [j2] = await pending.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + expect(j2.result).toEqual({ a: 1 }); + + const res2 = await userAgents[0].resource('jobs').submit({ + filterByTk: j1.id, + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 2 } + } + }); + expect(res2.status).toBe(400); + }); + + it('also could submit to users_jobs api', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id] + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const usersJobs = await UserJobModel.findAll(); + expect(usersJobs.length).toBe(1); + expect(usersJobs[0].get('status')).toBe(JOB_STATUS.PENDING); + expect(usersJobs[0].get('userId')).toBe(users[0].id); + + const res = await userAgents[0].resource('users_jobs').submit({ + filterByTk: usersJobs[0].get('id'), + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 1 } + } + }); + expect(res.status).toBe(202); + + await sleep(500); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED); + const [job] = await execution.getJobs(); + expect(job.status).toBe(JOB_STATUS.RESOLVED); + expect(job.result).toEqual({ a: 1 }); + }); + }); + + describe('mode: 1 (multiple record, all)', () => { + it('all resolved', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id], + mode: 1 + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const pendingJobs = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(pendingJobs.length).toBe(2); + + const res1 = await userAgents[0].resource('users_jobs').submit({ + filterByTk: pendingJobs[0].get('id'), + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 1 } + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + expect(j1.result).toBe(0.5); + const usersJobs1 = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(usersJobs1.length).toBe(2); + + const res2 = await userAgents[1].resource('users_jobs').submit({ + filterByTk: pendingJobs[1].get('id'), + values: { + status: JOB_STATUS.RESOLVED, + result: { a: 2 } + } + }); + expect(res2.status).toBe(202); + + await sleep(500); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toBe(EXECUTION_STATUS.RESOLVED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + expect(j2.result).toBe(1); + }); + + it('first rejected', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id], + mode: 1 + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const pendingJobs = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(pendingJobs.length).toBe(2); + + const res1 = await userAgents[0].resource('users_jobs').submit({ + filterByTk: pendingJobs[0].get('id'), + values: { + status: JOB_STATUS.REJECTED + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toBe(EXECUTION_STATUS.REJECTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.REJECTED); + expect(j1.result).toBe(0.5); + const usersJobs1 = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(usersJobs1.length).toBe(2); + + const res2 = await userAgents[1].resource('users_jobs').submit({ + filterByTk: pendingJobs[1].get('id'), + values: { + status: JOB_STATUS.REJECTED, + result: { a: 2 } + } + }); + expect(res2.status).toBe(400); + }); + + it('last rejected', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id], + mode: 1 + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const pendingJobs = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(pendingJobs.length).toBe(2); + + const res1 = await userAgents[0].resource('users_jobs').submit({ + filterByTk: pendingJobs[0].get('id'), + values: { + status: JOB_STATUS.RESOLVED + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + expect(j1.result).toBe(0.5); + const usersJobs1 = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(usersJobs1.length).toBe(2); + + const res2 = await userAgents[1].resource('users_jobs').submit({ + filterByTk: pendingJobs[1].get('id'), + values: { + status: JOB_STATUS.REJECTED, + result: { a: 2 } + } + }); + expect(res2.status).toBe(202); + + await sleep(500); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toBe(EXECUTION_STATUS.REJECTED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.REJECTED); + expect(j2.result).toBe(1); + }); + }); + + describe('mode: -1 (multiple record, any)', () => { + it('first resolved', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id], + mode: -1 + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const pendingJobs = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(pendingJobs.length).toBe(2); + + const res1 = await userAgents[0].resource('users_jobs').submit({ + filterByTk: pendingJobs[0].get('id'), + values: { + status: JOB_STATUS.RESOLVED + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toBe(EXECUTION_STATUS.RESOLVED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.RESOLVED); + expect(j1.result).toBe(0.5); + + const res2 = await userAgents[1].resource('users_jobs').submit({ + filterByTk: pendingJobs[1].get('id'), + values: { + status: JOB_STATUS.REJECTED + } + }); + expect(res2.status).toBe(400); + }); + + it('any resolved', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id], + mode: -1 + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const pendingJobs = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(pendingJobs.length).toBe(2); + + const res1 = await userAgents[0].resource('users_jobs').submit({ + filterByTk: pendingJobs[0].get('id'), + values: { + status: JOB_STATUS.REJECTED + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + expect(j1.result).toBe(0.5); + + const res2 = await userAgents[1].resource('users_jobs').submit({ + filterByTk: pendingJobs[1].get('id'), + values: { + status: JOB_STATUS.RESOLVED + } + }); + expect(res2.status).toBe(202); + + await sleep(500); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toBe(EXECUTION_STATUS.RESOLVED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.RESOLVED); + expect(j2.result).toBe(1); + }); + + it('all rejected', async () => { + const n1 = await workflow.createNode({ + type: 'prompt', + config: { + assignees: [users[0].id, users[1].id], + mode: -1 + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + const UserJobModel = db.getModel('users_jobs'); + const pendingJobs = await UserJobModel.findAll({ + order: [[ 'userId', 'ASC' ]] + }); + expect(pendingJobs.length).toBe(2); + + const res1 = await userAgents[0].resource('users_jobs').submit({ + filterByTk: pendingJobs[0].get('id'), + values: { + status: JOB_STATUS.REJECTED + } + }); + expect(res1.status).toBe(202); + + await sleep(500); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toBe(EXECUTION_STATUS.STARTED); + const [j1] = await e1.getJobs(); + expect(j1.status).toBe(JOB_STATUS.PENDING); + expect(j1.result).toBe(0.5); + + const res2 = await userAgents[1].resource('users_jobs').submit({ + filterByTk: pendingJobs[1].get('id'), + values: { + status: JOB_STATUS.REJECTED + } + }); + expect(res2.status).toBe(202); + + await sleep(500); + + const [e2] = await workflow.getExecutions(); + expect(e2.status).toBe(EXECUTION_STATUS.REJECTED); + const [j2] = await e2.getJobs(); + expect(j2.status).toBe(JOB_STATUS.REJECTED); + expect(j2.result).toBe(1); + }); + }); + + describe('mode: (0,1) (multiple record, all to percent)', () => { + + }); + + describe('mode: (-1,0) (multiple record, any to percent)', () => { + + }); + }); +}); diff --git a/packages/plugins/workflow/src/server/actions/index.ts b/packages/plugins/workflow/src/server/actions/index.ts index 243b6ab90..2922d5057 100644 --- a/packages/plugins/workflow/src/server/actions/index.ts +++ b/packages/plugins/workflow/src/server/actions/index.ts @@ -1,5 +1,6 @@ import * as workflows from './workflows'; import * as nodes from './nodes'; +import * as jobs from './jobs'; function make(name, mod) { return Object.keys(mod).reduce((result, key) => ({ @@ -17,6 +18,7 @@ export default function({ app }) { }), ...make('flow_nodes', { update: nodes.update - }) + }), + ...make('jobs', jobs) }); } diff --git a/packages/plugins/workflow/src/server/actions/jobs.ts b/packages/plugins/workflow/src/server/actions/jobs.ts new file mode 100644 index 000000000..ea84b6d47 --- /dev/null +++ b/packages/plugins/workflow/src/server/actions/jobs.ts @@ -0,0 +1,28 @@ +import { Context } from '@nocobase/actions'; +import { JOB_STATUS } from '../constants'; + +export async function submit(context: Context, next) { + const { values } = context.action.params; + + const { body: instance } = context; + + // NOTE: validate status + if (instance.status !== JOB_STATUS.PENDING) { + return context.throw(400); + } + + // NOTE: validate assignee + instance.set({ + status: values.status, + result: values.result + }); + + context.status = 202; + + await next(); + + const plugin = context.app.pm.get('@nocobase/plugin-workflow'); + const processor = plugin.createProcessor(instance.execution); + // NOTE: resume the process and no `await` for quick returning + processor.resume(instance); +} diff --git a/packages/plugins/workflow/src/server/actions/workflows.ts b/packages/plugins/workflow/src/server/actions/workflows.ts index 463d4efba..cf17cd236 100644 --- a/packages/plugins/workflow/src/server/actions/workflows.ts +++ b/packages/plugins/workflow/src/server/actions/workflows.ts @@ -33,7 +33,7 @@ function migrateConfig(config, oldToNew) { case 'string': return value .replace( - /(\{\{\$jobsMapByNodeId\.)(\d+)/, + /(\{\{\$jobsMapByNodeId\.)([\w-]+)/, (_, prefix, id) => `${prefix}${oldToNew.get(Number.parseInt(id, 10)).id}` ); default: diff --git a/packages/plugins/workflow/src/server/calculators/index.ts b/packages/plugins/workflow/src/server/calculators/index.ts index a8f3e33e5..f8573f5f3 100644 --- a/packages/plugins/workflow/src/server/calculators/index.ts +++ b/packages/plugins/workflow/src/server/calculators/index.ts @@ -1,4 +1,4 @@ -import { get as getWithPath } from 'lodash'; +import { toNumber, get as getWithPath } from 'lodash'; import { Registry } from "@nocobase/utils"; import JobModel from '../models/Job'; @@ -141,23 +141,23 @@ calculators.register('<=', lte); function add(...args) { - return args.reduce((sum, a) => sum + a, 0); + return args.reduce((sum, a) => sum + toNumber(a), 0); } function minus(a, b) { - return a - b; + return toNumber(a) - toNumber(b); } function multiple(...args) { - return args.reduce((result, a) => result * a, 1); + return args.reduce((result, a) => result * toNumber(a), 1); } function divide(a, b) { - return a / b; + return toNumber(a) / toNumber(b); } function mod(a, b) { - return a % b; + return toNumber(a) % toNumber(b); } calculators.register('add', add); diff --git a/packages/plugins/workflow/src/server/constants.ts b/packages/plugins/workflow/src/server/constants.ts index 8cc682187..a66cb4bb7 100644 --- a/packages/plugins/workflow/src/server/constants.ts +++ b/packages/plugins/workflow/src/server/constants.ts @@ -2,14 +2,14 @@ export const EXECUTION_STATUS = { STARTED: 0, RESOLVED: 1, REJECTED: -1, - CANCELLED: -2 + CANCELED: -2 }; export const JOB_STATUS = { PENDING: 0, RESOLVED: 1, REJECTED: -1, - CANCELLED: -2 + CANCELED: -2 }; export const BRANCH_INDEX = { diff --git a/packages/plugins/workflow/src/server/extensions/assignees/actions.ts b/packages/plugins/workflow/src/server/extensions/assignees/actions.ts new file mode 100644 index 000000000..e981a3227 --- /dev/null +++ b/packages/plugins/workflow/src/server/extensions/assignees/actions.ts @@ -0,0 +1,58 @@ +import { Context, utils } from '@nocobase/actions'; +import { EXECUTION_STATUS, JOB_STATUS } from '../../constants'; + + + +export async function submit(context: Context, next) { + const repository = utils.getRepositoryFromParams(context); + const { filterByTk, values } = context.action.params; + const { currentUser } = context.state; + + if (!currentUser) { + return context.throw(401); + } + + const instance = await repository.findOne({ + filterByTk, + // filter: { + // userId: currentUser?.id + // }, + appends: ['job', 'node', 'execution'], + context + }); + + const { actions, assignees } = instance.node.config; + + // NOTE: validate status + if (instance.status !== JOB_STATUS.PENDING + || instance.job.status !== JOB_STATUS.PENDING + || instance.execution.status !== EXECUTION_STATUS.STARTED + || (actions && !actions[values.status]) + ) { + context.throw(400); + } + + if (!assignees.includes(currentUser.id) + || instance.userId !== currentUser.id + ) { + return context.throw(404); + } + + // NOTE: validate assignee + await instance.update({ + status: values.status, + result: values.result + }); + + context.body = instance; + context.status = 202; + + await next(); + + instance.job.latestUserJob = instance; + + const plugin = context.app.pm.get('@nocobase/plugin-workflow'); + const processor = plugin.createProcessor(instance.execution); + // NOTE: resume the process and no `await` for quick returning + processor.resume(instance.job); +} diff --git a/packages/plugins/workflow/src/server/extensions/assignees/collections/jobs.ts b/packages/plugins/workflow/src/server/extensions/assignees/collections/jobs.ts new file mode 100644 index 000000000..93613f72e --- /dev/null +++ b/packages/plugins/workflow/src/server/extensions/assignees/collections/jobs.ts @@ -0,0 +1,18 @@ +import { extend } from '@nocobase/database'; + +export default extend({ + name: 'jobs', + fields: [ + { + type: 'belongsToMany', + name: 'users', + through: 'users_jobs', + }, + { + type: 'hasMany', + name: 'usersJobs', + target: 'users_jobs', + foreignKey: 'jobId' + } + ] +}); diff --git a/packages/plugins/workflow/src/server/extensions/assignees/collections/users.ts b/packages/plugins/workflow/src/server/extensions/assignees/collections/users.ts new file mode 100644 index 000000000..b83626ab4 --- /dev/null +++ b/packages/plugins/workflow/src/server/extensions/assignees/collections/users.ts @@ -0,0 +1,17 @@ +import { extend } from '@nocobase/database'; + +export default extend({ + name: 'users', + fields: [ + { + type: 'belongsToMany', + name: 'jobs', + through: 'users_jobs', + }, + { + type: 'hasMany', + name: 'usersJobs', + target: 'users_jobs' + } + ] +}); diff --git a/packages/plugins/workflow/src/server/extensions/assignees/collections/users_jobs.ts b/packages/plugins/workflow/src/server/extensions/assignees/collections/users_jobs.ts new file mode 100644 index 000000000..ff28a6a4f --- /dev/null +++ b/packages/plugins/workflow/src/server/extensions/assignees/collections/users_jobs.ts @@ -0,0 +1,42 @@ +import { CollectionOptions } from '@nocobase/database'; + +export default { + name: 'users_jobs', + fields: [ + { + type: 'integer', + name: 'id', + primaryKey: true, + autoIncrement: true + }, + { + type: 'belongsTo', + name: 'job' + }, + { + type: 'belongsTo', + name: 'user' + }, + { + type: 'belongsTo', + name: 'execution' + }, + { + type: 'belongsTo', + name: 'node', + target: 'flow_nodes' + }, + { + type: 'belongsTo', + name: 'workflow' + }, + { + type: 'integer', + name: 'status' + }, + { + type: 'jsonb', + name: 'result' + } + ] +} as CollectionOptions; diff --git a/packages/plugins/workflow/src/server/extensions/assignees/index.ts b/packages/plugins/workflow/src/server/extensions/assignees/index.ts new file mode 100644 index 000000000..6b1bdf47b --- /dev/null +++ b/packages/plugins/workflow/src/server/extensions/assignees/index.ts @@ -0,0 +1,210 @@ +import path from 'path'; + +import { requireModule } from '@nocobase/utils'; +import { Context } from '@nocobase/actions'; + +import Plugin from '../../Plugin'; +import Prompt, { PromptConfig } from '../../instructions/prompt'; +import { submit } from './actions'; +import { JOB_STATUS } from '../../constants'; + + + +interface AssignedPromptConfig extends PromptConfig { + assignees?: number[]; + mode?: number; +} + +// NOTE: for single record mode (mode: 0/null) +async function middleware(context: Context, next) { + const { body: job, state, action } = context; + const { assignees, mode } = job.node.config as AssignedPromptConfig; + + // NOTE: skip to no user implementation + if (!assignees) { + return next(); + } + + if (!state.currentUser) { + return context.throw(401); + } + + if (!assignees.includes(state.currentUser.id)) { + return context.throw(404); + } + + // NOTE: multiple record mode could not use jobs:submit action + // should use users_jobs:submit/:id instead + if (mode) { + return context.throw(400); + } + + await next(); + + const data = { + userId: context.state.currentUser.id, + jobId: job.id, + nodeId: job.nodeId, + executionId: job.executionId, + workflowId: job.execution.workflowId, + status: job.status, + result: job.result + }; + + // NOTE: update users job after main job is done + const UserJobModel = context.db.getModel('users_jobs'); + let userJob = await UserJobModel.findOne({ + where: { + userId: context.state.currentUser.id, + jobId: job.id, + } + }); + if (userJob) { + await userJob.update(data); + } else { + userJob = await UserJobModel.create(data); + } +} + +async function run(node, prevJob, processor) { + const { assignees, mode } = node.config as AssignedPromptConfig; + if (!assignees) { + const { plugin } = processor.options; + const origin = plugin.instructions.get('prompt') as Prompt; + return origin.constructor.prototype.run.call(this, node, prevJob, processor); + } + + const job = await processor.saveJob({ + status: JOB_STATUS.PENDING, + result: mode ? [] : null, + nodeId: node.id, + upstreamId: prevJob?.id ?? null + }); + + // NOTE: batch create users jobs + const UserJobModel = processor.options.plugin.db.getModel('users_jobs'); + await UserJobModel.bulkCreate(assignees.map(userId => ({ + userId, + jobId: job.id, + nodeId: node.id, + executionId: job.executionId, + workflowId: node.workflowId, + status: JOB_STATUS.PENDING + })), { + transaction: processor.transaction + }); + + return job; +} + +const PROMPT_ASSIGNED_MODE = { + SINGLE: Symbol('single'), + ALL: Symbol('all'), + ANY: Symbol('any'), + ALL_PERCENTAGE: Symbol('all percentage'), + ANY_PERCENTAGE: Symbol('any percentage') +}; + +const Modes = { + [PROMPT_ASSIGNED_MODE.SINGLE]: { + getStatus(distribution, assignees) { + const done = distribution.find(item => item.status !== JOB_STATUS.PENDING && item.count > 0); + return done ? done.status : null + } + }, + [PROMPT_ASSIGNED_MODE.ALL]: { + getStatus(distribution, assignees) { + const resolved = distribution.find(item => item.status === JOB_STATUS.RESOLVED); + if (resolved && resolved.count === assignees.length) { + return JOB_STATUS.RESOLVED; + } + // NOTE: `rejected` or `canceled` + const failed = distribution.find(item => item.status < JOB_STATUS.PENDING); + if (failed && failed.count) { + return failed.status; + } + + return null; + } + }, + [PROMPT_ASSIGNED_MODE.ANY]: { + getStatus(distribution, assignees) { + const resolved = distribution.find(item => item.status === JOB_STATUS.RESOLVED); + if (resolved && resolved.count) { + return JOB_STATUS.RESOLVED; + } + const failedCount = distribution.reduce((count, item) => item.status < JOB_STATUS.PENDING ? count + item.count : count, 0); + // NOTE: all failures are considered as rejected for now + if (failedCount === assignees.length) { + return JOB_STATUS.REJECTED; + } + + return null; + } + } +}; + +function getMode(mode) { + switch (true) { + case mode === 1: + return Modes[PROMPT_ASSIGNED_MODE.ALL]; + case mode === -1: + return Modes[PROMPT_ASSIGNED_MODE.ANY]; + case mode > 0: + return Modes[PROMPT_ASSIGNED_MODE.ALL_PERCENTAGE]; + case mode < 0: + return Modes[PROMPT_ASSIGNED_MODE.ANY_PERCENTAGE]; + default: + return Modes[PROMPT_ASSIGNED_MODE.SINGLE]; + } +} + +async function resume(node, job, processor) { + // NOTE: check all users jobs related if all done then continue as parallel + const { assignees, mode } = node.config as AssignedPromptConfig; + + if (!assignees) { + const { plugin } = processor.options; + const origin = plugin.instructions.get('prompt') as Prompt; + return origin.constructor.prototype.resume.call(this, node, job, processor); + } + + const UserJobModel = processor.options.plugin.db.getModel('users_jobs'); + const distribution = await UserJobModel.count({ + where: { + jobId: job.id + }, + group: ['status'] + }); + + const submitted = distribution.reduce((count, item) => item.status !== JOB_STATUS.PENDING ? count + item.count : count, 0); + const result = mode ? (submitted || 0) / assignees.length : job.latestUserJob?.result ?? job.result; + job.set({ + status: getMode(mode).getStatus(distribution, assignees) ?? JOB_STATUS.PENDING, + result + }); + + return job; +} + +export default async function(plugin: Plugin) { + const instruction = plugin.instructions.get('prompt') as Prompt; + instruction.extend({ + run, + resume + }); + + instruction.use(middleware); + + // TODO(bug): through table should be load first because primary + // await plugin.db.import({ + // directory: path.join(__dirname, './collections') + // }); + plugin.db.collection(requireModule(path.join(__dirname, './collections/users_jobs'))); + plugin.db.collection(requireModule(path.join(__dirname, './collections/users'))); + plugin.db.collection(requireModule(path.join(__dirname, './collections/jobs'))); + + plugin.app.actions({ + 'users_jobs:submit': submit + }); +} diff --git a/packages/plugins/workflow/src/server/extensions/index.ts b/packages/plugins/workflow/src/server/extensions/index.ts new file mode 100644 index 000000000..f3fd53529 --- /dev/null +++ b/packages/plugins/workflow/src/server/extensions/index.ts @@ -0,0 +1,5 @@ +import assignees from './assignees'; + +export default [ + assignees +]; diff --git a/packages/plugins/workflow/src/server/instructions/delay.ts b/packages/plugins/workflow/src/server/instructions/delay.ts index 0355019d6..35e52c9fc 100644 --- a/packages/plugins/workflow/src/server/instructions/delay.ts +++ b/packages/plugins/workflow/src/server/instructions/delay.ts @@ -3,6 +3,7 @@ import { EXECUTION_STATUS, JOB_STATUS } from "../constants"; import ExecutionModel from '../models/Execution'; import JobModel from '../models/Job'; import Processor from '../Processor'; +import { Instruction } from '.'; type ValueOf = T[keyof T]; @@ -11,12 +12,12 @@ interface DelayConfig { duration: number; } -export default class { +export default class implements Instruction { timers: Map = new Map(); constructor(protected plugin: Plugin) { plugin.app.on('beforeStart', () => this.load()); - plugin.app.on('beforeStop', () => this.unload()) + plugin.app.on('beforeStop', () => this.unload()); } async load() { diff --git a/packages/plugins/workflow/src/server/instructions/index.ts b/packages/plugins/workflow/src/server/instructions/index.ts index 4d3e81f01..5c50e2d1e 100644 --- a/packages/plugins/workflow/src/server/instructions/index.ts +++ b/packages/plugins/workflow/src/server/instructions/index.ts @@ -7,15 +7,6 @@ import FlowNodeModel from '../models/FlowNode'; import Plugin from '..'; import Processor from '../Processor'; -import prompt from './prompt'; -import calculation from './calculation'; -import condition from './condition'; -import parallel from './parallel'; -import query from './query'; -import create from './create'; -import update from './update'; -import destroy from './destroy'; - export type Job = { status: number; result?: unknown; @@ -51,18 +42,23 @@ export default function( ) { const { instructions } = plugin; - instructions.register('prompt', prompt); - instructions.register('calculation', calculation); - instructions.register('condition', condition); - instructions.register('parallel', parallel); - instructions.register('query', query); - instructions.register('create', create); - instructions.register('update', update); - instructions.register('destroy', destroy); + const natives = [ + 'calculation', + 'condition', + 'parallel', + 'delay', + 'prompt', + 'query', + 'create', + 'update', + 'destroy' + ].reduce((result, key) => Object.assign(result, { [key]: key }), {}); - instructions.register('delay', new (requireModule(path.join(__dirname, 'delay')))(plugin)); + for (const [name, value] of Object.entries({ ...more, ...natives })) { + const instruction = typeof value === 'string' + ? requireModule(path.isAbsolute(value) ? value : path.join(__dirname, value)) + : value; - 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/server/instructions/prompt.ts b/packages/plugins/workflow/src/server/instructions/prompt.ts index 80dc88f4b..df1df1173 100644 --- a/packages/plugins/workflow/src/server/instructions/prompt.ts +++ b/packages/plugins/workflow/src/server/instructions/prompt.ts @@ -1,14 +1,82 @@ -import { JOB_STATUS } from "../constants"; +import compose from 'koa-compose'; + +import { Context, utils } from '@nocobase/actions'; + +import Plugin from '..'; +import { JOB_STATUS } from "../constants"; +import { Instruction } from '.'; + + + +export interface PromptConfig { + fields: []; + actions; +} + +async function loadJob(context: Context, next) { + const { filterByTk, values } = context.action.params; + if (!context.body) { + const jobRepo = utils.getRepositoryFromParams(context); + const job = await jobRepo.findOne({ + filterByTk, + appends: ['node', 'execution'], + context + }); + + if (!filterByTk || !job) { + return context.throw(404); + } + + // cache + context.body = job; + } + + const { type, config } = context.body.node; + if (type === 'prompt' + && config.actions + && !config.actions[values.status]) { + return context.throw(400); + } + + await next(); +} + +export default class implements Instruction { + middlewares = []; + + constructor(protected plugin: Plugin) { + plugin.app.resourcer.use(this.middleware); + } + + middleware = async (context: Context, next) => { + const { actionName, resourceName } = context.action; + if (actionName === 'submit' + && resourceName === 'jobs' + && this.middlewares.length + ) { + return compose([loadJob, ...this.middlewares])(context, next); + } + await next(); + }; + + use(middleware) { + this.middlewares.push(middleware); + } -export default { run(node, input, processor) { return { status: JOB_STATUS.PENDING }; - }, + } resume(node, job, processor) { - job.set('status', JOB_STATUS.RESOLVED); + if (!node.config.actions) { + job.set('status', JOB_STATUS.RESOLVED); + } return job; } + + extend(options: Instruction) { + Object.assign(this, options); + } };