Merge pull request #1332 from automatisch/webhook-body

feat: Add run method to webhook triggers

packages/backend/src/apps/flowers-software/triggers/new-activity/index.ts

@@ -1,3 +1,4 @@
+import Crypto from 'crypto';
 import isEmpty from 'lodash/isEmpty';
 import defineTrigger from '../../../../helpers/define-trigger';
 import webhookFilters from '../../common/webhook-filters';
@@ -19,6 +20,17 @@ export default defineTrigger({
     },
   ],
 
+  async run($) {
+    const dataItem = {
+      raw: $.request.body,
+      meta: {
+        internalId: Crypto.randomUUID(),
+      },
+    };
+
+    $.pushTriggerItem(dataItem);
+  },
+
   async testRun($) {
     const lastExecutionStep = await $.getLastExecutionStep();
 

packages/backend/src/apps/gitlab/triggers/confidential-issue-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -19,6 +20,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.confidential_issues_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.confidential_issues_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/confidential-note-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -19,6 +20,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.confidential_note_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.confidential_note_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/deployment-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.deployment_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.deployment_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/feature-flag-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.feature_flag_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.feature_flag_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/issue-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.issues_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.issues_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/job-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -17,6 +18,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.job_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.job_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/lib.ts

@@ -22,6 +22,17 @@ export const projectArgumentDescriptor = {
   },
 };
 
