refactor(automations): one shared trigger tail, verify helpers, one options call — every provider

apps/api/package.json

@@ -15,6 +15,7 @@
 		"@better-auth/oauth-provider": "1.6.22",
 		"@linear/sdk": "68.1.1",
 		"@modelcontextprotocol/sdk": "1.30.0",
+		"@notionhq/client": "5.25.2",
 		"@octokit/app": "16.1.2",
 		"@octokit/rest": "22.0.1",
 		"@octokit/webhooks": "14.2.0",
@@ -48,6 +49,7 @@
 		"zod": "4.4.3"
 	},
 	"devDependencies": {
+		"@microsoft/microsoft-graph-types": "2.43.1",
 		"@superset/typescript": "workspace:*",
 		"@types/lodash.chunk": "4.2.9",
 		"@types/node": "24.12.0",

apps/api/src/app/api/automations/dispatch/[id]/route.ts

@@ -1,39 +1,26 @@
 import { dbWs } from "@superset/db/client";
 import { automations } from "@superset/db/schema";
 import { dispatchAutomation } from "@superset/trpc/automation-dispatch";
-import { Receiver } from "@upstash/qstash";
 import { eq } from "drizzle-orm";
-import { env } from "@/env";
 import { getRelayUrl } from "@/lib/relay-url";
+import { verifyQstashRequest } from "@/lib/verifyQstash";
 import { runPayloadSchema } from "../../runPayloadSchema";
 
 export const maxDuration = 60;
 export const dynamic = "force-dynamic";
 
-const receiver = new Receiver({
-	currentSigningKey: env.QSTASH_CURRENT_SIGNING_KEY,
-	nextSigningKey: env.QSTASH_NEXT_SIGNING_KEY,
-});
-
 export async function POST(
 	request: Request,
 	{ params }: { params: Promise<{ id: string }> },
 ): Promise<Response> {
 	const body = await request.text();
-	const signature = request.headers.get("upstash-signature");
-	if (!signature) {
-		return Response.json({ error: "Missing signature" }, { status: 401 });
-	}
-
 	const { id } = await params;
-	const valid = await receiver.verify({
+	const rejected = await verifyQstashRequest(
+		request,
 		body,
-		signature,
-		url: `${env.NEXT_PUBLIC_API_URL}/api/automations/dispatch/${id}`,
-	});
-	if (!valid) {
-		return Response.json({ error: "Invalid signature" }, { status: 401 });
-	}
+		`/api/automations/dispatch/${id}`,
+	);
+	if (rejected) return rejected;
 
 	const parsed = runPayloadSchema.safeParse(JSON.parse(body));
 	if (!parsed.success) {

apps/api/src/app/api/automations/evaluate/route.ts

@@ -5,22 +5,18 @@ import {
 	type TriggerConfig,
 } from "@superset/db/schema";
 import { nextOccurrenceAfter } from "@superset/shared/rrule";
-import { Client, Receiver } from "@upstash/qstash";
+import { Client } from "@upstash/qstash";
 import { and, eq, lte } from "drizzle-orm";
-
 import { env } from "@/env";
+import { redispatchUndispatched } from "@/lib/automations/redispatchUndispatched";
+import { verifyQstashRequest } from "@/lib/verifyQstash";
 
 export const dynamic = "force-dynamic";
 
 const qstash = new Client({
 	token: env.QSTASH_TOKEN,
 	baseUrl: env.QSTASH_URL,
 });
-const receiver = new Receiver({
-	currentSigningKey: env.QSTASH_CURRENT_SIGNING_KEY,
-	nextSigningKey: env.QSTASH_NEXT_SIGNING_KEY,
-});
-
 const BATCH_SIZE = 2000;
 
 function bucketToMinute(d: Date): Date {
@@ -45,19 +41,12 @@ function scheduleFromConfig(
 
 export async function POST(request: Request): Promise<Response> {
 	const body = await request.text();
-	const signature = request.headers.get("upstash-signature");
-	if (!signature) {
-		return Response.json({ error: "Missing signature" }, { status: 401 });
-	}
-
-	const valid = await receiver.verify({
+	const rejected = await verifyQstashRequest(
+		request,
 		body,
-		signature,
-		url: `${env.NEXT_PUBLIC_API_URL}/api/automations/evaluate`,
-	});
-	if (!valid) {
-		return Response.json({ error: "Invalid signature" }, { status: 401 });
-	}
+		"/api/automations/evaluate",
+	);
+	if (rejected) return rejected;
 
 	const now = new Date();
 
@@ -73,7 +62,6 @@ export async function POST(request: Request): Promise<Response> {
 		.where(
 			and(
 				eq(automationTriggers.kind, "schedule"),
-				eq(automationTriggers.enabled, true),
 				eq(automations.enabled, true),
 				lte(automationTriggers.nextRunAt, now),
 			),
@@ -129,7 +117,13 @@ export async function POST(request: Request): Promise<Response> {
 	}
 
 	if (planned.length === 0) {
-		return Response.json({ enqueued: 0, unusable: unusable.length });
+		// The event-handoff sweep still runs on a tick with no due schedules.
+		const redispatched = await redispatchUndispatched();
+		return Response.json({
+			enqueued: 0,
+			unusable: unusable.length,
+			redispatched,
+		});
 	}
 
 	await qstash.batchJSON(
@@ -150,7 +144,10 @@ export async function POST(request: Request): Promise<Response> {
 		planned.map(async ({ triggerId, next }) => {
 			await dbWs
 				.update(automationTriggers)
-				.set(next ? { nextRunAt: next } : { enabled: false })
+				// Always a date for rules the product accepts (finite recurrences
+				// are refused at save); legacy finite data writes null and simply
+				// leaves the select, which reads only set next_run_at.
+				.set({ nextRunAt: next })
 				.where(eq(automationTriggers.id, triggerId));
 		}),
 	);
@@ -175,9 +172,13 @@ export async function POST(request: Request): Promise<Response> {
 		);
 	}
 
+	// The same tick retries event handoffs that never reached QStash.
+	const redispatched = await redispatchUndispatched();
+
 	return Response.json({
 		enqueued: planned.length,
 		advanceFailed: advanceFailures.length,
 		unusable: unusable.length,
+		redispatched,
 	});
 }

apps/api/src/app/api/automations/run-failed/route.ts

@@ -1,20 +1,13 @@
 import * as Sentry from "@sentry/nextjs";
 import { dbWs } from "@superset/db/client";
 import { automationRuns, automations } from "@superset/db/schema";
-import { Receiver } from "@upstash/qstash";
 import { eq, sql } from "drizzle-orm";
 import { z } from "zod";
-
-import { env } from "@/env";
+import { verifyQstashRequest } from "@/lib/verifyQstash";
 import { runPayloadSchema } from "../runPayloadSchema";
 
 export const dynamic = "force-dynamic";
 
-const receiver = new Receiver({
-	currentSigningKey: env.QSTASH_CURRENT_SIGNING_KEY,
-	nextSigningKey: env.QSTASH_NEXT_SIGNING_KEY,
-});
-
 const failurePayloadSchema = z.object({
 	sourceMessageId: z.string(),
 	sourceBody: z.string(),
@@ -25,19 +18,12 @@ const failurePayloadSchema = z.object({
 
 export async function POST(request: Request): Promise<Response> {
 	const body = await request.text();
-	const signature = request.headers.get("upstash-signature");
-	if (!signature) {
-		return Response.json({ error: "Missing signature" }, { status: 401 });
-	}
-
-	const valid = await receiver.verify({
+	const rejected = await verifyQstashRequest(
+		request,
 		body,
-		signature,
-		url: `${env.NEXT_PUBLIC_API_URL}/api/automations/run-failed`,
-	});
-	if (!valid) {
-		return Response.json({ error: "Invalid signature" }, { status: 401 });
-	}
+		"/api/automations/run-failed",
+	);
+	if (rejected) return rejected;
 
 	let rawBody: unknown;
 	try {

apps/api/src/app/api/automations/webhook/[automationId]/normalizeWebhookDelivery.ts

@@ -0,0 +1,37 @@
+import { randomUUID } from "node:crypto";
+
+import type { NormalizedDelivery } from "@/lib/automations/ingestAutomationEvent";
+
+export const EVENT_TYPE = "webhook.received";
+
+/**
+ * Every authenticated delivery is its own event — there is no dedupe — and
+ * fires every enabled webhook trigger on the one automation the URL names.
+ */
+export function normalizeWebhookDelivery(params: {
+	organizationId: string;
+	automationId: string;
+	payload: Record<string, unknown> | unknown[];
+}): NormalizedDelivery {
+	return {
+		event: {
+			organizationId: params.organizationId,
+			integrationConnectionId: null,
+			provider: "webhook",
+			eventType: EVENT_TYPE,
+			externalEventId: randomUUID(),
+			title: "Webhook",
+			payload: params.payload,
+		},
+		dispatch: {
+			automationId: params.automationId,
+			event: {
+				provider: "webhook",
+				eventType: EVENT_TYPE,
+				actorId: null,
+				actorLogin: null,
+				body: null,
+			},
+		},
+	};
+}

apps/api/src/app/api/automations/webhook/[automationId]/route.ts

@@ -1,11 +1,6 @@
-import { createHash, randomUUID } from "node:crypto";
+import { createHash } from "node:crypto";
 import { db } from "@superset/db/client";
-import {
-	automationEvents,
-	automations,
-	automationTriggers,
-} from "@superset/db/schema";
-import type { WebhookMatchableEvent } from "@superset/shared/automation-matching";
+import { automations, automationTriggers } from "@superset/db/schema";
 import {
 	bearerToken,
 	WEBHOOK_TOKEN_PREFIX,
@@ -17,8 +12,9 @@ import { and, eq } from "drizzle-orm";
 import { z } from "zod";
 
 import { env } from "@/env";
-import { dispatchMatchingTriggers } from "@/lib/automations/dispatchMatchingTriggers";
-import { recordAutomationEvent } from "@/lib/automations/recordAutomationEvent";
+import { ingestAutomationEvent } from "@/lib/automations/ingestAutomationEvent";
+import { cappedBody } from "@/lib/webhooks/body";
+import { normalizeWebhookDelivery } from "./normalizeWebhookDelivery";
 
 export const dynamic = "force-dynamic";
 
@@ -31,9 +27,6 @@ const rateLimit = new Ratelimit({
 	prefix: "ratelimit:automations:webhook",
 });
 
-const EVENT_TYPE = "webhook.received";
-const MAX_BODY_BYTES = 1024 * 1024;
-
 function parseBody(body: string): Record<string, unknown> | unknown[] {
 	if (body.trim() === "") return {};
 	const parsed: unknown = JSON.parse(body);
@@ -75,11 +68,6 @@ export async function POST(
 		return Response.json({ error: "Rate limit exceeded" }, { status: 429 });
 	}
 
-	const contentLength = Number(request.headers.get("content-length"));
-	if (contentLength > MAX_BODY_BYTES) {
-		return Response.json({ error: "Body too large" }, { status: 413 });
-	}
-
 	const triggers = await db
 		.select({
 			secretHash: automationTriggers.secretHash,
@@ -92,7 +80,6 @@ export async function POST(
 			and(
 				eq(automationTriggers.automationId, automationId),
 				eq(automationTriggers.kind, "webhook"),
-				eq(automationTriggers.enabled, true),
 			),
 		);
 
@@ -110,10 +97,8 @@ export async function POST(
 		);
 	}
 
-	const body = await request.text();
-	if (Buffer.byteLength(body) > MAX_BODY_BYTES) {
-		return Response.json({ error: "Body too large" }, { status: 413 });
-	}
+	const body = await cappedBody(request);
+	if (body instanceof Response) return body;
 	let payload: Record<string, unknown> | unknown[];
 	try {
 		payload = parseBody(body);
@@ -124,53 +109,13 @@ export async function POST(
 		);
 	}
 
-	const inserted = await recordAutomationEvent(db, {
-		organizationId,
-		integrationConnectionId: null,
-		provider: "webhook",
-		eventType: EVENT_TYPE,
-		externalEventId: randomUUID(),
-		title: "Webhook",
-		payload,
-	});
-
-	if (!inserted) {
-		return Response.json({ error: "Failed to record event" }, { status: 500 });
-	}
-
-	const event: WebhookMatchableEvent = {
-		provider: "webhook",
-		eventType: EVENT_TYPE,
-		actorId: null,
-		actorLogin: null,
-		body: null,
-	};
-	let result: { matched: number; considered: number };
-	try {
-		result = await dispatchMatchingTriggers({
-			organizationId,
-			eventId: inserted.id,
-			event,
-			automationId,
-		});
-	} catch (error) {
-		console.error(
-			`[automations/webhook] dispatch failed for event ${inserted.id}:`,
-			error,
-		);
-		await db
-			.delete(automationEvents)
-			.where(eq(automationEvents.id, inserted.id));
-		return Response.json({ error: "Dispatch failed" }, { status: 500 });
-	}
-
-	console.log(
-		`[automations/webhook] ${result.matched}/${result.considered} triggers matched:`,
-		inserted.id,
+	const outcome = await ingestAutomationEvent(
+		db,
+		normalizeWebhookDelivery({ organizationId, automationId, payload }),
 	);
 	return Response.json({
 		ok: true,
-		eventId: inserted.id,
-		runs: result.matched,
+		eventId: "eventId" in outcome ? outcome.eventId : null,
+		runs: outcome.status === "dispatched" ? outcome.matched : 0,
 	});
 }