diff --git a/packages/core/client/src/locale/zh_CN.ts b/packages/core/client/src/locale/zh_CN.ts index fe126b55d..5e5f2dc99 100644 --- a/packages/core/client/src/locale/zh_CN.ts +++ b/packages/core/client/src/locale/zh_CN.ts @@ -453,6 +453,7 @@ export default { 'Minutes': '分钟', 'Hours': '小时', 'Days': '天', + 'Weeks': '周', 'Months': '月', 'No repeat': '不重复', @@ -460,13 +461,15 @@ export default { 'By minute': '按分钟', 'By hour': '按小时', - 'By date': '按日(月)', + 'By day': '按天', + 'By week': '按周', 'By month': '按月', - 'By day of week': '按天(周)', 'By field': '数据表字段', 'By custom date': '自定义时间', + 'Advance': '高级模式', + 'End': '结束', 'Trigger context': '触发数据', diff --git a/packages/core/client/src/workflow/nodes/condition.tsx b/packages/core/client/src/workflow/nodes/condition.tsx index ecc619bd0..96bcffce7 100644 --- a/packages/core/client/src/workflow/nodes/condition.tsx +++ b/packages/core/client/src/workflow/nodes/condition.tsx @@ -89,6 +89,7 @@ function CalculationGroup({ value, onChange }) { .ant-select{ width: auto; + min-width: 6em; } `}> diff --git a/packages/core/client/src/workflow/triggers/schedule.tsx b/packages/core/client/src/workflow/triggers/schedule.tsx index 4f9f6488c..81b973003 100644 --- a/packages/core/client/src/workflow/triggers/schedule.tsx +++ b/packages/core/client/src/workflow/triggers/schedule.tsx @@ -32,7 +32,7 @@ const DateFieldsSelect: React.FC = observer((props) => { ); }); -const OnField = ({ value, onChange }) => { +function OnField({ value, onChange }) { const { t } = useTranslation(); const [dir, setDir] = useState(value.offset ? value.offset / Math.abs(value.offset) : 0); @@ -99,11 +99,6 @@ function EndsByField({ value, onChange }) { } function parseCronRule(cron: string) { - if (!cron) { - return { - mode: 0 - } - } const rules = cron.split(/\s+/).slice(1).map(v => v.split('/')); let index = rules.findIndex(rule => rule[0] === '*'); if (index === -1) { @@ -129,6 +124,34 @@ const CronUnits = [ { value: 5, option: 'By day of week', unitText: 'Days', conflict: true }, ]; +const RepeatOptions = [ + { value: 'none', text: 'No repeat' }, + { value: 60_000, text: 'By minute', unitText: 'Minutes' }, + { value: 3600_000, text: 'By hour', unitText: 'Hours' }, + { value: 86400_000, text: 'By day', unitText: 'Days' }, + { value: 604800_000, text: 'By week', unitText: 'Weeks' }, + // { value: 18144_000_000, text: 'By 30 days' }, + { value: 'cron', text: 'Advance', disabled: true } +]; + +function getNumberOption(v) { + const opts = RepeatOptions.filter(option => typeof option.value === 'number').reverse() as any[]; + return opts.find(item => !(v % item.value)); +} + +function getRepeatTypeValue(v) { + switch (typeof v) { + case 'number': + const option = getNumberOption(v); + return option ? option.value : 'none'; + case 'string': + return 'cron'; + default: + break; + } + return 'none'; +} + function getChangedCron({ mode, step }) { const m = mode - 1; const left = [0, ...Array(m).fill(null).map((_, i) => { @@ -146,34 +169,77 @@ function getChangedCron({ mode, step }) { return `${left} ${!step || step == 1 ? '*' : `*/${step}`}${right ? ` ${right}` : ''}`; } -const CronField = ({ value = '', onChange }) => { +function CronField({ value, onChange }) { const { t } = useTranslation(); const cron = parseCronRule(value); const unit = CronUnits[cron.mode - 1]; + + return ( + + onChange(getChangedCron({ step: v, mode: cron.mode }))} + min={1} + addonBefore={t('Every')} + addonAfter={t(unit.unitText)} + /> + ); +} + +function CommonRepeatField({ value, onChange }) { + const { t } = useTranslation(); + const option = getNumberOption(value); + + return ( + onChange(v * option.value)} + min={1} + addonBefore={t('Every')} + addonAfter={t(option.unitText)} + /> + ); +} + +function RepeatField({ value = null, onChange }) { + const { t } = useTranslation(); + const typeValue = getRepeatTypeValue(value); + function onTypeChange(v) { + if (v === 'none') { + onChange(null); + return; + } + if (v === 'cron') { + onChange('0 * * * * *'); + return; + } + onChange(v); + } + return (
- {cron.mode - ? ( - onChange(getChangedCron({ step: v, mode: cron.mode }))} - min={1} - addonBefore={t('Every')} - addonAfter={t(unit.unitText)} - /> - ) + {typeof typeValue === 'number' + ? + : null} + {typeValue === 'cron' + ? : null}
); @@ -192,12 +258,12 @@ const ModeFieldsets = { }, required: true }, - cron: { + repeat: { type: 'string', - name: 'cron', + name: 'repeat', title: '{{t("Repeat mode")}}', 'x-decorator': 'FormItem', - 'x-component': 'CronField', + 'x-component': 'RepeatField', 'x-reactions': [ { target: 'config.endsOn', @@ -260,14 +326,24 @@ const ModeFieldsets = { title: '{{t("Starts on")}}', 'x-decorator': 'FormItem', 'x-component': 'OnField', + 'x-reactions': [ + { + target: 'config.repeat', + fulfill: { + state: { + visible: '{{!!$self.value}}', + }, + } + } + ], required: true }, - cron: { + repeat: { type: 'string', - name: 'cron', + name: 'repeat', title: '{{t("Repeat mode")}}', 'x-decorator': 'FormItem', - 'x-component': 'CronField', + 'x-component': 'RepeatField', 'x-reactions': [ { target: 'config.endsOn', @@ -316,8 +392,9 @@ const ScheduleConfig = () => { setMode(field.value); clearFormGraph('config.collection'); clearFormGraph('config.startsOn'); - clearFormGraph('config.cron'); + clearFormGraph('config.repeat'); clearFormGraph('config.endsOn'); + clearFormGraph('config.limit'); }) }); @@ -350,7 +427,7 @@ const ScheduleConfig = () => { className: css` .ant-select{ width: auto; - min-width: 4em; + min-width: 6em; } .ant-input-number{ @@ -369,7 +446,7 @@ const ScheduleConfig = () => { components={{ DateFieldsSelect, OnField, - CronField, + RepeatField, EndsByField }} /> diff --git a/packages/plugins/workflow/package.json b/packages/plugins/workflow/package.json index d7ecda9a3..ce6360e43 100644 --- a/packages/plugins/workflow/package.json +++ b/packages/plugins/workflow/package.json @@ -15,8 +15,7 @@ "@nocobase/server": "0.7.0-alpha.83", "@nocobase/utils": "0.7.0-alpha.83", "cron-parser": "4.4.0", - "json-templates": "^4.2.0", - "lodash": "^4.17.21" + "json-templates": "^4.2.0" }, "devDependencies": { "@nocobase/test": "0.7.0-alpha.83" diff --git a/packages/plugins/workflow/src/__tests__/triggers/schedule.test.ts b/packages/plugins/workflow/src/__tests__/triggers/schedule.test.ts index a039aa2c8..fbed74530 100644 --- a/packages/plugins/workflow/src/__tests__/triggers/schedule.test.ts +++ b/packages/plugins/workflow/src/__tests__/triggers/schedule.test.ts @@ -22,7 +22,7 @@ describe.skip('workflow > triggers > schedule', () => { afterEach(() => app.stop()); describe('constant mode', () => { - it('no cron configurated', async () => { + it('no repeat configurated', async () => { const workflow = await WorkflowModel.create({ enabled: true, type: 'schedule', @@ -47,7 +47,7 @@ describe.skip('workflow > triggers > schedule', () => { type: 'schedule', config: { mode: 0, - cron: '*/2 * * * * *', + repeat: '*/2 * * * * *', } }); @@ -59,7 +59,7 @@ describe.skip('workflow > triggers > schedule', () => { expect(executions.length).toBe(2); }); - it('on every 2 seconds and limit once', async () => { + it('on every even seconds and limit 1', async () => { const now = new Date(); // NOTE: align to even(0, 2, ...) + 0.5 seconds to start await sleep((2.5 - now.getSeconds() % 2) * 1000 - now.getMilliseconds()); @@ -69,7 +69,28 @@ describe.skip('workflow > triggers > schedule', () => { type: 'schedule', config: { mode: 0, - cron: '*/2 * * * * *', + repeat: '*/2 * * * * *', + limit: 1 + } + }); + + await sleep(5000); + + const executions = await workflow.getExecutions(); + expect(executions.length).toBe(1); + }); + + it('on every 2 seconds after created and limit 1', async () => { + const now = new Date(); + // NOTE: align to even(0, 2, ...) + 0.5 seconds to start + await sleep((2.5 - now.getSeconds() % 2) * 1000 - now.getMilliseconds()); + + const workflow = await WorkflowModel.create({ + enabled: true, + type: 'schedule', + config: { + mode: 0, + repeat: 2000, limit: 1 } }); @@ -90,7 +111,7 @@ describe.skip('workflow > triggers > schedule', () => { type: 'schedule', config: { mode: 0, - cron: `${now.getSeconds()} * * * * *`, + repeat: `${now.getSeconds()} * * * * *`, } }); @@ -111,7 +132,7 @@ describe.skip('workflow > triggers > schedule', () => { type: 'schedule', config: { mode: 0, - cron: `${now.getSeconds()} * * * * *`, + repeat: `${now.getSeconds()} * * * * *`, } }); @@ -120,7 +141,7 @@ describe.skip('workflow > triggers > schedule', () => { type: 'schedule', config: { mode: 0, - cron: `${now.getSeconds()} * * * * *`, + repeat: `${now.getSeconds()} * * * * *`, } }); @@ -168,7 +189,7 @@ describe.skip('workflow > triggers > schedule', () => { expect(execution.context.date).toBe(triggerTime.toISOString()); }); - it('starts on post.createdAt and cron', async () => { + it('starts on post.createdAt and repeat', async () => { const workflow = await WorkflowModel.create({ enabled: true, type: 'schedule', @@ -178,7 +199,7 @@ describe.skip('workflow > triggers > schedule', () => { startsOn: { field: 'createdAt' }, - cron: '*/2 * * * * *' + repeat: '*/2 * * * * *' } }); @@ -201,7 +222,7 @@ describe.skip('workflow > triggers > schedule', () => { expect(d2 - 3500).toBe(startTime.getTime()); }); - it('starts on post.createdAt and cron with endsOn at certain time', async () => { + it('starts on post.createdAt and repeat with endsOn at certain time', async () => { const now = new Date(); await sleep((2.5 - now.getSeconds() % 2) * 1000 - now.getMilliseconds()); const startTime = new Date(); @@ -216,13 +237,12 @@ describe.skip('workflow > triggers > schedule', () => { startsOn: { field: 'createdAt' }, - cron: '*/2 * * * * *', + repeat: '*/2 * * * * *', endsOn: new Date(startTime.getTime() + 2500).toISOString() } }); const post = await PostRepo.create({ values: { title: 't1' }}); - console.log(startTime); await sleep(5000); @@ -232,7 +252,7 @@ describe.skip('workflow > triggers > schedule', () => { expect(d1 - 1500).toBe(startTime.getTime()); }); - it('starts on post.createdAt and cron with endsOn by offset', async () => { + it('starts on post.createdAt and repeat with endsOn by offset', async () => { const workflow = await WorkflowModel.create({ enabled: true, type: 'schedule', @@ -242,7 +262,39 @@ describe.skip('workflow > triggers > schedule', () => { startsOn: { field: 'createdAt' }, - cron: '*/2 * * * * *', + repeat: '*/2 * * * * *', + endsOn: { + field: 'createdAt', + offset: 3 + } + } + }); + + const now = new Date(); + await sleep((2.5 - now.getSeconds() % 2) * 1000 - now.getMilliseconds()); + const startTime = new Date(); + startTime.setMilliseconds(500); + + const post = await PostRepo.create({ values: { title: 't1' }}); + + await sleep(5000); + const executions = await workflow.getExecutions(); + expect(executions.length).toBe(1); + const d1 = Date.parse(executions[0].context.date); + expect(d1 - 1500).toBe(startTime.getTime()); + }); + + it('starts on post.createdAt and repeat by number', async () => { + const workflow = await WorkflowModel.create({ + enabled: true, + type: 'schedule', + config: { + mode: 1, + collection: 'posts', + startsOn: { + field: 'createdAt' + }, + repeat: 2000, endsOn: { field: 'createdAt', offset: 3 diff --git a/packages/plugins/workflow/src/triggers/schedule.ts b/packages/plugins/workflow/src/triggers/schedule.ts index fa70e0e18..e2ae26f33 100644 --- a/packages/plugins/workflow/src/triggers/schedule.ts +++ b/packages/plugins/workflow/src/triggers/schedule.ts @@ -1,5 +1,5 @@ import parser from 'cron-parser'; -import { merge } from 'lodash'; +import { literal, Op } from 'sequelize'; import { Trigger } from '.'; export type ScheduleOnField = string | { @@ -12,7 +12,7 @@ export interface ScheduleTriggerConfig { // trigger mode mode: number; // how to repeat - cron?: string; + repeat?: string | number | null; // limit of repeat times limit?: number; @@ -36,13 +36,19 @@ const ScheduleModes = new Map(); ScheduleModes.set(SCHEDULE_MODE.CONSTANT, { shouldCache(workflow, now) { - const { startsOn, endsOn } = workflow.config; + const { startsOn, endsOn, repeat } = workflow.config; const timestamp = now.getTime(); if (startsOn) { const startTime = Date.parse(startsOn); if (!startTime || (startTime > timestamp + this.cacheCycle)) { return false; } + if (typeof repeat === 'number' + && repeat > this.cacheCycle + && (timestamp - startTime) % repeat > this.cacheCycle + ) { + return false; + } } if (endsOn) { const endTime = Date.parse(endsOn); @@ -54,13 +60,20 @@ ScheduleModes.set(SCHEDULE_MODE.CONSTANT, { return true; }, trigger(workflow, date) { + const { startsOn, endsOn, repeat } = workflow.config; + if (startsOn && typeof repeat === 'number') { + const startTime = Date.parse(startsOn); + if ((startTime - date.getTime()) % repeat) { + return; + } + } return workflow.trigger({ date }); } }); function getDateRangeFilter(on: ScheduleOnField, now: Date, dir: number) { const timestamp = now.getTime(); - const op = dir < 0 ? '$lt' : '$gte'; + const op = dir < 0 ? Op.lt : Op.gte; switch (typeof on) { case 'string': const time = Date.parse(on); @@ -70,6 +83,9 @@ function getDateRangeFilter(on: ScheduleOnField, now: Date, dir: number) { break; case 'object': const { field, offset = 0, unit = 1000 } = on; + if (!field) { + return {}; + } return { [field]: { [op]: new Date(timestamp + offset * unit * dir) } }; default: break; @@ -78,7 +94,7 @@ function getDateRangeFilter(on: ScheduleOnField, now: Date, dir: number) { return {}; } -function getDataOptionTime(data, on, now: Date, dir = 1) { +function getDataOptionTime(data, on, dir = 1) { switch (typeof on) { case 'string': const time = Date.parse(on); @@ -95,39 +111,33 @@ function getHookId(workflow, type) { return `${type}#${workflow.id}`; } +const DialectTimestampFnMap: { [key: string]: Function } = { + postgres(col) { + return `extract(epoch from "${col}")`; + }, + mysql(col) { + return `UNIX_TIMESTAMP(${col})`; + }, + sqlite(col) { + return `unixepoch(${col})`; + } +}; +DialectTimestampFnMap.mariadb = DialectTimestampFnMap.mysql; + ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { on(workflow) { - const { collection, startsOn, endsOn, cron } = workflow.config; + const { collection, startsOn, endsOn, repeat } = workflow.config; const event = `${collection}.afterSave`; const name = getHookId(workflow, event); if (!this.events.has(name)) { // NOTE: toggle cache depends on new date - const listener = (data, options) => { - // check if saved collection data in cache cycle - // in: add workflow to cache - // out: 1. do nothing because if any other data in - // 2. another way is always check all data to match cycle - // by calling: inspect(workflow) - // this may lead to performance issues - // so we can only check single row and only set in if true - // how to check? - // * startsOn only : startsOn in cycle - // * endsOn only : invalid - // * cron only : invalid - // * startsOn and endsOn: equal to only startsOn - // * startsOn and cron : startsOn in cycle and cron in cycle - // * endsOn and cron : invalid - // * all : all rules effect - // * none : invalid - // this means, startsOn and cron should be present at least one - // and no startsOn equals run on cron, and could ends on endsOn, - // this will be a little wired, only means the end date should use collection field. + const listener = async (data, options) => { const now = new Date(); now.setMilliseconds(0); const timestamp = now.getTime(); - const startTime = getDataOptionTime(data, startsOn, now); - const endTime = getDataOptionTime(data, endsOn, now, -1); - if (!startTime && !cron) { + const startTime = getDataOptionTime(data, startsOn); + const endTime = getDataOptionTime(data, endsOn, -1); + if (!startTime && !repeat) { return; } if (startTime && startTime > timestamp + this.cacheCycle) { @@ -136,7 +146,13 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { if (endTime && endTime <= timestamp) { return; } - if (!cronInCycle.call(this, workflow, now)) { + if (!nextInCycle.call(this, workflow, now)) { + return; + } + if (typeof repeat === 'number' + && repeat > this.cacheCycle + && (timestamp - startTime) % repeat > this.cacheCycle + ) { return; } console.log('set cache', now); @@ -160,25 +176,32 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { }, async shouldCache(workflow, now) { - const { startsOn, endsOn, collection } = workflow.config; + const { startsOn, endsOn, repeat, collection } = workflow.config; const starts = getDateRangeFilter(startsOn, now, -1); - if (!starts) { + if (!starts || !Object.keys(starts).length) { return false; } const ends = getDateRangeFilter(endsOn, now, 1); if (!ends) { return false; } - const filter = merge(starts, ends); - // if neither startsOn nor endsOn is provided - if (!Object.keys(filter).length) { - // consider as invalid - return false; + + const conditions: any[] = [starts, ends].filter(item => Boolean(Object.keys(item).length)); + // when repeat is number, means repeat after startsOn + // (now - startsOn) % repeat <= cacheCycle + const tsFn = DialectTimestampFnMap[this.db.options.dialect]; + if (repeat + && typeof repeat === 'number' + && repeat > this.cacheCycle + && tsFn + ) { + const uts = now.getTime(); + conditions.push(literal(`mod(${uts} - ${tsFn(startsOn.field)} * 1000, ${repeat}) < ${this.cacheCycle}`)); } - const repo = this.db.getCollection(collection).repository; - const count = await repo.count({ - filter + const { model } = this.db.getCollection(collection); + const count = await model.count({ + where: { [Op.and]: conditions } }); return Boolean(count); @@ -189,7 +212,7 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { collection, startsOn, endsOn, - cron + repeat } = workflow.config; if (typeof startsOn !== 'object') { @@ -199,22 +222,28 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { const timestamp = date.getTime(); const startTimestamp = timestamp - (startsOn.offset ?? 0) * (startsOn.unit ?? 1000); - let filter - if (!cron) { + const conditions = []; + if (!repeat) { // startsOn exactly equal to now in 1s - filter = { + conditions.push({ [startsOn.field]: { - $gte: new Date(startTimestamp), - $lt: new Date(startTimestamp + 1000) + [Op.gte]: new Date(startTimestamp), + [Op.lt]: new Date(startTimestamp + 1000) } - }; + }); } else { // startsOn not after now - filter = { + conditions.push({ [startsOn.field]: { - $lt: new Date(startTimestamp) + [Op.lt]: new Date(startTimestamp) } - }; + }); + + const tsFn = DialectTimestampFnMap[this.db.options.dialect]; + if (typeof repeat === 'number' && tsFn) { + const uts = timestamp; + conditions.push(literal(`mod(${uts} - floor(${tsFn(startsOn.field)}) * 1000, ${repeat}) = 0`)); + } switch (typeof endsOn) { case 'string': @@ -224,21 +253,28 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { } break; case 'object': - filter[endsOn.field] = { - $gte: new Date(timestamp - (endsOn.offset ?? 0) * (endsOn.unit ?? 1000) + 1000) - }; + if (endsOn.field) { + conditions.push({ + [endsOn.field]: { + [Op.gte]: new Date(timestamp - (endsOn.offset ?? 0) * (endsOn.unit ?? 1000) + 1000) + } + }); + } break; default: break; } } - const repo = this.db.getCollection(collection).repository; - const instances = await repo.find({ - filter + + const { model } = this.db.getCollection(collection); + const instances = await model.findAll({ + where: { + [Op.and]: conditions + } }); if (instances.length) { - console.log(workflow.id, 'trigger at', date); + console.log(instances.length, 'rows trigger at', date); } instances.forEach(item => { @@ -251,20 +287,27 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, { }); -function cronInCycle(this: ScheduleTrigger, workflow, now: Date): boolean { - const { cron } = workflow.config; - // no cron means no need to rerun +function nextInCycle(this: ScheduleTrigger, workflow, now: Date): boolean { + const { repeat } = workflow.config; + // no repeat means no need to rerun // but if in current cycle, should be put in cache - // no cron but in current cycle means startsOn or endsOn has been configured + // no repeat but in current cycle means startsOn has been configured // so we need to more info to determine if necessary config items - if (!cron) { + if (!repeat) { return true; } + switch (typeof repeat) { + case 'string': + break; + default: + return true; + } + const currentDate = new Date(now); currentDate.setMilliseconds(-1); const timestamp = now.getTime(); - const interval = parser.parseExpression(cron, { currentDate }); + const interval = parser.parseExpression(repeat, { currentDate }); let next = interval.next(); // NOTE: cache all workflows will be matched in current cycle @@ -278,8 +321,8 @@ export default class ScheduleTrigger implements Trigger { static CacheRules = [ // ({ enabled }) => enabled, ({ config, executed }) => config.limit ? executed < config.limit : true, - ({ config }) => ['cron', 'startsOn'].some(key => config[key]), - cronInCycle, + ({ config }) => ['repeat', 'startsOn'].some(key => config[key]), + nextInCycle, function(workflow, now) { const { mode } = workflow.config; const modeHandlers = ScheduleModes.get(mode); @@ -289,17 +332,17 @@ export default class ScheduleTrigger implements Trigger { static TriggerRules = [ ({ config, executed }) => config.limit ? executed < config.limit : true, - ({ config }) => ['cron', 'startsOn'].some(key => config[key]), + ({ config }) => ['repeat', 'startsOn'].some(key => config[key]), function (workflow, now) { - const { cron } = workflow.config; - if (!cron) { + const { repeat } = workflow.config; + if (typeof repeat !== 'string') { return true; } const currentDate = new Date(now); currentDate.setMilliseconds(-1); const timestamp = now.getTime(); - const interval = parser.parseExpression(cron, { currentDate }); + const interval = parser.parseExpression(repeat, { currentDate }); let next = interval.next(); if (next.getTime() === timestamp) {