refactor: Use trigger queue for webhook handler
This commit is contained in:
@@ -4,7 +4,7 @@ import isEmpty from 'lodash/isEmpty';
|
||||
|
||||
import Flow from '../models/flow';
|
||||
import { processTrigger } from '../services/trigger';
|
||||
import actionQueue from '../queues/action';
|
||||
import triggerQueue from '../queues/trigger';
|
||||
import globalVariable from './global-variable';
|
||||
import QuotaExceededError from '../errors/quote-exceeded';
|
||||
import {
|
||||
@@ -59,32 +59,31 @@ export default async (
|
||||
}
|
||||
|
||||
for (const triggerItem of reversedTriggerItems) {
|
||||
const { executionId } = await processTrigger({
|
||||
if (testRun) {
|
||||
await processTrigger({
|
||||
flowId,
|
||||
stepId: triggerStep.id,
|
||||
triggerItem,
|
||||
testRun,
|
||||
});
|
||||
|
||||
if (testRun) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const nextStep = await triggerStep.getNextStep();
|
||||
const jobName = `${executionId}-${nextStep.id}`;
|
||||
|
||||
const jobPayload = {
|
||||
flowId,
|
||||
executionId,
|
||||
stepId: nextStep.id,
|
||||
};
|
||||
const jobName = `${triggerStep.id}-${triggerItem.meta.internalId}`;
|
||||
|
||||
const jobOptions = {
|
||||
removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
|
||||
removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
|
||||
};
|
||||
|
||||
await actionQueue.add(jobName, jobPayload, jobOptions);
|
||||
const jobPayload = {
|
||||
flowId,
|
||||
stepId: triggerStep.id,
|
||||
triggerItem,
|
||||
};
|
||||
|
||||
await triggerQueue.add(jobName, jobPayload, jobOptions);
|
||||
}
|
||||
|
||||
return response.status(204);
|
||||
|
Reference in New Issue
Block a user