Merge pull request #1644 from automatisch/sync-webhook

feat(webhook/catch-raw-webhook): add sync support and custom response

packages/backend/src/apps/webhook/actions/index.js

@@ -0,0 +1,3 @@
+import respondWith from './respond-with/index.js';
+
+export default [respondWith];

packages/backend/src/apps/webhook/actions/respond-with/index.js

@@ -0,0 +1,38 @@
+import defineAction from '../../../../helpers/define-action.js';
+
+export default defineAction({
+  name: 'Respond with',
+  key: 'respondWith',
+  description: 'Respond with defined JSON body.',
+  arguments: [
+    {
+      label: 'Status code',
+      key: 'statusCode',
+      type: 'string',
+      required: true,
+      variables: true,
+      value: '200',
+    },
+    {
+      label: 'JSON body',
+      key: 'stringifiedJsonBody',
+      type: 'string',
+      required: true,
+      description: 'The content of the JSON body. It must be a valid JSON.',
+      variables: true,
+    },
+  ],
+
+  async run($) {
+    const parsedStatusCode = parseInt($.step.parameters.statusCode, 10);
+    const stringifiedJsonBody = $.step.parameters.stringifiedJsonBody;
+    const parsedJsonBody = JSON.parse(stringifiedJsonBody);
+
+    $.setActionItem({
+      raw: {
+        body: parsedJsonBody,
+        statusCode: parsedStatusCode,
+      },
+    });
+  },
+});

packages/backend/src/apps/webhook/index.js

@@ -1,4 +1,5 @@
 import defineApp from '../../helpers/define-app.js';
+import actions from './actions/index.js';
 import triggers from './triggers/index.js';
 
 export default defineApp({
@@ -10,5 +11,6 @@ export default defineApp({
   baseUrl: '',
   apiBaseUrl: '',
   primaryColor: '0059F7',
+  actions,
   triggers,
 });

packages/backend/src/apps/webhook/triggers/catch-raw-webhook/index.js

@@ -7,7 +7,20 @@ export default defineTrigger({
   key: 'catchRawWebhook',
   type: 'webhook',
   showWebhookUrl: true,
-  description: 'Triggers when the webhook receives a request.',
+  description:
+    'Triggers (immediately if configured) when the webhook receives a request.',
+  arguments: [
+    {
+      label: 'Wait until flow is done',
+      key: 'workSynchronously',
+      type: 'dropdown',
+      required: true,
+      options: [
+        { label: 'Yes', value: true },
+        { label: 'No', value: false },
+      ],
+    },
+  ],
 
   async run($) {
     const dataItem = {

packages/backend/src/controllers/webhooks/handler-sync-by-flow-id.js

@@ -0,0 +1,31 @@
+import Flow from '../../models/flow.js';
+import logger from '../../helpers/logger.js';
+import handlerSync from '../../helpers/webhook-handler-sync.js';
+
+export default async (request, response) => {
+  const computedRequestPayload = {
+    headers: request.headers,
+    body: request.body,
+    query: request.query,
+    params: request.params,
+  };
+
+  logger.debug(`Handling incoming webhook request at ${request.originalUrl}.`);
+  logger.debug(JSON.stringify(computedRequestPayload, null, 2));
+
+  const flowId = request.params.flowId;
+  const flow = await Flow.query().findById(flowId).throwIfNotFound();
+  const triggerStep = await flow.getTriggerStep();
+
+  if (triggerStep.appKey !== 'webhook') {
+    const connection = await triggerStep.$relatedQuery('connection');
+
+    if (!(await connection.verifyWebhook(request))) {
+      return response.sendStatus(401);
+    }
+  }
+
+  await handlerSync(flowId, request, response);
+
+  response.sendStatus(204);
+};

packages/backend/src/helpers/webhook-handler-sync.js

@@ -0,0 +1,86 @@
+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) {
+        // we send the response only if it's not sent yet. This allows us to early respond from the flow.
+        response.status(actionExecutionStep.dataOut.statusCode);
+        response.send(actionExecutionStep.dataOut.body);
+      }
+    }
+  }
+
+  return response;
+};

packages/backend/src/models/step.js

@@ -103,6 +103,10 @@ class Step extends Base {
       return `/webhooks/connections/${this.connectionId}`;
     }
 
+    if (this.parameters.workSynchronously) {
+      return `/webhooks/flows/${this.flowId}/sync`;
+    }
+
     return `/webhooks/flows/${this.flowId}`;
   }
 

packages/backend/src/routes/webhooks.js

@@ -3,6 +3,7 @@ import multer from 'multer';
 
 import appConfig from '../config/app.js';
 import webhookHandlerByFlowId from '../controllers/webhooks/handler-by-flow-id.js';
+import webhookHandlerSyncByFlowId from '../controllers/webhooks/handler-sync-by-flow-id.js';
 import webhookHandlerByConnectionIdAndRefValue from '../controllers/webhooks/handler-by-connection-id-and-ref-value.js';
 
 const router = Router();
@@ -46,6 +47,7 @@ createRouteHandler(
   '/connections/:connectionId',
   webhookHandlerByConnectionIdAndRefValue
 );
+createRouteHandler('/flows/:flowId/sync', webhookHandlerSyncByFlowId);
 createRouteHandler('/flows/:flowId', webhookHandlerByFlowId);
 createRouteHandler('/:flowId', webhookHandlerByFlowId);