feat: introduce singleton webhook URL

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

@@ -3,7 +3,6 @@ import Connection from '../models/connection';
 import Flow from '../models/flow';
 import Step from '../models/step';
 import Execution from '../models/execution';
-import appConfig from '../config/app';
 import {
   IJSONObject,
   IApp,
@@ -17,7 +16,7 @@ import AlreadyProcessedError from '../errors/already-processed';
 
 type GlobalVariableOptions = {
   connection?: Connection;
-  app: IApp;
+  app?: IApp;
   flow?: Flow;
   step?: Step;
   execution?: Execution;
@@ -117,32 +116,36 @@ const globalVariable = async (
     $.request = request;
   }
 
-  $.http = createHttpClient({
-    $,
-    baseURL: app.apiBaseUrl,
-    beforeRequest: app.beforeRequest,
-  });
-
-  if (flow) {
-    const webhookUrl = appConfig.webhookUrl + '/webhooks/' + flow.id;
-
-    $.webhookUrl = webhookUrl;
+  if (app) {
+    $.http = createHttpClient({
+      $,
+      baseURL: app.apiBaseUrl,
+      beforeRequest: app.beforeRequest,
+    });
   }
 
-  if (isTrigger && (await step.getTriggerCommand()).type === 'webhook') {
-    $.flow.setRemoteWebhookId = async (remoteWebhookId) => {
-      await flow.$query().patchAndFetch({
-        remoteWebhookId,
-      });
+  if (step) {
+    $.webhookUrl = await step.getWebhookUrl();
+  }
 
-      $.flow.remoteWebhookId = remoteWebhookId;
-    };
+  if (isTrigger) {
+    const triggerCommand = await step.getTriggerCommand();
 
-    $.flow.remoteWebhookId = flow.remoteWebhookId;
+    if (triggerCommand.type === 'webhook') {
+      $.flow.setRemoteWebhookId = async (remoteWebhookId) => {
+        await flow.$query().patchAndFetch({
+          remoteWebhookId,
+        });
+
+        $.flow.remoteWebhookId = remoteWebhookId;
+      };
+
+      $.flow.remoteWebhookId = flow.remoteWebhookId;
+    }
   }
 
   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

@@ -0,0 +1,84 @@
+import Crypto from 'node:crypto';
+import { Response } from 'express';
+import { IRequest, ITriggerItem } from '@automatisch/types';
+
+import Flow from '../models/flow';
+import { processTrigger } from '../services/trigger';
+import actionQueue from '../queues/action';
+import globalVariable from './global-variable';
+import QuotaExceededError from '../errors/quote-exceeded';
+import {
+  REMOVE_AFTER_30_DAYS_OR_150_JOBS,
+  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();
+
+  const user = await flow.$relatedQuery('user');
+
+  const testRun = !flow.active;
+  const quotaExceeded = !testRun && !(await user.isAllowedToRunFlows());
+
+  if (quotaExceeded) {
+    throw new QuotaExceededError();
+  }
+
+  const triggerStep = await flow.getTriggerStep();
+  const app = await triggerStep.getApp();
+  const isWebhookApp = app.key === 'webhook';
+
+  if ((testRun && !isWebhookApp)) {
+    return response.status(404);
+  }
+
+  // in case trigger type is 'webhook'
+  if (!isWebhookApp) {
+    computedRequestPayload = request.body;
+  }
+
+  const triggerItem: ITriggerItem = {
+    raw: computedRequestPayload,
+    meta: {
+      internalId: Crypto.randomUUID(),
+    },
+  };
+
+  const { executionId } = await processTrigger({
+    flowId,
+    stepId: triggerStep.id,
+    triggerItem,
+    testRun,
+  });
+
+  if (testRun) {
+    return response.status(204);
+  }
+
+  const nextStep = await triggerStep.getNextStep();
+  const jobName = `${executionId}-${nextStep.id}`;
+
+  const jobPayload = {
+    flowId,
+    executionId,
+    stepId: nextStep.id,
+  };
+
+  const jobOptions = {
+    removeOnComplete: REMOVE_AFTER_7_DAYS_OR_50_JOBS,
+    removeOnFail: REMOVE_AFTER_30_DAYS_OR_150_JOBS,
+  };
+
+  await actionQueue.add(jobName, jobPayload, jobOptions);
+
+  return response.status(204);
+};