Webhooks GA

apps/web/app/api/slack/callback/route.ts

@@ -5,6 +5,7 @@ import { getSlackEnv } from "@/lib/integrations/slack/env";
 import { SlackCredential } from "@/lib/integrations/slack/type";
 import { prisma } from "@/lib/prisma";
 import { redis } from "@/lib/upstash";
+import { addWebhook } from "@/lib/webhook/api";
 import z from "@/lib/zod";
 import {
   APP_DOMAIN_WITH_NGROK,
@@ -24,7 +25,7 @@ const oAuthCallbackSchema = z.object({
 export const GET = async (req: Request) => {
   const env = getSlackEnv();
 
-  let workspace: Pick<Project, "slug"> | null = null;
+  let workspace: Pick<Project, "id" | "slug"> | null = null;
 
   try {
     const session = await getSession();
@@ -53,6 +54,7 @@ export const GET = async (req: Request) => {
         id: workspaceId,
       },
       select: {
+        id: true,
         slug: true,
       },
     });
@@ -73,6 +75,13 @@ export const GET = async (req: Request) => {
 
     const data = await response.json();
 
+    const webhook = await addWebhook({
+      name: "Slack",
+      url: data.incoming_webhook.url,
+      triggers: ["link.created"],
+      workspace,
+    });
+
     const credentials: SlackCredential = {
       appId: data.app_id,
       botUserId: data.bot_user_id,
@@ -81,6 +90,11 @@ export const GET = async (req: Request) => {
       tokenType: data.token_type,
       authUser: data.authed_user,
       team: data.team,
+      incomingWebhook: {
+        channel: data.incoming_webhook.channel,
+        channelId: data.incoming_webhook.channel_id,
+        webhookId: webhook.id,
+      },
     };
 
     await installIntegration({

apps/web/app/api/stripe/webhook/customer-subscription-deleted.ts

@@ -146,5 +146,24 @@ export async function customerSubscriptionDeleted(event: Stripe.Event) {
     sendCancellationFeedback({
       owners: workspaceUsers,
     }),
+
+    // Disable the webhooks
+    prisma.webhook.updateMany({
+      where: {
+        projectId: workspace.id,
+      },
+      data: {
+        disabledAt: new Date(),
+      },
+    }),
+
+    prisma.project.update({
+      where: {
+        id: workspace.id,
+      },
+      data: {
+        webhookEnabled: false,
+      },
+    }),
   ]);
 }

apps/web/app/api/stripe/webhook/customer-subscription-updated.ts

@@ -59,6 +59,7 @@ export async function customerSubscriptionUpdated(event: Stripe.Event) {
   }
 
   const newPlan = plan.name.toLowerCase();
+  const shouldDisableWebhooks = newPlan === "free" || newPlan === "pro";
 
   // If a workspace upgrades/downgrades their subscription, update their usage limit in the database.
   if (workspace.plan !== newPlan) {
@@ -78,6 +79,7 @@ export async function customerSubscriptionUpdated(event: Stripe.Event) {
           paymentFailedAt: null,
         },
       }),
+
       prisma.restrictedToken.updateMany({
         where: {
           projectId: workspace.id,
@@ -86,6 +88,25 @@ export async function customerSubscriptionUpdated(event: Stripe.Event) {
           rateLimit: plan.limits.api,
         },
       }),
+
+      // Disable the webhooks if the new plan does not support webhooks
+      ...(shouldDisableWebhooks
+        ? [
+            prisma.webhook.updateMany({
+              where: { projectId: workspace.id },
+              data: { disabledAt: new Date() },
+            }),
+
+            prisma.project.update({
+              where: {
+                id: workspace.id,
+              },
+              data: {
+                webhookEnabled: false,
+              },
+            }),
+          ]
+        : []),
     ]);
   } else if (workspace.paymentFailedAt) {
     await prisma.project.update({

apps/web/app/api/track/click/route.ts

@@ -57,6 +57,7 @@ export const POST = async (req: Request) => {
           linkId: link.id,
           url: link.url,
           skipRatelimit: true,
+          workspaceId: workspace.id,
         }),
       );
     }

apps/web/app/api/webhooks/[webhookId]/route.ts

@@ -3,6 +3,7 @@ import { linkCache } from "@/lib/api/links/cache";
 import { parseRequestBody } from "@/lib/api/utils";
 import { withWorkspace } from "@/lib/auth";
 import { prisma } from "@/lib/prisma";
+import { updateWebhookStatusForWorkspace } from "@/lib/webhook/api";
 import { webhookCache } from "@/lib/webhook/cache";
 import { transformWebhook } from "@/lib/webhook/transform";
 import { isLinkLevelWebhook } from "@/lib/webhook/utils";
@@ -26,6 +27,7 @@ export const GET = withWorkspace(
         url: true,
         secret: true,
         triggers: true,
+        disabledAt: true,
         links: true,
       },
     });
