refactor(plugin-workflow): refactor apis (#3267)

* refactor(plugin-workflow): refactor apis

* fix(plugin-workflow-parallel): fix import in test cases

* fix(plugin-workflow): fix some module import source

* fix(plugin-workflow): move manual table acl to manual plugin

* fix(plugin-workflow-manual): fix folder typo
This commit is contained in:
Junyi 2023-12-27 13:55:48 +08:00 committed by GitHub
parent 57c7dd3e95
commit 8ee8ab7d6d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
60 changed files with 222 additions and 261 deletions

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const aggregateInstruction = new AggregateInstruction();
workflow.instructions.register(aggregateInstruction.type, aggregateInstruction);
workflow.registerInstruction('aggregate', AggregateInstruction);
}
}

View File

@ -28,7 +28,7 @@ export default class extends Instruction {
const result = await repo.aggregate({
...options,
method: aggregators[aggregator],
transaction: processor.transaction,
// transaction: processor.transaction,
});
return {

View File

@ -4,11 +4,8 @@ import WorkflowPlugin from '@nocobase/plugin-workflow';
import AggregateInstruction from './AggregateInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('aggregate', new AggregateInstruction(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('aggregate', AggregateInstruction);
}
}

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const delayInstruction = new DelayInstruction();
workflow.instructions.register(delayInstruction.type, delayInstruction);
workflow.registerInstruction('delay', DelayInstruction);
}
}

View File

