import path from 'path'; import LRUCache from 'lru-cache'; import winston from 'winston'; import { Op } from '@nocobase/database'; import { Plugin } from '@nocobase/server'; import { Registry } from '@nocobase/utils'; import { createLogger, getLoggerFilePath, getLoggerLevel, Logger, LoggerOptions } from '@nocobase/logger'; import Processor from './Processor'; import initActions from './actions'; import { EXECUTION_STATUS } from './constants'; import initFunctions, { CustomFunction } from './functions'; import type Trigger from './triggers'; import CollectionTrigger from './triggers/CollectionTrigger'; import ScheduleTrigger from './triggers/ScheduleTrigger'; import type Instruction from './instructions'; import CalculationInstruction from './instructions/CalculationInstruction'; import ConditionInstruction from './instructions/ConditionInstruction'; import CreateInstruction from './instructions/CreateInstruction'; import DestroyInstruction from './instructions/DestroyInstruction'; import QueryInstruction from './instructions/QueryInstruction'; import UpdateInstruction from './instructions/UpdateInstruction'; import type { ExecutionModel, JobModel, WorkflowModel } from './types'; type ID = number | string; type Pending = [ExecutionModel, JobModel?]; type CachedEvent = [WorkflowModel, any, { context?: any }]; export default class WorkflowPlugin extends Plugin { instructions: Registry = new Registry(); triggers: Registry = new Registry(); functions: Registry = new Registry(); private ready = false; private executing: Promise | null = null; private pending: Pending[] = []; private events: CachedEvent[] = []; private loggerCache: LRUCache; getLogger(workflowId: ID): Logger { const now = new Date(); const date = `${now.getFullYear()}-${`0${now.getMonth() + 1}`.slice(-2)}-${`0${now.getDate()}`.slice(-2)}`; const key = `${date}-${workflowId}}`; if (this.loggerCache.has(key)) { return this.loggerCache.get(key); } const logger = createLogger({ transports: [ ...(process.env.NODE_ENV !== 'production' ? ['console'] : []), new winston.transports.File({ filename: getLoggerFilePath('workflows', date, `${workflowId}.log`), level: getLoggerLevel(), }), ], } as LoggerOptions); this.loggerCache.set(key, logger); return logger; } onBeforeSave = async (instance: WorkflowModel, options) => { const Model = instance.constructor; if (instance.enabled) { instance.set('current', true); } else if (!instance.current) { const count = await Model.count({ where: { key: instance.key, }, transaction: options.transaction, }); if (!count) { instance.set('current', true); } } if (!instance.changed('enabled') || !instance.enabled) { return; } const previous = await Model.findOne({ where: { key: instance.key, current: true, id: { [Op.ne]: instance.id, }, }, transaction: options.transaction, }); if (previous) { // NOTE: set to `null` but not `false` will not violate the unique index await previous.update( { enabled: false, current: null }, { transaction: options.transaction, hooks: false, }, ); this.toggle(previous, false); } }; initTriggers(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); } } initInstructions(more: { [key: string]: T | { new (p: Plugin): T } } = {}) { const { instructions } = this; 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)); for (const [name, instruction] of Object.entries({ ...more })) { instructions.register(name, typeof instruction === 'function' ? new instruction(this) : instruction); } } async load() { const { db, options } = this; initActions(this); this.initTriggers(options.triggers); this.initInstructions(options.instructions); initFunctions(this, options.functions); this.loggerCache = new LRUCache({ max: 20, updateAgeOnGet: true, dispose(logger) { (logger).end(); }, }); this.app.acl.registerSnippet({ name: `pm.${this.name}.workflows`, actions: [ 'workflows:*', 'workflows.nodes:*', 'executions:list', 'executions:get', 'flow_nodes:update', 'flow_nodes:destroy', ], }); this.app.acl.registerSnippet({ name: 'ui.*', actions: ['workflows:list'], }); this.app.acl.allow('users_jobs', ['list', 'get', 'submit'], 'loggedIn'); this.app.acl.allow('workflows', ['trigger'], 'loggedIn'); await db.import({ directory: path.resolve(__dirname, 'collections'), }); this.db.addMigrations({ namespace: this.name, directory: path.resolve(__dirname, 'migrations'), context: { plugin: this, }, }); db.on('workflows.beforeSave', this.onBeforeSave); db.on('workflows.afterSave', (model: WorkflowModel) => this.toggle(model)); db.on('workflows.afterDestroy', (model: WorkflowModel) => this.toggle(model, false)); // [Life Cycle]: // * load all workflows in db // * add all hooks for enabled workflows // * add hooks for create/update[enabled]/delete workflow to add/remove specific hooks this.app.on('beforeStart', async () => { const collection = db.getCollection('workflows'); const workflows = await collection.repository.find({ filter: { enabled: true }, }); workflows.forEach((workflow: WorkflowModel) => { this.toggle(workflow); }); }); this.app.on('afterStart', () => { this.app.setMaintainingMessage('check for not started executions'); this.ready = true; // check for not started executions this.dispatch(); }); 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); }); this.ready = false; if (this.events.length) { await this.prepare(); } if (this.executing) { await this.executing; } }); } toggle(workflow: WorkflowModel, enable?: boolean) { const type = workflow.get('type'); const trigger = this.triggers.get(type); if (typeof enable !== 'undefined' ? enable : workflow.get('enabled')) { // NOTE: remove previous listener if config updated const prev = workflow.previous(); if (prev.config) { trigger.off({ ...workflow.get(), ...prev }); } trigger.on(workflow); } else { trigger.off(workflow); } } public trigger(workflow: WorkflowModel, context: object, options: { context?: any } = {}): void { const logger = this.getLogger(workflow.id); if (!this.ready) { logger.warn(`app is not ready, event of workflow ${workflow.id} will be ignored`); logger.debug(`ignored event data:`, { data: context }); return; } // `null` means not to trigger if (context == null) { logger.warn(`workflow ${workflow.id} event data context is null, event will be ignored`); return; } this.events.push([workflow, context, options]); logger.info(`new event triggered, now events: ${this.events.length}`); logger.debug(`event data:`, { data: context, }); if (this.events.length > 1) { return; } // NOTE: no await for quick return setTimeout(this.prepare); } public async resume(job) { if (!job.execution) { job.execution = await job.getExecution(); } this.getLogger(job.execution.workflowId).info( `execution (${job.execution.id}) resuming from job (${job.id}) added to pending list`, ); this.pending.push([job.execution, job]); this.dispatch(); } public createProcessor(execution: ExecutionModel, options = {}): Processor { return new Processor(execution, { ...options, plugin: this }); } private async createExecution(event: CachedEvent): Promise { const [workflow, context, options] = event; if (options.context?.executionId) { // NOTE: no transaction here for read-uncommitted execution const existed = await workflow.countExecutions({ where: { id: options.context.executionId, }, }); if (existed) { this.getLogger(workflow.id).warn( `workflow ${workflow.id} has already been triggered in same execution (${options.context.executionId}), and newly triggering will be skipped.`, ); return null; } } const execution = await this.db.sequelize.transaction(async (transaction) => { const execution = await workflow.createExecution( { context, key: workflow.key, status: EXECUTION_STATUS.QUEUEING, }, { transaction }, ); await workflow.increment(['executed', 'allExecuted'], { transaction }); // NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment if (this.db.options.dialect !== 'postgres') { await workflow.reload({ transaction }); } await (workflow.constructor).update( { allExecuted: workflow.allExecuted, }, { where: { key: workflow.key, }, transaction, }, ); execution.workflow = workflow; return execution; }); this.getLogger(workflow.id).info(`execution of workflow ${workflow.id} created as ${execution.id}`); // NOTE: cache first execution for most cases if (!this.executing && !this.pending.length) { this.pending.push([execution]); } return execution; } private prepare = async () => { const event = this.events.shift(); if (!event) { this.getLogger('dispatcher').warn(`events queue is empty, no need to prepare`); return; } const logger = this.getLogger(event[0].id); logger.info(`preparing execution for event`); try { await this.createExecution(event); } catch (err) { logger.error(`failed to create execution: ${err.message}`, err); // this.events.push(event); // NOTE: retry will cause infinite loop } if (this.events.length) { await this.prepare(); } else { this.dispatch(); } }; private dispatch() { if (!this.ready) { this.getLogger('dispatcher').warn(`app is not ready, new dispatching will be ignored`); return; } if (this.executing) { this.getLogger('dispatcher').warn(`workflow executing is not finished, new dispatching will be ignored`); return; } this.executing = (async () => { let next: Pending | null = null; // resuming has high priority if (this.pending.length) { next = this.pending.shift() as Pending; this.getLogger(next[0].workflowId).info(`pending execution (${next[0].id}) ready to process`); } else { const execution = (await this.db.getRepository('executions').findOne({ filter: { status: EXECUTION_STATUS.QUEUEING, 'workflow.enabled': true, 'workflow.id': { [Op.not]: null, }, }, appends: ['workflow'], sort: 'createdAt', })) as ExecutionModel; if (execution) { this.getLogger(execution.workflowId).info(`execution (${execution.id}) fetched from db`); next = [execution]; } } if (next) { await this.process(...next); } this.executing = null; if (next) { this.dispatch(); } })(); } private async process(execution: ExecutionModel, job?: JobModel) { if (execution.status === EXECUTION_STATUS.QUEUEING) { await execution.update({ status: EXECUTION_STATUS.STARTED }); } const processor = this.createProcessor(execution); this.getLogger(execution.workflowId).info(`execution (${execution.id}) ${job ? 'resuming' : 'starting'}...`); try { await (job ? processor.resume(job) : processor.start()); this.getLogger(execution.workflowId).info( `execution (${execution.id}) finished with status: ${execution.status}`, ); if (execution.status && execution.workflow.options?.deleteExecutionOnStatus?.includes(execution.status)) { await execution.destroy(); } } catch (err) { this.getLogger(execution.workflowId).error(`execution (${execution.id}) error: ${err.message}`, err); } } }