New integration system: support for webhooks

.changeset/many-needles-jam.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Added support for INTEGRATION_WEBHOOK trigger type

.changeset/serious-maps-exercise.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/typeform": patch
+---
+
+First release of the Typeform integration, with webhook support

apps/integrations/package.json

@@ -14,33 +14,43 @@
     "lint-fix": "eslint --no-error-on-unmatched-pattern --fix src/**",
     "generate-integrations": "NODE_OPTIONS=--max-old-space-size=16384 tsx src/trigger/sdk/generate.ts generate",
     "generate": "prisma generate",
-    "convert": "tsx src/tools/schemas/typescript-to-json-schema.ts convert",
+    "convert-ts": "tsx src/tools/schemas/typescript-to-json-schema.ts convert",
+    "convert-3": "tsx src/tools/schemas/openapi-3-to-3.1.ts convert",
     "db:migrate:deploy": "prisma migrate deploy",
     "db:migrate:dev": "prisma migrate dev"
   },
   "dependencies": {
     "@cfworker/json-schema": "^1.12.5",
+    "@nangohq/node": "^0.8.4",
     "@prisma/client": "^4.3.0",
     "@sentry/node": "^7.38.0",
     "ajv": "^8.12.0",
+    "body-parser": "^1.20.2",
     "express": "^4.18.1",
     "express-async-errors": "^3.1.1",
+    "graphile-worker": "^0.13.0",
     "json-pointer": "^0.6.2",
+    "json-schema": "^0.4.0",
     "json-schema-deref-sync": "^0.14.0",
     "json-schema-to-typescript": "^11.0.3",
     "loglevel": "^1.8.1",
     "morgan": "^1.10.0",
     "node-fetch": "^3.3.0",
     "node-object-hash": "^2.3.10",
+    "qs": "^6.11.0",
+    "stripe": "^11.13.0",
     "zod": "^3.20.2"
   },
   "devDependencies": {
     "@apidevtools/json-schema-ref-parser": "^10.1.0",
+    "@types/body-parser": "^1.19.2",
     "@types/eslint": "^8.4.6",
     "@types/express": "^4.17.13",
     "@types/json-pointer": "^1.0.31",
+    "@types/json-schema": "^7.0.11",
     "@types/morgan": "^1.9.3",
     "@types/node": "^18.13.0",
+    "@types/qs": "^6.9.7",
     "@typescript-eslint/eslint-plugin": "^5.51.0",
     "commander": "^9.4.1",
     "concurrently": "^7.6.0",
@@ -51,6 +61,7 @@
     "eslint-plugin-import": "^2.27.5",
     "eslint-plugin-n": "^15.6.1",
     "eslint-plugin-promise": "^6.1.1",
+    "json-schema-to-zod": "^0.6.3",
     "nock": "^13.3.0",
     "nodemon": "^2.0.19",
     "prisma": "^4.3.0",

apps/integrations/prisma/migrations/20230303164722_webhook_table/migration.sql

@@ -0,0 +1,30 @@
+-- CreateEnum
+CREATE TYPE "WebhookType" AS ENUM ('SERVICE', 'GENERIC');
+
+-- CreateEnum
+CREATE TYPE "WebhookStatus" AS ENUM ('CREATED', 'READY', 'CANCELLED');
+
+-- CreateEnum
+CREATE TYPE "WebhookSubscriptionType" AS ENUM ('AUTOMATIC', 'MANUAL');
+
+-- CreateTable
+CREATE TABLE "Webhook" (
+    "id" TEXT NOT NULL,
+    "type" "WebhookType" NOT NULL,
+    "status" "WebhookStatus" NOT NULL,
+    "consumerId" TEXT NOT NULL,
+    "key" TEXT NOT NULL,
+    "secret" TEXT,
+    "subscriptionType" "WebhookSubscriptionType" NOT NULL,
+    "externalData" JSONB,
+    "service" TEXT,
+    "webhookName" TEXT,
+    "authenticationData" JSONB,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "Webhook_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "Webhook_consumerId_key_key" ON "Webhook"("consumerId", "key");

apps/integrations/prisma/migrations/20230303170626_destiantion_table/migration.sql

@@ -0,0 +1,19 @@
+-- CreateTable
+CREATE TABLE "Destination" (
+    "id" TEXT NOT NULL,
+    "webhookId" TEXT NOT NULL,
+    "destinationUrl" TEXT NOT NULL,
+    "destinationSecret" TEXT NOT NULL,
+    "destinationData" JSONB,
+    "destinationEvent" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "Destination_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "Destination_destinationUrl_key" ON "Destination"("destinationUrl");
+
+-- AddForeignKey
+ALTER TABLE "Destination" ADD CONSTRAINT "Destination_webhookId_fkey" FOREIGN KEY ("webhookId") REFERENCES "Webhook"("id") ON DELETE CASCADE ON UPDATE CASCADE;

apps/integrations/prisma/migrations/20230303174056_added_url_webhook/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - Added the required column `url` to the `Webhook` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "Webhook" ADD COLUMN     "url" TEXT NOT NULL;

apps/integrations/prisma/migrations/20230303174800_removed_webhook_url/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - You are about to drop the column `url` on the `Webhook` table. All the data in the column will be lost.
+
+*/
+-- AlterTable
+ALTER TABLE "Webhook" DROP COLUMN "url";

apps/integrations/prisma/migrations/20230304133437_added_webhook_event_delivery_table/migration.sql

@@ -0,0 +1,14 @@
+-- CreateTable
+CREATE TABLE "WebhookEventDelivery" (
+    "id" TEXT NOT NULL,
+    "eventName" TEXT NOT NULL,
+    "destinationId" TEXT NOT NULL,
+    "payload" JSONB NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "WebhookEventDelivery_pkey" PRIMARY KEY ("id")
+);
+
+-- AddForeignKey
+ALTER TABLE "WebhookEventDelivery" ADD CONSTRAINT "WebhookEventDelivery_destinationId_fkey" FOREIGN KEY ("destinationId") REFERENCES "Destination"("id") ON DELETE CASCADE ON UPDATE CASCADE;

apps/integrations/prisma/migrations/20230304175813_added_displayproperties_event_delivery/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - Added the required column `displayProperties` to the `WebhookEventDelivery` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "WebhookEventDelivery" ADD COLUMN     "displayProperties" JSONB NOT NULL;

apps/integrations/prisma/schema.prisma

@@ -19,4 +19,75 @@ model Cache {
   createdAt DateTime @default(now())
 
   @@unique([namespace, key])
+}
+
+model Webhook {
+  id    String @id @default(cuid())
+  type WebhookType
+  status WebhookStatus
+  consumerId String
+  key String
+  secret String?
+  subscriptionType WebhookSubscriptionType
+  /// automatic subscriptions responses are stored here
+  externalData Json?
+
+  //service
+  service String?
+  webhookName String?
+  authenticationData Json?
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  destinations Destination[]
+
+  @@unique([consumerId, key])
+}
+
+enum WebhookType {
+  SERVICE
+  GENERIC
+}
+
+enum WebhookStatus {
+  CREATED
+  READY
+  CANCELLED
+}
+
+enum WebhookSubscriptionType {
+  AUTOMATIC
+  MANUAL
+}
+
+model Destination {
+  id    String @id @default(cuid())
+
+  webhook Webhook @relation(fields: [webhookId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  webhookId String
+
+  destinationUrl String @unique
+  destinationSecret String
+  destinationData Json?
+  destinationEvent String
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  deliveries WebhookEventDelivery[]
+}
+
+model WebhookEventDelivery {
+  id    String @id @default(cuid())
+  eventName String
+
+  destination Destination @relation(fields: [destinationId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+  destinationId String
+
+  displayProperties Json
+  payload Json
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
 }

apps/integrations/src/api/v2/action/validation.ts

@@ -30,6 +30,18 @@ export function getServiceAction(
     };
   }
 
+  if (!matchingService.actions) {
+    return {
+      success: false,
+      error: {
+        type: "missing_action",
+        message: "Action not found",
+        service,
+        action,
+      },
+    };
+  }
+
   const matchingAction = Object.values(matchingService.actions).find(
     (a) => a.name === action
   );

apps/integrations/src/api/v2/fetch.ts

@@ -1,6 +1,6 @@
 import { AuthCredentials } from "core/authentication/types";
-import { HTTPMethodSchema } from "core/endpoint/types";
 import { serviceFetch } from "core/fetch/serviceFetch";
+import { HTTPMethodSchema } from "core/request/types";
 import { Request, Response } from "express";
 import { catalog } from "integrations/catalog";
 import { z } from "zod";

apps/integrations/src/api/v2/services.ts

@@ -32,7 +32,7 @@ export async function handleServices(_req: Request, res: Response) {
 
 function omitExtraInfo(
   service: Service
-): Omit<Service, "actions" | "retryableStatusCodes"> {
-  const { actions, retryableStatusCodes, ...rest } = service;
+): Omit<Service, "actions" | "retryableStatusCodes" | "webhooks"> {
+  const { actions, retryableStatusCodes, webhooks, ...rest } = service;
   return rest;
 }

apps/integrations/src/api/v2/webhooks/create.ts

@@ -0,0 +1,51 @@
+import { Request, Response } from "express";
+import {
+  SubscribeInputSchema,
+  SubscribeResult,
+} from "core/webhook/subscribe/types";
+import { SubscribeToWebhook } from "core/webhook/subscribe";
+import { Prisma } from "db/db.server";
+
+export async function handleCreateWebhook(req: Request, res: Response) {
+  const parsedBody = SubscribeInputSchema.safeParse(req.body);
+
+  if (!parsedBody.success) {
+    const badBodyResponse: SubscribeResult = {
+      success: false,
+      error: {
+        code: "bad_body",
+        message: parsedBody.error.toString(),
+      },
+    };
+    res.status(400).json(badBodyResponse);
+    return;
+  }
+
+  try {
+    const subscribeToWebhook = new SubscribeToWebhook();
+    const result = await subscribeToWebhook.call(parsedBody.data);
+
+    if (result.success) {
+      res.status(201).json(result);
+      return;
+    }
+
+    res.status(400).json(result);
+    return;
+  } catch (error) {
+    if (error instanceof Prisma.PrismaClientKnownRequestError) {
+      if (error.code === "P2002") {
+        res.status(400).json({
+          success: false,
+          error: {
+            code: "duplicate_webhook",
+            message: `A webhook with this ${error.meta?.target} already exists`,
+          },
+        });
+        return;
+      }
+    }
+
+    res.status(500).json(error);
+  }
+}

apps/integrations/src/api/v2/webhooks/receive.ts

@@ -0,0 +1,47 @@
+import { createDeliveriesAndTasks } from "core/jobs/tasks/webhookJob";
+import { HTTPMethod } from "core/request/types";
+import { ReceiveWebhook } from "core/webhook/receive";
+import {
+  WebhookIncomingRequest,
+  WebhookReceiveRequest,
+} from "core/webhook/types";
+import { Request, Response } from "express";
+
+export async function handleReceivingWebhook(req: Request, res: Response) {
+  const { webhookId } = req.params;
+
+  if (!webhookId) {
+    res.status(400).json({ success: false, error: "Missing webhook ID" });
+    return;
+  }
+
+  const headers = Object.entries(req.headers)
+    .filter(([key, value]) => value !== undefined)
+    .reduce((acc, [key, value]) => {
+      acc[key] =
+        typeof value === "string" ? (value as string) : value?.join(", ") ?? "";
+      return acc;
+    }, {} as Record<string, string>);
+
+  const request: WebhookIncomingRequest = {
+    method: req.method as HTTPMethod,
+    searchParams: new URLSearchParams(req.url),
+    headers,
+    body: req.body,
+    rawBody: req.rawBody,
+  };
+
+  try {
+    const receiver = new ReceiveWebhook();
+    const result = await receiver.call({ request, webhookId });
+
+    const response = res.status(result.response.status);
+    Object.entries(result.response.headers).forEach(([key, value]) => {
+      response.header(key, value);
+    });
+    response.json(result.response.body);
+  } catch (error) {
+    console.error(error);
+    res.status(500).json({ success: false, error: "Internal server error" });
+  }
+}

apps/integrations/src/cache/postgresCache.ts

@@ -1,6 +1,5 @@
-import type { Prisma, Cache } from ".prisma/client";
-import { prisma } from "./db.server";
 import { CacheService } from "core/cache/types";
+import { prisma } from "../db/db.server";
 
 export class PostgresCacheService implements CacheService {
   constructor(private readonly namespace: string) {}

apps/integrations/src/core/action/makeAction.ts

@@ -2,7 +2,7 @@ import { CacheService } from "core/cache/types";
 import { Endpoint } from "core/endpoint/types";
 import { RequestData, RequestResponse } from "core/request/types";
 import { getDisplayProperties } from "./getDisplayProperties";
-import { Action, InputSpec, Metadata, OutputSpec } from "./types";
+import { Action, Metadata } from "./types";
 import { makeInputSpec, makeOutputSpec } from "./utilities";
 
 /** Create an action where you specify the spec and action */