@ -16,15 +16,15 @@ interface DelayConfig {
export default class extends Instruction {
timers: Map<number, NodeJS.Timeout> = new Map();
constructor(public plugin: WorkflowPlugin) {
super(plugin);
constructor(public workflow: WorkflowPlugin) {
super(workflow);
plugin.app.on('afterStart', this.load);
plugin.app.on('beforeStop', this.unload);
workflow.app.on('afterStart', this.load);
workflow.app.on('beforeStop', this.unload);
}
load = async () => {
const { model } = this.plugin.db.getCollection('jobs');
const { model } = this.workflow.app.db.getCollection('jobs');
const jobs = (await model.findAll({
where: {
status: JOB_STATUS.PENDING,
@ -79,7 +79,7 @@ export default class extends Instruction {
job.execution = await job.getExecution();
}
if (job.execution.status === EXECUTION_STATUS.STARTED) {
this.plugin.resume(job);
this.workflow.resume(job);
}
if (this.timers.get(job.id)) {
this.timers.delete(job.id);

View File

@ -4,11 +4,8 @@ import WorkflowPlugin from '@nocobase/plugin-workflow';
import DelayInstruction from './DelayInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('delay', new DelayInstruction(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('delay', DelayInstruction);
}
}

View File

@ -5,15 +5,12 @@ import { ExpressionField } from './expression-field';
import { DynamicCalculation } from './DynamicCalculation';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
this.db.registerFieldTypes({
expression: ExpressionField,
});
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('dynamic-calculation', new DynamicCalculation(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('dynamic-calculation', DynamicCalculation);
}
}

View File

@ -6,7 +6,6 @@ import { NAMESPACE, useLang } from '../locale';
export default class extends Trigger {
title = `{{t("Form event", { ns: "${NAMESPACE}" })}}`;
type = 'form';
description = `{{t("Event triggers when submitted a workflow bound form action.", { ns: "${NAMESPACE}" })}}`;
fieldset = {
collection: {

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const formTrigger = new FormTrigger();
workflow.triggers.register(formTrigger.type, formTrigger);
workflow.registerTrigger('form', FormTrigger);
}
}

View File

@ -5,10 +5,10 @@ import { Model, modelAssociationByKey } from '@nocobase/database';
import WorkflowPlugin, { Trigger, WorkflowModel, toJSON } from '@nocobase/plugin-workflow';
export default class extends Trigger {
constructor(plugin: WorkflowPlugin) {
super(plugin);
constructor(workflow: WorkflowPlugin) {
super(workflow);
plugin.app.resourcer.use(this.middleware);
workflow.app.resourcer.use(this.middleware);
}
async triggerAction(context, next) {
@ -58,7 +58,7 @@ export default class extends Trigger {
};
const triggers = triggerWorkflows.split(',').map((trigger) => trigger.split('!'));
const workflowRepo = this.plugin.db.getRepository('workflows');
const workflowRepo = this.workflow.db.getRepository('workflows');
const workflows = await workflowRepo.find({
filter: {
key: triggers.map((trigger) => trigger[0]),
@ -95,11 +95,11 @@ export default class extends Trigger {
appends,
});
}
this.plugin.trigger(workflow, { data: toJSON(payload), ...userInfo });
this.workflow.trigger(workflow, { data: toJSON(payload), ...userInfo });
});
} else {
const data = trigger[1] ? get(values, trigger[1]) : values;
this.plugin.trigger(workflow, {
this.workflow.trigger(workflow, {
data,
...userInfo,
});

View File

@ -4,11 +4,8 @@ import WorkflowPlugin from '@nocobase/plugin-workflow';
import FormTrigger from './FormTrigger';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.triggers.register('form', new FormTrigger(workflowPlugin));
}
}

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const loopInstruction = new LoopInstruction();
workflow.instructions.register(loopInstruction.type, loopInstruction);
workflow.registerInstruction('loop', LoopInstruction);
}
}

View File

@ -4,11 +4,8 @@ import { default as WorkflowPlugin } from '@nocobase/plugin-workflow';
import LoopInstruction from './LoopInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('loop', new LoopInstruction(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('loop', LoopInstruction);
}
}

View File

@ -1,6 +1,6 @@
import Database from '@nocobase/database';
import { Application } from '@nocobase/server';
import { BRANCH_INDEX, EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
import { EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import Plugin from '..';
@ -404,7 +404,7 @@ describe('workflow > instructions > loop', () => {
const n2 = await workflow.createNode({
type: 'loop',
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
upstreamId: n1.id,
config: {
target: 0,
@ -442,7 +442,7 @@ describe('workflow > instructions > loop', () => {
const n2 = await workflow.createNode({
type: 'loop',
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
upstreamId: n1.id,
config: {
target: 2,

View File

@ -85,8 +85,8 @@ function getMode(mode) {
export default class extends Instruction {
formTypes = new Registry<FormHandler>();
constructor(public plugin: WorkflowPlugin) {
super(plugin);
constructor(public workflow: WorkflowPlugin) {
super(workflow);
initFormTypes(this);
}
@ -103,7 +103,7 @@ export default class extends Instruction {
});
// NOTE: batch create users jobs
const UserJobModel = processor.options.plugin.db.getModel('users_jobs');
const UserJobModel = this.workflow.app.db.getModel('users_jobs');
await UserJobModel.bulkCreate(
assignees.map((userId) => ({
userId,
@ -114,7 +114,7 @@ export default class extends Instruction {
status: JOB_STATUS.PENDING,
})),
{
transaction: processor.transaction,
// transaction: processor.transaction,
},
);
@ -125,13 +125,13 @@ export default class extends Instruction {
// NOTE: check all users jobs related if all done then continue as parallel
const { assignees = [], mode } = node.config as ManualConfig;
const UserJobModel = processor.options.plugin.db.getModel('users_jobs');
const UserJobModel = this.workflow.app.db.getModel('users_jobs');
const distribution = await UserJobModel.count({
where: {
jobId: job.id,
},
group: ['status'],
transaction: processor.transaction,
// transaction: processor.transaction,
});
const submitted = distribution.reduce(

View File

@ -3,16 +3,14 @@ import actions from '@nocobase/actions';
import { HandlerType } from '@nocobase/resourcer';
import WorkflowPlugin, { JOB_STATUS } from '@nocobase/plugin-workflow';
import jobsCollection from './collecions/jobs';
import usersCollection from './collecions/users';
import usersJobsCollection from './collecions/users_jobs';
import jobsCollection from './collections/jobs';
import usersCollection from './collections/users';
import usersJobsCollection from './collections/users_jobs';
import { submit } from './actions';
import ManualInstruction from './ManualInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
this.app.db.collection(usersJobsCollection);
this.app.db.extendCollection(usersCollection);
@ -41,8 +39,9 @@ export default class extends Plugin {
},
});
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('manual', new ManualInstruction(workflowPlugin));
this.app.acl.allow('users_jobs', ['list', 'get', 'submit'], 'loggedIn');
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('manual', ManualInstruction);
}
}

View File

@ -81,7 +81,7 @@ export async function submit(context: Context, next) {
await handler.call(instruction, userJob, forms[formKey], processor);
}
await userJob.save({ transaction: processor.transaction });
await userJob.save();
await processor.exit();

View File

@ -2,7 +2,7 @@ import { Processor } from '@nocobase/plugin-workflow';
import ManualInstruction from '../ManualInstruction';
export default async function (this: ManualInstruction, instance, { collection }, processor: Processor) {
const repo = this.plugin.db.getRepository(collection);
const repo = this.workflow.db.getRepository(collection);
if (!repo) {
throw new Error(`collection ${collection} for create data on manual node not found`);
}
@ -18,6 +18,6 @@ export default async function (this: ManualInstruction, instance, { collection }
context: {
executionId: processor.execution.id,
},
transaction: processor.transaction,
// transaction: processor.transaction,
});
}

View File

@ -2,7 +2,7 @@ import { Processor } from '@nocobase/plugin-workflow';
import ManualInstruction from '../ManualInstruction';
export default async function (this: ManualInstruction, instance, { collection, filter = {} }, processor: Processor) {
const repo = this.plugin.db.getRepository(collection);
const repo = this.workflow.db.getRepository(collection);
if (!repo) {
throw new Error(`collection ${collection} for update data on manual node not found`);
}
@ -18,6 +18,6 @@ export default async function (this: ManualInstruction, instance, { collection,
context: {
executionId: processor.execution.id,
},
transaction: processor.transaction,
// transaction: processor.transaction,
});
}

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get(WorkflowPlugin);
const parallelInstruction = new ParallelInstruction();
workflow.instructions.register(parallelInstruction.type, parallelInstruction);
workflow.registerInstruction('parallel', ParallelInstruction);
}
}

View File

@ -110,7 +110,7 @@ export default class extends Instruction {
});
if (job.status === JOB_STATUS.PENDING) {
await job.save({ transaction: processor.transaction });
await job.save();
return processor.exit();
}

View File

@ -4,11 +4,8 @@ import WorkflowPlugin from '@nocobase/plugin-workflow';
import ParallelInstruction from './ParallelInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('parallel', new ParallelInstruction(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('parallel', ParallelInstruction);
}
}

View File

@ -1,6 +1,6 @@
import Database from '@nocobase/database';
import { Application } from '@nocobase/server';
import { BRANCH_INDEX, EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
import { EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import Plugin from '..';
@ -453,7 +453,7 @@ describe('workflow > instructions > parallel', () => {
const n2 = await workflow.createNode({
type: 'parallel',
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
upstreamId: n1.id,
});
@ -521,7 +521,7 @@ describe('workflow > instructions > parallel', () => {
const n4 = await workflow.createNode({
type: 'echo',
upstreamId: n3.id,
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
});
const n5 = await workflow.createNode({

View File

@ -1,8 +1,11 @@
import { ArrayItems } from '@formily/antd-v5';
import { defaultFieldNames } from '@nocobase/client';
import { Instruction, WorkflowVariableInput, WorkflowVariableJSON } from '@nocobase/plugin-workflow/client';
import {
Instruction,
WorkflowVariableInput,
WorkflowVariableJSON,
defaultFieldNames,
} from '@nocobase/plugin-workflow/client';
import { NAMESPACE } from '../locale';

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const requestInstruction = new RequestInstruction();
workflow.instructions.register(requestInstruction.type, requestInstruction);
workflow.registerInstruction('request', RequestInstruction);
}
}

View File

@ -4,11 +4,8 @@ import WorkflowPlugin from '@nocobase/plugin-workflow';
import RequestInstruction from './RequestInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('request', new RequestInstruction(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('request', RequestInstruction);
}
}

View File

@ -65,7 +65,7 @@ export default class extends Instruction {
})
.finally(() => {
processor.logger.info(`request (#${node.id}) response received, status: ${job.get('status')}`);
this.plugin.resume(job);
this.workflow.resume(job);
});
processor.logger.info(`request (#${node.id}) sent to "${config.url}", waiting for response...`);

View File

@ -1,6 +1,6 @@
import { css, defaultFieldNames } from '@nocobase/client';
import { css } from '@nocobase/client';
import { Instruction, WorkflowVariableRawTextArea } from '@nocobase/plugin-workflow/client';
import { Instruction, WorkflowVariableRawTextArea, defaultFieldNames } from '@nocobase/plugin-workflow/client';
import { NAMESPACE } from '../locale';

View File

@ -13,7 +13,6 @@ export default class extends Plugin {
// You can get and modify the app instance here
async load() {
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const sqlInstruction = new SQLInstruction();
workflow.instructions.register(sqlInstruction.type, sqlInstruction);
workflow.registerInstruction('sql', SQLInstruction);
}
}

View File

@ -4,11 +4,8 @@ import WorkflowPlugin from '@nocobase/plugin-workflow';
import SQLInstruction from './SQLInstruction';
export default class extends Plugin {
workflow: WorkflowPlugin;
async load() {
const workflowPlugin = this.app.getPlugin('workflow') as WorkflowPlugin;
this.workflow = workflowPlugin;
workflowPlugin.instructions.register('sql', new SQLInstruction(workflowPlugin));
const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);
workflowPlugin.registerInstruction('sql', SQLInstruction);
}
}

View File

@ -11,7 +11,7 @@ export default class extends Instruction {
}
const result = await sequelize.query(sql, {
transaction: processor.transaction,
// transaction: processor.transaction,
// plain: true,
// model: db.getCollection(node.config.collection).model
});

View File

@ -45,6 +45,26 @@ export default class extends Plugin {
}));
};
registerTrigger(type: string, trigger: Trigger | { new (): Trigger }) {
if (typeof trigger === 'function') {
this.triggers.register(type, new trigger());
} else if (trigger) {
this.triggers.register(type, trigger);
} else {
throw new TypeError('invalid trigger type to register');
}
}
registerInstruction(type: string, instruction: Instruction | { new (): Instruction }) {
if (typeof instruction === 'function') {
this.instructions.register(type, new instruction());
} else if (instruction instanceof Instruction) {
this.instructions.register(type, instruction);
} else {
throw new TypeError('invalid instruction type to register');
}
}
async load() {
this.addRoutes();
this.addScopes();
@ -57,15 +77,15 @@ export default class extends Plugin {
aclSnippet: 'pm.workflow.workflows',
});
this.triggers.register('collection', new CollectionTrigger());
this.triggers.register('schedule', new ScheduleTrigger());
this.registerTrigger('collection', CollectionTrigger);
this.registerTrigger('schedule', ScheduleTrigger);
this.instructions.register('calculation', new CalculationInstruction());
this.instructions.register('condition', new ConditionInstruction());
this.instructions.register('query', new QueryInstruction());
this.instructions.register('create', new CreateInstruction());
this.instructions.register('update', new UpdateInstruction());
this.instructions.register('destroy', new DestroyInstruction());
this.registerInstruction('calculation', CalculationInstruction);
this.registerInstruction('condition', ConditionInstruction);
this.registerInstruction('query', QueryInstruction);
this.registerInstruction('create', CreateInstruction);
this.registerInstruction('update', UpdateInstruction);
this.registerInstruction('destroy', DestroyInstruction);
}
addScopes() {

View File

@ -1,11 +1,11 @@
import { SchemaInitializerItemType, defaultFieldNames } from '@nocobase/client';
import { SchemaInitializerItemType } from '@nocobase/client';
import { Evaluator, evaluators, getOptions } from '@nocobase/evaluators/client';
import { RadioWithTooltip } from '../components/RadioWithTooltip';
import { ValueBlock } from '../components/ValueBlock';
import { renderEngineReference } from '../components/renderEngineReference';
import { NAMESPACE, lang } from '../locale';
import { BaseTypeSets, WorkflowVariableTextArea } from '../variable';
import { BaseTypeSets, WorkflowVariableTextArea, defaultFieldNames } from '../variable';
import { Instruction } from '.';
export default class extends Instruction {

View File

@ -30,7 +30,13 @@ import { JobStatusOptionsMap } from '../constants';
import { useGetAriaLabelOfAddButton } from '../hooks/useGetAriaLabelOfAddButton';
import { lang } from '../locale';
import useStyles from '../style';
import { VariableOption, VariableOptions } from '../variable';
import { UseVariableOptions, VariableOption } from '../variable';
export type NodeAvailableContext = {
workflow: object;
upstream: object;
branchIndex: number;
};
export abstract class Instruction {
title: string;
@ -43,10 +49,10 @@ export abstract class Instruction {
scope?: { [key: string]: any };
components?: { [key: string]: any };
Component?(props): JSX.Element;
useVariables?(node, options?): VariableOption;
useScopeVariables?(node, options?): VariableOptions;
useVariables?(node, options?: UseVariableOptions): VariableOption;
useScopeVariables?(node, options?): VariableOption[];
useInitializers?(node): SchemaInitializerItemType | null;
isAvailable?(ctx: object): boolean;
isAvailable?(ctx: NodeAvailableContext): boolean;
}
function useUpdateAction() {
@ -362,7 +368,7 @@ export function NodeDefaultView(props) {
className: 'workflow-node-config-button',
},
},
[`${instruction.type}_${data.id}`]: {
[data.id]: {
type: 'void',
title: (
<div

View File

@ -101,14 +101,6 @@ const workflowFieldset = {
type: 'object',
'x-component': 'fieldset',
properties: {
// NOTE: not to expose this option for now, because hard to track errors
// useTransaction: {
// type: 'boolean',
// title: `{{ t("Use transaction", { ns: "${NAMESPACE}" }) }}`,
// description: `{{ t("Data operation nodes in workflow will run in a same transaction until any interruption. Any failure will cause data rollback, and will also rollback the history of the execution.", { ns: "${NAMESPACE}" }) }}`,
// 'x-decorator': 'FormItem',
// 'x-component': 'Checkbox',
// },
deleteExecutionOnStatus: {
type: 'array',
title: `{{ t("Auto delete history when execution is on end status", { ns: "${NAMESPACE}" }) }}`,

View File

@ -22,7 +22,6 @@ const collectionModeOptions = [
export default class extends Trigger {
title = `{{t("Collection event", { ns: "${NAMESPACE}" })}}`;
type = 'collection';
description = `{{t("Event will be triggered on collection data row created, updated or deleted.", { ns: "${NAMESPACE}" })}}`;
fieldset = {
collection: {

View File

@ -7,6 +7,7 @@ import { ISchema, useForm } from '@formily/react';
import {
ActionContextProvider,
FieldNames,
FormProvider,
SchemaComponent,
SchemaInitializerItemType,
@ -24,7 +25,7 @@ import { useFlowContext } from '../FlowContext';
import { DrawerDescription } from '../components/DrawerDescription';
import { NAMESPACE, lang } from '../locale';
import useStyles from '../style';
import { VariableOptions } from '../variable';
import { UseVariableOptions, VariableOption } from '../variable';
function useUpdateConfigAction() {
const form = useForm();
@ -54,10 +55,9 @@ function useUpdateConfigAction() {
export abstract class Trigger {
title: string;
type: string;
description?: string;
// group: string;
useVariables?(config: any, options?): VariableOptions;
useVariables?(config: Record<string, any>, options?: UseVariableOptions): VariableOption[];
fieldset: { [key: string]: ISchema };
view?: ISchema;
scope?: { [key: string]: any };

View File

@ -9,7 +9,6 @@ import { SCHEDULE_MODE } from './constants';
export default class extends Trigger {
title = `{{t("Schedule event", { ns: "${NAMESPACE}" })}}`;
type = 'schedule';
description = `{{t("Event will be scheduled and triggered base on time conditions.", { ns: "${NAMESPACE}" })}}`;
fieldset = {
config: {

View File

@ -11,21 +11,26 @@ export type VariableOption = {
key?: string;
value?: string;
label?: string;
children?: VariableOptions;
children?: VariableOption[] | null;
[key: string]: any;
};
export type VariableOptions = VariableOption[] | null;
export type VariableDataType =
| string
| 'boolean'
| 'number'
| 'string'
| 'date'
| {
type: string;
options?: { entity?: boolean; collection?: string };
type: 'reference';
options: {
collection: string;
multiple?: boolean;
entity?: boolean;
};
}
| ((field: any, appends?: string[]) => boolean);
| ((field: any) => boolean);
export type OptionsOfUseVariableOptions = {
export type UseVariableOptions = {
types?: VariableDataType[];
fieldNames?: {
label?: string;
@ -34,7 +39,6 @@ export type OptionsOfUseVariableOptions = {
};
appends?: string[] | null;
depth?: number;
current?: any;
};
export const defaultFieldNames = { label: 'label', value: 'value', children: 'children' } as const;
@ -42,7 +46,7 @@ export const defaultFieldNames = { label: 'label', value: 'value', children: 'ch
export const nodesOptions = {
label: `{{t("Node result", { ns: "${NAMESPACE}" })}}`,
value: '$jobsMapByNodeKey',
useOptions(options: OptionsOfUseVariableOptions) {
useOptions(options: UseVariableOptions) {
const { instructions } = usePlugin(WorkflowPlugin);
const current = useNodeContext();
const upstreams = useAvailableUpstreams(current);
@ -61,7 +65,7 @@ export const nodesOptions = {
export const triggerOptions = {
label: `{{t("Trigger variables", { ns: "${NAMESPACE}" })}}`,
value: '$context',
useOptions(options: OptionsOfUseVariableOptions) {
useOptions(options: UseVariableOptions) {
const { triggers } = usePlugin(WorkflowPlugin);
const { workflow } = useFlowContext();
const trigger = triggers.get(workflow.type);
@ -72,7 +76,7 @@ export const triggerOptions = {
export const scopeOptions = {
label: `{{t("Scope variables", { ns: "${NAMESPACE}" })}}`,
value: '$scopes',
useOptions(options: OptionsOfUseVariableOptions) {
useOptions(options: UseVariableOptions & { current: any }) {
const { fieldNames = defaultFieldNames, current } = options;
const { instructions } = usePlugin(WorkflowPlugin);
const source = useNodeContext();
@ -98,7 +102,7 @@ export const scopeOptions = {
export const systemOptions = {
label: `{{t("System variables", { ns: "${NAMESPACE}" })}}`,
value: '$system',
useOptions({ types, fieldNames = defaultFieldNames }: OptionsOfUseVariableOptions) {
useOptions({ types, fieldNames = defaultFieldNames }: UseVariableOptions) {
return [
...(!types || types.includes('date')
? [
@ -136,13 +140,12 @@ export const BaseTypeSets = {
// { type: 'reference', options: { collection: 'attachments', multiple: false } }
// { type: 'reference', options: { collection: 'myExpressions', entity: false } }
function matchFieldType(field, type): boolean {
const inputType = typeof type;
if (inputType === 'string') {
function matchFieldType(field, type: VariableDataType): boolean {
if (typeof type === 'string') {
return BaseTypeSets[type]?.has(field.interface);
}
if (inputType === 'object' && type.type === 'reference') {
if (typeof type === 'object' && type.type === 'reference') {
if (isAssociationField(field)) {
return (
type.options?.entity && (field.collectionName === type.options?.collection || type.options?.collection === '*')
@ -157,7 +160,7 @@ function matchFieldType(field, type): boolean {
}
}
if (inputType === 'function') {
if (typeof type === 'function') {
return type(field);
}
@ -232,7 +235,7 @@ function useOptions(scope, opts) {
};
}
export function useWorkflowVariableOptions(options: OptionsOfUseVariableOptions = {}) {
export function useWorkflowVariableOptions(options: UseVariableOptions = {}) {
const fieldNames = Object.assign({}, defaultFieldNames, options.fieldNames ?? {});
const opts = Object.assign(options, { fieldNames });
const result = [

View File

@ -12,10 +12,10 @@ import Processor from './Processor';
import initActions from './actions';
import { EXECUTION_STATUS } from './constants';
import initFunctions, { CustomFunction } from './functions';
import type Trigger from './triggers';
import Trigger from './triggers';
import CollectionTrigger from './triggers/CollectionTrigger';
import ScheduleTrigger from './triggers/ScheduleTrigger';
import type Instruction from './instructions';
import Instruction from './instructions';
import CalculationInstruction from './instructions/CalculationInstruction';
import ConditionInstruction from './instructions/ConditionInstruction';
import CreateInstruction from './instructions/CreateInstruction';
@ -112,29 +112,45 @@ export default class WorkflowPlugin extends Plugin {
}
};
initTriggers<T extends Trigger>(more: { [key: string]: T | { new (p: Plugin): T } } = {}) {
const { triggers } = this;
triggers.register('collection', new CollectionTrigger(this));
triggers.register('schedule', new ScheduleTrigger(this));
for (const [name, trigger] of Object.entries(more)) {
triggers.register(name, typeof trigger === 'function' ? new trigger(this) : trigger);
registerTrigger<T extends Trigger>(type: string, trigger: T | { new (p: Plugin): T }) {
if (typeof trigger === 'function') {
this.triggers.register(type, new trigger(this));
} else if (trigger) {
this.triggers.register(type, trigger);
} else {
throw new Error('invalid trigger type to register');
}
}
initInstructions<T extends Instruction>(more: { [key: string]: T | { new (p: Plugin): T } } = {}) {
const { instructions } = this;
registerInstruction<T extends Instruction>(type: string, instruction: T | { new (p: Plugin): T }) {
if (typeof instruction === 'function') {
this.instructions.register(type, new instruction(this));
} else if (instruction) {
this.instructions.register(type, instruction);
} else {
throw new Error('invalid instruction type to register');
}
}
instructions.register('calculation', new CalculationInstruction(this));
instructions.register('condition', new ConditionInstruction(this));
instructions.register('create', new CreateInstruction(this));
instructions.register('destroy', new DestroyInstruction(this));
instructions.register('query', new QueryInstruction(this));
instructions.register('update', new UpdateInstruction(this));
private initTriggers<T extends Trigger>(more: { [key: string]: T | { new (p: Plugin): T } } = {}) {
this.registerTrigger('collection', CollectionTrigger);
this.registerTrigger('schedule', ScheduleTrigger);
for (const [name, trigger] of Object.entries(more)) {
this.registerTrigger(name, trigger);
}
}
private initInstructions<T extends Instruction>(more: { [key: string]: T | { new (p: Plugin): T } } = {}) {
this.registerInstruction('calculation', CalculationInstruction);
this.registerInstruction('condition', ConditionInstruction);
this.registerInstruction('create', CreateInstruction);
this.registerInstruction('destroy', DestroyInstruction);
this.registerInstruction('query', QueryInstruction);
this.registerInstruction('update', UpdateInstruction);
for (const [name, instruction] of Object.entries({ ...more })) {
instructions.register(name, typeof instruction === 'function' ? new instruction(this) : instruction);
this.registerInstruction(name, instruction);
}
}
@ -171,7 +187,6 @@ export default class WorkflowPlugin extends Plugin {
actions: ['workflows:list'],
});
this.app.acl.allow('users_jobs', ['list', 'get', 'submit'], 'loggedIn');
this.app.acl.allow('workflows', ['trigger'], 'loggedIn');
await db.import({

View File

@ -25,8 +25,6 @@ export default class Processor {
logger: Logger;
transaction?: Transaction;
nodes: FlowNodeModel[] = [];
nodesMap = new Map<number, FlowNodeModel>();
jobsMap = new Map<number, JobModel>();
@ -67,35 +65,18 @@ export default class Processor {
});
}
private async getTransaction() {
if (!this.execution.workflow.options?.useTransaction) {
return;
}
const { options } = this;
// @ts-ignore
return options.transaction && !options.transaction.finished
? options.transaction
: await options.plugin.db.sequelize.transaction();
}
public async prepare() {
const { execution } = this;
if (!execution.workflow) {
execution.workflow = await execution.getWorkflow();
}
const transaction = await this.getTransaction();
this.transaction = transaction;
const nodes = await execution.workflow.getNodes();
this.makeNodes(nodes);
const jobs = await execution.getJobs({
order: [['id', 'ASC']],
transaction,
});
this.makeJobs(jobs);
@ -125,13 +106,6 @@ export default class Processor {
await this.recall(node, job);
}
private async commit() {
// @ts-ignore
if (this.transaction && (!this.options.transaction || this.options.transaction.finished)) {
await this.transaction.commit();
}
}
private async exec(instruction: Runner, node: FlowNodeModel, prevJob) {
let job;
try {
@ -224,10 +198,9 @@ export default class Processor {
async exit(s?: number) {
if (typeof s === 'number') {
const status = (<typeof Processor>this.constructor).StatusMap[s] ?? Math.sign(s);
await this.execution.update({ status }, { transaction: this.transaction });
await this.execution.update({ status });
}
this.logger.info(`execution (${this.execution.id}) exiting with status ${this.execution.status}`);
await this.commit();
return null;
}
@ -237,22 +210,15 @@ export default class Processor {
const { model } = database.getCollection('jobs');
let job;
if (payload instanceof model) {
job = await payload.save({ transaction: this.transaction });
job = await payload.save();
} else if (payload.id) {
job = await model.findByPk(payload.id);
await job.update(payload, {
transaction: this.transaction,
});
await job.update(payload);
} else {
job = await model.create(
{
...payload,
executionId: this.execution.id,
},
{
transaction: this.transaction,
},
);
job = await model.create({
...payload,
executionId: this.execution.id,
});
}
this.jobsMap.set(job.id, job);

View File

@ -1,11 +1,11 @@
import Database from '@nocobase/database';
import { MockDatabase } from '@nocobase/database';
import { MockServer } from '@nocobase/test';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import { BRANCH_INDEX, EXECUTION_STATUS, JOB_STATUS } from '../constants';
import { EXECUTION_STATUS, JOB_STATUS } from '../constants';
describe('workflow > Processor', () => {
let app: MockServer;
let db: Database;
let db: MockDatabase;
let PostRepo;
let WorkflowModel;
let workflow;
@ -251,13 +251,13 @@ describe('workflow > Processor', () => {
const n2 = await workflow.createNode({
type: 'echo',
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
upstreamId: n1.id,
});
await workflow.createNode({
type: 'echo',
branchIndex: BRANCH_INDEX.ON_FALSE,
branchIndex: 0,
upstreamId: n1.id,
});
@ -283,7 +283,7 @@ describe('workflow > Processor', () => {
const n2 = await workflow.createNode({
type: 'prompt',
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
upstreamId: n1.id,
});
@ -323,7 +323,7 @@ describe('workflow > Processor', () => {
const n2 = await workflow.createNode({
type: 'prompt->error',
branchIndex: BRANCH_INDEX.ON_TRUE,
branchIndex: 1,
upstreamId: n1.id,
});

View File

@ -1,7 +1,8 @@
import Database from '@nocobase/database';
import { Application } from '@nocobase/server';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import { BRANCH_INDEX, EXECUTION_STATUS, JOB_STATUS } from '../../constants';
import { EXECUTION_STATUS, JOB_STATUS } from '../../constants';
import { BRANCH_INDEX } from '../../instructions/ConditionInstruction';
describe('workflow > instructions > condition', () => {
let app: Application;

View File

@ -54,12 +54,11 @@ export async function destroy(context: Context, next) {
}
export async function revision(context: Context, next) {
const plugin = context.app.getPlugin('workflow') as Plugin;
const { db } = context;
const plugin = context.app.getPlugin(Plugin);
const repository = utils.getRepositoryFromParams(context);
const { filterByTk, filter = {}, values = {} } = context.action.params;
context.body = await db.sequelize.transaction(async (transaction) => {
context.body = await context.db.sequelize.transaction(async (transaction) => {
const origin = await repository.findOne({
filterByTk,
filter,
@ -140,7 +139,7 @@ export async function revision(context: Context, next) {
}
export async function sync(context: Context, next) {
const plugin = context.app.getPlugin('workflow');
const plugin = context.app.getPlugin(Plugin);
const repository = utils.getRepositoryFromParams(context);
const { filterByTk, filter = {} } = context.action.params;

View File

@ -35,11 +35,6 @@ export default function () {
required: true,
defaultValue: {},
},
{
type: 'boolean',
name: 'useTransaction',
// defaultValue: true,
},
{
type: 'hasMany',
name: 'nodes',

View File

@ -8,7 +8,7 @@ export const EXECUTION_STATUS = {
CANCELED: -4,
REJECTED: -5,
RETRY_NEEDED: -6,
};
} as const;
export const JOB_STATUS = {
PENDING: 0,
@ -19,10 +19,4 @@ export const JOB_STATUS = {
CANCELED: -4,
REJECTED: -5,
RETRY_NEEDED: -6,
};
export const BRANCH_INDEX = {
DEFAULT: null,
ON_TRUE: 1,
ON_FALSE: 0,
};
} as const;

View File

@ -7,6 +7,12 @@ import type { FlowNodeModel, JobModel } from '../types';
type Comparer = (a: any, b: any) => boolean;
export const BRANCH_INDEX = {
DEFAULT: null,
ON_TRUE: 1,
ON_FALSE: 0,
} as const;
export const calculators = new Registry<Comparer>();
// built-in functions

View File

@ -15,7 +15,7 @@ export class CreateInstruction extends Instruction {
context: {
executionId: processor.execution.id,
},
transaction: processor.transaction,
// transaction: processor.transaction,
});
let result = created;
@ -28,7 +28,7 @@ export class CreateInstruction extends Instruction {
result = await repository.findOne({
filterByTk: created[model.primaryKeyAttribute],
appends: Array.from(includeFields),
transaction: processor.transaction,
// transaction: processor.transaction,
});
}

View File

@ -14,7 +14,7 @@ export class DestroyInstruction extends Instruction {
context: {
executionId: processor.execution.id,
},
transaction: processor.transaction,
// transaction: processor.transaction,
});
return {

View File

@ -33,7 +33,7 @@ export class QueryInstruction extends Instruction {
.filter((item) => item.field)
.map((item) => `${item.direction?.toLowerCase() === 'desc' ? '-' : ''}${item.field}`),
appends,
transaction: processor.transaction,
// transaction: processor.transaction,
});
if (failOnEmpty && (multiple ? !result.length : !result)) {

View File

@ -14,7 +14,7 @@ export class UpdateInstruction extends Instruction {
context: {
executionId: processor.execution.id,
},
transaction: processor.transaction,
// transaction: processor.transaction,
});
return {

View File

@ -18,7 +18,7 @@ export type Runner = (node: FlowNodeModel, input: any, processor: Processor) =>
// what should a instruction do?
// - base on input and context, do any calculations or system call (io), and produce a result or pending.
export abstract class Instruction {
constructor(public plugin: Plugin) {}
constructor(public workflow: Plugin) {}
abstract run(node: FlowNodeModel, input: any, processor: Processor): InstructionResult;

View File

@ -85,7 +85,7 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M
// TODO: `result.toJSON()` throws error
const json = toJSON(result);
this.plugin.trigger(
this.workflow.trigger(
workflow,
{ data: json },
{
@ -98,7 +98,7 @@ export default class CollectionTrigger extends Trigger {
events = new Map();
on(workflow: WorkflowModel) {
const { db } = this.plugin.app;
const { db } = this.workflow.app;
const { collection, mode } = workflow.config;
const Collection = db.getCollection(collection);
if (!Collection) {
@ -125,7 +125,7 @@ export default class CollectionTrigger extends Trigger {
}
off(workflow: WorkflowModel) {
const { db } = this.plugin.app;
const { db } = this.workflow.app;
const { collection, mode } = workflow.config;
const Collection = db.getCollection(collection);
if (!Collection) {

View File

@ -103,7 +103,7 @@ ScheduleModes.set(SCHEDULE_MODE.CONSTANT, {
}
}
this.plugin.trigger(workflow, { date: now });
this.workflow.trigger(workflow, { date: now });
return 1;
},
@ -201,7 +201,7 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
this.setCache(workflow);
};
this.events.set(name, listener);
this.plugin.app.db.on(event, listener);
this.workflow.app.db.on(event, listener);
},
off(workflow) {
@ -211,12 +211,12 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
if (this.events.has(name)) {
const listener = this.events.get(name);
this.events.delete(name);
this.plugin.app.db.off(event, listener);
this.workflow.app.db.off(event, listener);
}
},
async shouldCache(workflow, now) {
const { db } = this.plugin.app;
const { db } = this.workflow.app;
const { startsOn, endsOn, repeat, collection } = workflow.config;
const timestamp = now.getTime();
@ -305,7 +305,7 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
},
});
const tsFn = DialectTimestampFnMap[this.plugin.app.db.options.dialect];
const tsFn = DialectTimestampFnMap[this.workflow.app.db.options.dialect];
if (typeof repeat === 'number' && tsFn) {
const modExp = fn(
'MOD',
@ -343,7 +343,7 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
});
}
const repo = this.plugin.app.db.getRepository(collection);
const repo = this.workflow.app.db.getRepository(collection);
const instances = await repo.find({
where: {
[Op.and]: conditions,
@ -357,7 +357,7 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
});
instances.forEach((item) => {
this.plugin.trigger(workflow, {
this.workflow.trigger(workflow, {
date: now,
data: item.toJSON(),
});
@ -423,10 +423,10 @@ export default class ScheduleTrigger extends Trigger {
// caching workflows in range, default to 1min
cacheCycle = 60_000;
constructor(plugin: Plugin) {
super(plugin);
constructor(workflow: Plugin) {
super(workflow);
plugin.app.on('beforeStop', () => {
workflow.app.on('beforeStop', () => {
if (this.timer) {
clearInterval(this.timer);
}
@ -434,7 +434,7 @@ export default class ScheduleTrigger extends Trigger {
}
init() {
if (this.plugin.app.getPlugin('multi-app-share-collection')?.enabled && this.plugin.app.name !== 'main') {
if (this.workflow.app.getPlugin('multi-app-share-collection')?.enabled && this.workflow.app.name !== 'main') {
return;
}
@ -474,7 +474,7 @@ export default class ScheduleTrigger extends Trigger {
async onTick(now) {
// NOTE: trigger workflows in sequence when sqlite due to only one transaction
const isSqlite = this.plugin.app.db.options.dialect === 'sqlite';
const isSqlite = this.workflow.app.db.options.dialect === 'sqlite';
return Array.from(this.cache.values()).reduce(
(prev, workflow) => {
if (!this.shouldTrigger(workflow, now)) {
@ -491,7 +491,7 @@ export default class ScheduleTrigger extends Trigger {
}
async reload() {
const WorkflowRepo = this.plugin.app.db.getRepository('workflows');
const WorkflowRepo = this.workflow.app.db.getRepository('workflows');
const workflows = await WorkflowRepo.find({
filter: { enabled: true, type: 'schedule' },
});
@ -510,7 +510,7 @@ export default class ScheduleTrigger extends Trigger {
const should = await this.shouldCache(workflow, now);
if (should) {
this.plugin.getLogger(workflow.id).info('caching scheduled workflow will run in next minute');
this.workflow.getLogger(workflow.id).info('caching scheduled workflow will run in next minute');
}
this.setCache(workflow, !should);

View File

@ -3,7 +3,7 @@ import type Plugin from '../Plugin';
import type { WorkflowModel } from '../types';
export abstract class Trigger {
constructor(public readonly plugin: Plugin) {}
constructor(public readonly workflow: Plugin) {}
abstract on(workflow: WorkflowModel): void;
abstract off(workflow: WorkflowModel): void;
duplicateConfig?(workflow: WorkflowModel, options: Transactionable): object | Promise<object>;

View File

@ -9,9 +9,6 @@ export default class ExecutionModel extends Model {
declare title: string;
declare context: any;
declare status: number;
// NOTE: this duplicated column is for transaction in preparing cycle from workflow
declare useTransaction: boolean;
declare transaction: string;
declare createdAt: Date;
declare updatedAt: Date;

View File

@ -19,7 +19,6 @@ export default class WorkflowModel extends Model {
declare description?: string;
declare type: string;
declare config: any;
declare useTransaction: boolean;
declare executed: number;
declare createdAt: Date;