File size: 2,723 Bytes
3206347 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
import isEmpty from 'lodash/isEmpty.js';
import Flow from '../models/flow.js';
import { processTrigger } from '../services/trigger.js';
import { processAction } from '../services/action.js';
import globalVariable from './global-variable.js';
import QuotaExceededError from '../errors/quote-exceeded.js';
export default async (flowId, request, response) => {
const flow = await Flow.query().findById(flowId).throwIfNotFound();
const user = await flow.$relatedQuery('user');
const testRun = !flow.active;
const quotaExceeded = !testRun && !(await user.isAllowedToRunFlows());
if (quotaExceeded) {
throw new QuotaExceededError();
}
const [triggerStep, ...actionSteps] = await flow
.$relatedQuery('steps')
.withGraphFetched('connection')
.orderBy('position', 'asc');
const app = await triggerStep.getApp();
const isWebhookApp = app.key === 'webhook';
if (testRun && !isWebhookApp) {
return response.status(404);
}
const connection = await triggerStep.$relatedQuery('connection');
const $ = await globalVariable({
flow,
connection,
app,
step: triggerStep,
testRun,
request,
});
const triggerCommand = await triggerStep.getTriggerCommand();
await triggerCommand.run($);
const reversedTriggerItems = $.triggerOutput.data.reverse();
// This is the case when we filter out the incoming data
// in the run method of the webhook trigger.
// In this case, we don't want to process anything.
if (isEmpty(reversedTriggerItems)) {
return response.status(204);
}
// set default status, but do not send it yet!
response.status(204);
for (const triggerItem of reversedTriggerItems) {
const { executionId } = await processTrigger({
flowId,
stepId: triggerStep.id,
triggerItem,
testRun,
});
if (testRun) {
// in case of testing, we do not process the whole process.
continue;
}
for (const actionStep of actionSteps) {
const { executionStep: actionExecutionStep } = await processAction({
flowId: flow.id,
stepId: actionStep.id,
executionId,
});
if (actionStep.key === 'respondWith' && !response.headersSent) {
const { headers, statusCode, body } = actionExecutionStep.dataOut;
// we set the custom response headers
if (headers) {
for (const [key, value] of Object.entries(headers)) {
if (key) {
response.set(key, value);
}
}
}
// we send the response only if it's not sent yet. This allows us to early respond from the flow.
response.status(statusCode);
response.send(body);
}
}
}
return response;
};
|