@@ -54,21 +56,23 @@ export const PATCH = withWorkspace(
       await parseRequestBody(req),
     );
 
-    const webhookUrlExists = await prisma.webhook.findFirst({
-      where: {
-        projectId: workspace.id,
-        url,
-        id: {
-          not: webhookId,
+    if (url) {
+      const webhookUrlExists = await prisma.webhook.findFirst({
+        where: {
+          projectId: workspace.id,
+          url,
+          id: {
+            not: webhookId,
+          },
         },
-      },
-    });
-
-    if (webhookUrlExists) {
-      throw new DubApiError({
-        code: "conflict",
-        message: "A Webhook with this URL already exists.",
       });
+
+      if (webhookUrlExists) {
+        throw new DubApiError({
+          code: "conflict",
+          message: "A Webhook with this URL already exists.",
+        });
+      }
     }
 
     if (linkIds && linkIds.length > 0) {
@@ -124,6 +128,7 @@ export const PATCH = withWorkspace(
         url: true,
         secret: true,
         triggers: true,
+        disabledAt: true,
         links: {
           select: {
             linkId: true,
@@ -258,24 +263,6 @@ export const DELETE = withWorkspace(
       },
     });
 
-    const webhooksCount = await prisma.webhook.count({
-      where: {
-        projectId: workspace.id,
-      },
-    });
-
-    // Disable webhooks for the workspace if there are no more webhooks
-    if (webhooksCount === 0) {
-      await prisma.project.update({
-        where: {
-          id: workspace.id,
-        },
-        data: {
-          webhookEnabled: false,
-        },
-      });
-    }
-
     waitUntil(
       (async () => {
         const links = await prisma.link.findMany({
@@ -299,6 +286,7 @@ export const DELETE = withWorkspace(
         });
 
         await Promise.all([
+          updateWebhookStatusForWorkspace({ workspace }),
           linkCache.mset(formatedLinks),
           webhookCache.delete(webhookId),
         ]);

apps/web/app/api/webhooks/callback/route.ts

@@ -1,8 +1,20 @@
 import { verifyQstashSignature } from "@/lib/cron/verify-qstash";
+import { prisma } from "@/lib/prisma";
 import { recordWebhookEvent } from "@/lib/tinybird/record-webhook-event";
-import { webhookPayloadSchema } from "@/lib/webhook/schemas";
+import { WEBHOOK_TRIGGERS } from "@/lib/webhook/constants";
+import {
+  handleWebhookFailure,
+  resetWebhookFailureCount,
+} from "@/lib/webhook/failure";
 import { webhookCallbackSchema } from "@/lib/zod/schemas/webhooks";
 import { getSearchParams } from "@dub/utils";
+import { z } from "zod";
+
+const searchParamsSchema = z.object({
+  webhookId: z.string(),
+  eventId: z.string(),
+  event: z.enum(WEBHOOK_TRIGGERS),
+});
 
 // POST /api/webhooks/callback – listen to webhooks status from QStash
 export const POST = async (req: Request) => {
@@ -13,25 +25,44 @@ export const POST = async (req: Request) => {
   const { url, status, body, sourceBody, sourceMessageId } =
     webhookCallbackSchema.parse(rawBody);
 
+  const { webhookId, eventId, event } = searchParamsSchema.parse(
+    getSearchParams(req.url),
+  );
+
+  const webhook = await prisma.webhook.findUnique({
+    where: { id: webhookId },
+  });
+
+  if (!webhook) {
+    console.error("Webhook not found", { webhookId });
+    return new Response("Webhook not found");
+  }
+
   const request = Buffer.from(sourceBody, "base64").toString("utf-8");
   const response = Buffer.from(body, "base64").toString("utf-8");
+  const isFailed = status >= 400;
 
-  const { id: eventId, event } = webhookPayloadSchema.parse(
-    JSON.parse(request),
-  );
+  await Promise.all([
+    // Record the webhook event
+    recordWebhookEvent({
+      url,
+      event,
+      event_id: eventId,
+      http_status: status,
+      webhook_id: webhookId,
+      request_body: request,
+      response_body: response,
+      message_id: sourceMessageId,
+    }),
 
-  const { webhookId } = getSearchParams(req.url);
-
-  await recordWebhookEvent({
-    url,
-    event,
-    event_id: eventId,
-    http_status: status,
-    webhook_id: webhookId,
-    request_body: request,
-    response_body: response,
-    message_id: sourceMessageId,
-  });
+    // Handle the webhook delivery failure if it's the last retry
+    ...(isFailed ? [handleWebhookFailure(webhookId)] : []),
+
+    // Only reset if there were previous failures
+    ...(webhook.consecutiveFailures > 0 && !isFailed
+      ? [resetWebhookFailureCount(webhookId)]
+      : []),
+  ]);
 
   return new Response("OK");
 };