+export const getRunFn = async ($: IGlobalVariable) => {
+  const dataItem = {
+    raw: $.request.body,
+    meta: {
+      internalId: Crypto.randomUUID(),
+    },
+  };
+
+  $.pushTriggerItem(dataItem);
+};
+
 export const getTestRunFn =
   (eventData: IJSONObject) => ($: IGlobalVariable) => {
     /*

packages/backend/src/apps/gitlab/triggers/merge-request-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.merge_requests_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.merge_requests_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/note-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.note_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.note_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/pipeline-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.pipeline_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.pipeline_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/push-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -54,6 +55,7 @@ export const triggerDescriptor: IRawTrigger = {
     branchFilterStrategyArgumentDescriptor,
     pushEventsBranchFilterArgumentDescriptor,
   ],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.push_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/release-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -17,6 +18,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.releases_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.releases_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/tag-push-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.tag_push_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.tag_push_events),
   unregisterHook,

packages/backend/src/apps/gitlab/triggers/wiki-page-event/index.ts

@@ -3,6 +3,7 @@ import defineTrigger from '../../../../helpers/define-trigger';
 import { GITLAB_EVENT_TYPE } from '../types';
 import {
   getRegisterHookFn,
+  getRunFn,
   getTestRunFn,
   projectArgumentDescriptor,
   unregisterHook,
@@ -18,6 +19,7 @@ export const triggerDescriptor: IRawTrigger = {
   key: GITLAB_EVENT_TYPE.wiki_page_events,
   type: 'webhook',
   arguments: [projectArgumentDescriptor],
+  run: ($) => getRunFn($),
   testRun: getTestRunFn(data),
   registerHook: getRegisterHookFn(GITLAB_EVENT_TYPE.wiki_page_events),
   unregisterHook,

packages/backend/src/apps/twilio/triggers/receive-sms/index.ts

@@ -1,4 +1,5 @@
 import { URLSearchParams } from 'node:url';
+import Crypto from 'crypto';
 import isEmpty from 'lodash/isEmpty';
 import defineTrigger from '../../../../helpers/define-trigger';
 import fetchMessages from './fetch-messages';
@@ -37,6 +38,17 @@ export default defineTrigger({
   useSingletonWebhook: true,
   singletonWebhookRefValueParameter: 'phoneNumberSid',
 
+  async run($) {
+    const dataItem = {
+      raw: $.request.body,
+      meta: {
+        internalId: Crypto.randomUUID(),
+      },
+    };
+
+    $.pushTriggerItem(dataItem);
+  },
+
   async testRun($) {
     await fetchMessages($);
 

packages/backend/src/apps/typeform/triggers/new-entry/index.ts

@@ -1,3 +1,4 @@
+import Crypto from 'crypto';
 import appConfig from '../../../../config/app';
 import defineTrigger from '../../../../helpers/define-trigger';
 
@@ -27,6 +28,17 @@ export default defineTrigger({
     },
   ],
 
+  async run($) {
+    const dataItem = {
+      raw: $.request.body,
+      meta: {
+        internalId: Crypto.randomUUID(),
+      },
+    };
+
+    $.pushTriggerItem(dataItem);
+  },
+
   async testRun($) {
     const { data: form } = await $.http.get(
       `/forms/${$.step.parameters.formId}`

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

@@ -1,3 +1,4 @@
+import Crypto from 'crypto';
 import isEmpty from 'lodash/isEmpty';
 import defineTrigger from '../../../../helpers/define-trigger';
 
@@ -8,6 +9,21 @@ export default defineTrigger({
   showWebhookUrl: true,
   description: 'Triggers when the webhook receives a request.',
 
+  async run($) {
+    const dataItem = {
+      raw: {
+        headers: $.request.headers,
+        body: $.request.body,
+        query: $.request.query,
+      },
+      meta: {
+        internalId: Crypto.randomUUID(),
+      },
+    };
+
+    $.pushTriggerItem(dataItem);
+  },
+
   async testRun($) {
     const lastExecutionStep = await $.getLastExecutionStep();
 

packages/backend/src/helpers/global-variable.ts

@@ -102,7 +102,9 @@ const globalVariable = async (
 
       $.triggerOutput.data.push(triggerItem);
 
-      if ($.execution.testRun) {
+      const isWebhookApp = app.key === 'webhook';
+
+      if ($.execution.testRun && !isWebhookApp) {
         // early exit after receiving one item as it is enough for test execution
         throw new EarlyExitError();
       }
@@ -145,7 +147,9 @@ const globalVariable = async (
   }
 
   const lastInternalIds =
-    testRun || (flow && step?.isAction) ? [] : await flow?.lastInternalIds(2000);
+    testRun || (flow && step?.isAction)
+      ? []
+      : await flow?.lastInternalIds(2000);
 
   const isAlreadyProcessed = (internalId: string) => {
     return lastInternalIds?.includes(internalId);

packages/backend/src/helpers/webhook-handler.ts

@@ -1,10 +1,10 @@
-import Crypto from 'node:crypto';
 import { Response } from 'express';
-import { IRequest, ITriggerItem } from '@automatisch/types';
+import { IRequest } from '@automatisch/types';
+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 {
@@ -12,18 +12,12 @@ import {
   REMOVE_AFTER_7_DAYS_OR_50_JOBS,
 } from './remove-job-configuration';
 
-export default async (flowId: string, request: IRequest, response: Response) => {
-  // in case it's our built-in generic webhook trigger
-  let computedRequestPayload = {
-    headers: request.headers,
-    body: request.body,
-    query: request.query,
-  };
-
-  const flow = await Flow.query()
-    .findById(flowId)
-    .throwIfNotFound();
-
+export default async (
+  flowId: string,
+  request: IRequest,
+  response: Response
+) => {
+  const flow = await Flow.query().findById(flowId).throwIfNotFound();
   const user = await flow.$relatedQuery('user');
 
   const testRun = !flow.active;
@@ -37,48 +31,60 @@ export default async (flowId: string, request: IRequest, response: Response) =>
   const app = await triggerStep.getApp();
   const isWebhookApp = app.key === 'webhook';
 
-  if ((testRun && !isWebhookApp)) {
+  if (testRun && !isWebhookApp) {
     return response.status(404);
   }
 
-  // in case trigger type is 'webhook'
-  if (!isWebhookApp) {
-    computedRequestPayload = request.body;
-  }
+  const connection = await triggerStep.$relatedQuery('connection');
 
-  const triggerItem: ITriggerItem = {
-    raw: computedRequestPayload,
-    meta: {
-      internalId: Crypto.randomUUID(),
-    },
-  };
-
-  const { executionId } = await processTrigger({
-    flowId,
-    stepId: triggerStep.id,
-    triggerItem,
+  const $ = await globalVariable({
+    flow,
+    connection,
+    app,
+    step: triggerStep,
     testRun,
+    request,
   });
 
-  if (testRun) {
+  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);
   }
 
-  const nextStep = await triggerStep.getNextStep();
-  const jobName = `${executionId}-${nextStep.id}`;
+  for (const triggerItem of reversedTriggerItems) {
+    if (testRun) {
+      await processTrigger({
+        flowId,
+        stepId: triggerStep.id,
+        triggerItem,
+        testRun,
+      });
 
-  const jobPayload = {
-    flowId,
-    executionId,
-    stepId: nextStep.id,
-  };
+      continue;
+    }
 
-  const jobOptions = {
-    removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
-    removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
-  };
+    const jobName = `${triggerStep.id}-${triggerItem.meta.internalId}`;
 
-  await actionQueue.add(jobName, jobPayload, jobOptions);
+    const jobOptions = {
+      removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
+      removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
+    };
+
+    const jobPayload = {
+      flowId,
+      stepId: triggerStep.id,
+      triggerItem,
+    };
+
+    await triggerQueue.add(jobName, jobPayload, jobOptions);
+  }
 
   return response.status(204);
 };