From ccb620807c1d36dc4bbb392e21277ac93f0c081d Mon Sep 17 00:00:00 2001 From: nicktrn <55853254+nicktrn@users.noreply.github.com> Date: Thu, 30 Nov 2023 23:35:09 +0000 Subject: [PATCH] Webhook delivery batching and some fixes --- apps/webapp/app/platform/zodWorker.server.ts | 3 +- .../route.tsx | 1 + .../route.tsx | 1 + .../route.tsx | 1 + .../webapp/app/services/endpointApi.server.ts | 20 +++- .../services/events/deliverEvent.server.ts | 37 ++++-- ...er.server.ts => dispatchBatcher.server.ts} | 29 +++-- .../events/invokeDispatcher.server.ts | 5 +- .../app/services/jobs/invokeJob.server.ts | 1 + .../app/services/jobs/registerJob.server.ts | 97 +++++++++------ .../app/services/jobs/testJob.server.ts | 1 + .../app/services/runs/createRun.server.ts | 4 + .../runs/performRunExecutionV3.server.ts | 2 + apps/webapp/app/services/runs/reRun.server.ts | 11 +- .../schedules/deliverScheduledEvent.server.ts | 2 +- .../sources/deliverWebhookRequest.server.ts | 91 ++++++++------ .../sources/handleWebhookRequest.server.ts | 62 ++++++++-- .../sources/webhookDeliveryBatcher.server.ts | 98 +++++++++++++++ .../triggers/registerWebhook.server.ts | 60 +++++++++- apps/webapp/app/services/worker.server.ts | 45 ++++++- packages/core/src/schemas/api.ts | 29 ++++- packages/core/src/schemas/triggers.ts | 6 +- .../migration.sql | 35 ++++++ packages/database/prisma/schema.prisma | 42 ++++++- packages/testing/src/index.ts | 14 ++- packages/trigger-sdk/src/triggerClient.ts | 112 ++++++++++++------ .../trigger-sdk/src/triggers/eventTrigger.ts | 13 +- packages/trigger-sdk/src/triggers/webhook.ts | 6 + 28 files changed, 647 insertions(+), 181 deletions(-) rename apps/webapp/app/services/events/{dispatchChunker.server.ts => dispatchBatcher.server.ts} (76%) create mode 100644 apps/webapp/app/services/sources/webhookDeliveryBatcher.server.ts create mode 100644 packages/database/prisma/migrations/20231130233325_add_webhook_delivery_batcher/migration.sql diff --git a/apps/webapp/app/platform/zodWorker.server.ts b/apps/webapp/app/platform/zodWorker.server.ts index 3b6e66fbb..e57c59e62 100644 --- a/apps/webapp/app/platform/zodWorker.server.ts +++ b/apps/webapp/app/platform/zodWorker.server.ts @@ -573,7 +573,8 @@ export class ZodWorker { await this.enqueue(helpers.job.task_identifier, payload, { runAt: helpers.job.run_at, - queueName: helpers.job.queue_name ?? undefined, + // TODO: fix queueName, maybe use reschedule helper + // queueName: helpers.job.job_queue_id ?? undefined, priority: helpers.job.priority, jobKey: helpers.job.key ?? undefined, flags: Object.keys(helpers.job.flags ?? []), diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.trigger/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.trigger/route.tsx index 4dc25127d..9715800be 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.trigger/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.trigger/route.tsx @@ -28,6 +28,7 @@ export default function Page() { ); diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.$runParam.trigger/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.$runParam.trigger/route.tsx index e186e2ba0..decdd162b 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.$runParam.trigger/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.$runParam.trigger/route.tsx @@ -28,6 +28,7 @@ export default function Page() { ); diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.delivery.$runParam.trigger/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.delivery.$runParam.trigger/route.tsx index 478dfbe77..99a1d239f 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.delivery.$runParam.trigger/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.delivery.$runParam.trigger/route.tsx @@ -28,6 +28,7 @@ export default function Page() { ); diff --git a/apps/webapp/app/services/endpointApi.server.ts b/apps/webapp/app/services/endpointApi.server.ts index 1f64fbbdc..791975742 100644 --- a/apps/webapp/app/services/endpointApi.server.ts +++ b/apps/webapp/app/services/endpointApi.server.ts @@ -254,12 +254,22 @@ export class EndpointApi { return HttpSourceResponseSchema.parse(anyBody); } - async deliverWebhookRequest(options: { + // TODO: ensure backward compat + async deliverWebhookRequests(options: { key: string; secret: string; params: any; - request: HttpSourceRequest; + requests: HttpSourceRequest[]; + batched: boolean; }) { + const serializedRequests = options.requests.map((request) => { + const { rawBody, ...requestWithoutRawBody } = request; + return { + ...requestWithoutRawBody, + body: rawBody?.toString(), + }; + }); + const response = await safeFetch(this.url, { method: "POST", headers: { @@ -267,13 +277,11 @@ export class EndpointApi { "x-trigger-api-key": this.apiKey, "x-trigger-action": "DELIVER_WEBHOOK_REQUEST", "x-ts-key": options.key, + "x-ts-batched": JSON.stringify(options.batched), "x-ts-secret": options.secret, "x-ts-params": JSON.stringify(options.params ?? {}), - "x-ts-http-url": options.request.url, - "x-ts-http-method": options.request.method, - "x-ts-http-headers": JSON.stringify(options.request.headers), }, - body: options.request.rawBody, + body: JSON.stringify(serializedRequests), }); if (!response) { diff --git a/apps/webapp/app/services/events/deliverEvent.server.ts b/apps/webapp/app/services/events/deliverEvent.server.ts index b9b565e57..6dea08e8f 100644 --- a/apps/webapp/app/services/events/deliverEvent.server.ts +++ b/apps/webapp/app/services/events/deliverEvent.server.ts @@ -5,6 +5,7 @@ import { EventFilterSchema, RequestWithRawBodySchema, eventFilterMatches } from import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server"; import { logger } from "~/services/logger.server"; import { workerQueue } from "../worker.server"; +import { ZodWorkerBatchEnqueueOptions } from "~/platform/zodWorker.server"; export class DeliverEventService { #prismaClient: PrismaClientOrTransaction; @@ -68,20 +69,16 @@ export class DeliverEventService { eventRecord: eventRecord.id, }); - const DEFAULT_MAX_PAYLOADS = 10; - const DEFAULT_MAX_INTERVAL_IN_SECONDS = 10; - await Promise.all( matchingEventDispatchers.map((eventDispatcher) => { if (eventDispatcher.batcher) { + const { maxPayloads, runAt } = this.#getBatchEnqueueOptions(eventDispatcher.batcher); + return workerQueue.batchEnqueue("events.invokeDispatchChunker", [eventRecord.id], { tx, jobKey: eventDispatcher.id, - maxPayloads: eventDispatcher.batcher.maxPayloads ?? DEFAULT_MAX_PAYLOADS, - runAt: new Date( - Date.now() + - (eventDispatcher.batcher.maxInterval ?? DEFAULT_MAX_INTERVAL_IN_SECONDS) * 1000 - ), + maxPayloads, + runAt, }); } else { return workerQueue.enqueue( @@ -140,6 +137,30 @@ export class DeliverEventService { context: contextFilter.data, }); } + + #getBatchEnqueueOptions(batcherConfig?: { + maxPayloads: number | null; + maxInterval: number | null; + }): Pick { + const DEFAULT_MAX_PAYLOADS = 500; + const DEFAULT_MAX_INTERVAL_IN_SECONDS = 10 * 60; + + const MAX_PAYLOADS = DEFAULT_MAX_PAYLOADS; + const MAX_INTERVAL_IN_SECONDS = DEFAULT_MAX_INTERVAL_IN_SECONDS; + + const maxPayloads = Math.min(batcherConfig?.maxPayloads ?? DEFAULT_MAX_PAYLOADS, MAX_PAYLOADS); + + const runAt = new Date( + Date.now() + + Math.min( + batcherConfig?.maxInterval ?? DEFAULT_MAX_INTERVAL_IN_SECONDS, + MAX_INTERVAL_IN_SECONDS + ) * + 1000 + ); + + return { maxPayloads, runAt }; + } } export class EventMatcher { diff --git a/apps/webapp/app/services/events/dispatchChunker.server.ts b/apps/webapp/app/services/events/dispatchBatcher.server.ts similarity index 76% rename from apps/webapp/app/services/events/dispatchChunker.server.ts rename to apps/webapp/app/services/events/dispatchBatcher.server.ts index 8d59f5be0..9905ac13d 100644 --- a/apps/webapp/app/services/events/dispatchChunker.server.ts +++ b/apps/webapp/app/services/events/dispatchBatcher.server.ts @@ -4,9 +4,8 @@ import { logger } from "~/services/logger.server"; import { workerQueue } from "../worker.server"; const DEFAULT_MAX_PAYLOAD_SIZE = 2 * 1024 * 1024; // 2MB -const DEFAULT_MAX_INTERVAL_IN_SECONDS = 20; -export class DispatchChunkerService { +export class DispatchBatcherService { #prismaClient: PrismaClientOrTransaction; constructor( @@ -52,12 +51,14 @@ export class DispatchChunkerService { const chunks: Record = { 0: [] }; - const maxInterval = eventDispatcher.batcher.maxInterval ?? DEFAULT_MAX_INTERVAL_IN_SECONDS; - for (const event of eventRecords) { if (chunkSize + event.payloadSize > this.maxPayloadSize) { // enqueue full chunk - await this.#enqueueChunk(eventDispatcher.id, chunks[chunkIndex], maxInterval); + await this.#enqueueChunk( + eventDispatcher.id, + chunks[chunkIndex], + eventDispatcher.batcher.maxInterval + ); // start new chunk chunkIndex++; @@ -70,22 +71,28 @@ export class DispatchChunkerService { } if (chunks[chunkIndex].length) { - await this.#enqueueChunk(eventDispatcher.id, chunks[chunkIndex], maxInterval); + await this.#enqueueChunk( + eventDispatcher.id, + chunks[chunkIndex], + eventDispatcher.batcher.maxInterval + ); } } - async #enqueueChunk(dispatcherId: string, eventRecordIds: string[], deliverAfter?: number) { + async #enqueueChunk(dispatcherId: string, eventRecordIds: string[], maxInterval: number | null) { logger.debug("Invoking batch event dispatcher", { dispatcherId, totalEvents: eventRecordIds.length, }); + const MAX_INTERVAL_IN_SECONDS = 10 * 60; + + const deliverAfter = maxInterval ? Math.max(maxInterval, MAX_INTERVAL_IN_SECONDS) : undefined; + await workerQueue.enqueue( "events.invokeBatchDispatcher", - { id: dispatcherId, eventRecordIds }, - // { - // runAt: deliverAfter ? deliverAfterToDate(deliverAfter) : undefined, - // } + { id: dispatcherId, eventRecordIds } + // { runAt: deliverAfter ? deliverAfterToDate(deliverAfter) : undefined } ); } } diff --git a/apps/webapp/app/services/events/invokeDispatcher.server.ts b/apps/webapp/app/services/events/invokeDispatcher.server.ts index 7077e841c..a0785d89e 100644 --- a/apps/webapp/app/services/events/invokeDispatcher.server.ts +++ b/apps/webapp/app/services/events/invokeDispatcher.server.ts @@ -18,6 +18,7 @@ export class InvokeDispatcherService { id, }, include: { + batcher: true, environment: { include: { project: true, @@ -53,7 +54,7 @@ export class InvokeDispatcherService { return; } - logger.debug("Invoking batch event dispatcher", { + logger.debug("Invoking event dispatcher", { eventDispatcher, eventRecordIds, }); @@ -83,6 +84,7 @@ export class InvokeDispatcherService { const createRunService = new CreateRunService(this.#prismaClient); await createRunService.call({ + batched: !!eventDispatcher.batcher, eventIds: eventRecords.map((e) => e.id), job: jobVersion.job, version: jobVersion, @@ -133,6 +135,7 @@ export class InvokeDispatcherService { const createRunService = new CreateRunService(this.#prismaClient); await createRunService.call({ + batched: !!eventDispatcher.batcher, eventIds: eventRecords.map((e) => e.id), job: job, version: latestJobVersion, diff --git a/apps/webapp/app/services/jobs/invokeJob.server.ts b/apps/webapp/app/services/jobs/invokeJob.server.ts index f72757d5b..e89b90a41 100644 --- a/apps/webapp/app/services/jobs/invokeJob.server.ts +++ b/apps/webapp/app/services/jobs/invokeJob.server.ts @@ -131,6 +131,7 @@ export class InvokeJobService { eventIds: [eventLog.id], job: job, version, + batched: false }, { callbackUrl: options.callbackUrl, diff --git a/apps/webapp/app/services/jobs/registerJob.server.ts b/apps/webapp/app/services/jobs/registerJob.server.ts index 367eb01ac..63265a474 100644 --- a/apps/webapp/app/services/jobs/registerJob.server.ts +++ b/apps/webapp/app/services/jobs/registerJob.server.ts @@ -1,13 +1,22 @@ import { + BatcherOptions, IntegrationConfig, JobMetadata, SCHEDULED_EVENT, TriggerMetadata, assertExhaustive, } from "@trigger.dev/core"; -import type { Endpoint, Integration, Job, JobIntegration, JobVersion } from "@trigger.dev/database"; +import type { + Endpoint, + EventDispatcher, + Integration, + Job, + JobIntegration, + JobVersion, + Prisma, +} from "@trigger.dev/database"; import { DEFAULT_MAX_CONCURRENT_RUNS } from "~/consts"; -import type { PrismaClient } from "~/db.server"; +import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server"; import { prisma } from "~/db.server"; import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server"; import type { RuntimeEnvironment } from "~/models/runtimeEnvironment.server"; @@ -16,6 +25,12 @@ import { logger } from "../logger.server"; import { RegisterScheduleSourceService } from "../schedules/registerScheduleSource.server"; import { executionRateLimiter } from "../runExecutionRateLimiter.server"; +type ExtendedEventDispatcher = Prisma.EventDispatcherGetPayload<{ + include: { + batcher: true; + }; +}>; + export class RegisterJobService { #prismaClient: PrismaClient; @@ -338,41 +353,7 @@ export class RegisterJobService { }, }); - if (trigger.batch) { - let maxPayloads: number | null = null; - let maxInterval: number | null = null; - - if (typeof trigger.batch !== "boolean") { - maxPayloads = trigger.batch.maxPayloads ?? null; - maxInterval = trigger.batch.maxInterval ?? null; - } - - await this.#prismaClient.eventDispatchBatcher.upsert({ - where: { - eventDispatcherId: eventDispatcher.id, - environmentId: eventDispatcher.environmentId, - }, - create: { - eventDispatcherId: eventDispatcher.id, - environmentId: eventDispatcher.environmentId, - maxPayloads, - maxInterval, - }, - update: { - maxPayloads, - maxInterval, - }, - }); - } else { - if (eventDispatcher.batcher) { - await this.#prismaClient.eventDispatchBatcher.delete({ - where: { - eventDispatcherId: eventDispatcher.id, - environmentId: eventDispatcher.environmentId, - }, - }); - } - } + await this.#registerDispatchBatcher(this.#prismaClient, eventDispatcher, trigger.batch); if (trigger.properties || trigger.link || trigger.help) { await this.#prismaClient.jobVersion.update({ @@ -434,6 +415,48 @@ export class RegisterJobService { } } + async #registerDispatchBatcher( + tx: PrismaClientOrTransaction, + eventDispatcher: ExtendedEventDispatcher, + batchOptions?: BatcherOptions + ) { + if (batchOptions) { + let maxPayloads: number | null = null; + let maxInterval: number | null = null; + + if (typeof batchOptions !== "boolean") { + maxPayloads = batchOptions.maxPayloads ?? null; + maxInterval = batchOptions.maxInterval ?? null; + } + + await tx.eventDispatchBatcher.upsert({ + where: { + eventDispatcherId: eventDispatcher.id, + environmentId: eventDispatcher.environmentId, + }, + create: { + eventDispatcherId: eventDispatcher.id, + environmentId: eventDispatcher.environmentId, + maxPayloads, + maxInterval, + }, + update: { + maxPayloads, + maxInterval, + }, + }); + } else { + if (eventDispatcher.batcher) { + await tx.eventDispatchBatcher.delete({ + where: { + eventDispatcherId: eventDispatcher.id, + environmentId: eventDispatcher.environmentId, + }, + }); + } + } + } + async #upsertIntegrationForJobIntegration( environment: AuthenticatedEnvironment, jobIntegration: IntegrationConfig diff --git a/apps/webapp/app/services/jobs/testJob.server.ts b/apps/webapp/app/services/jobs/testJob.server.ts index aa6791949..9b9b1280a 100644 --- a/apps/webapp/app/services/jobs/testJob.server.ts +++ b/apps/webapp/app/services/jobs/testJob.server.ts @@ -105,6 +105,7 @@ export class TestJobService { eventIds: [eventLog.id], job: version.job, version, + batched: false }); }, { timeout: 10000 } diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index 22da8f89e..de573f539 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/apps/webapp/app/services/runs/createRun.server.ts @@ -17,11 +17,13 @@ export class CreateRunService { eventIds, job, version, + batched, }: { environment: AuthenticatedEnvironment; eventIds: string[]; job: Job; version: JobVersion; + batched: boolean; }, options: { callbackUrl?: string } = {} ) { @@ -54,10 +56,12 @@ export class CreateRunService { jobId: job.id, versionId: version.id, eventId: firstEvent.id, + eventIds: eventRecords.map((event) => event.id), environmentId: environment.id, organizationId: environment.organizationId, projectId: environment.projectId, endpointId: endpoint.id, + batched, payload: JSON.stringify( eventRecords.length > 1 ? eventRecords.map((event) => event.payload) ?? [{}] diff --git a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts index 4cfe1e0f6..e68e58dd6 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts @@ -434,6 +434,7 @@ export class PerformRunExecutionV3Service { return { event, + batched: run.batched, payload: run.payload, job: { id: run.version.job.slug, @@ -486,6 +487,7 @@ export class PerformRunExecutionV3Service { return { event, + batched: run.batched, payload: run.payload, job: { id: run.version.job.slug, diff --git a/apps/webapp/app/services/runs/reRun.server.ts b/apps/webapp/app/services/runs/reRun.server.ts index 8c36fe54b..c36bbfe04 100644 --- a/apps/webapp/app/services/runs/reRun.server.ts +++ b/apps/webapp/app/services/runs/reRun.server.ts @@ -1,4 +1,4 @@ -import { PrismaClient, prisma } from "~/db.server"; +import { Prisma, PrismaClient, prisma } from "~/db.server"; import { CreateRunService } from "../runs/createRun.server"; export class ReRunService { @@ -51,10 +51,14 @@ export class ReRunService { }, } : undefined, - eventId: `${existingRun.event.eventId}:retry:${new Date().getTime()}`, + eventId: `${existingRun.id}-batch:retry:${new Date().getTime()}`, name: existingRun.event.name, timestamp: new Date(), - payload: existingRun.event.payload ?? {}, + // Get payload directly from Run if batched + payload: + existingRun.batched && existingRun.payload + ? (JSON.parse(existingRun.payload) as Prisma.InputJsonValue) + : existingRun.event.payload ?? {}, context: existingRun.event.context ?? {}, source: existingRun.event.source, isTest: existingRun.event.isTest, @@ -72,6 +76,7 @@ export class ReRunService { eventIds: [eventLog.id], job: existingRun.job, version: existingRun.version, + batched: existingRun.batched, }); }); } catch (error) { diff --git a/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts b/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts index 3dff3a9d9..7a110506f 100644 --- a/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts +++ b/apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts @@ -65,7 +65,7 @@ export class DeliverScheduledEventService { const invokeDispatcherService = new InvokeDispatcherService(tx); - await invokeDispatcherService.call(scheduleSource.dispatcher.id, eventRecord.id); + await invokeDispatcherService.call(scheduleSource.dispatcher.id, [eventRecord.id]); logger.debug("updating lastEventTimestamp", { id, diff --git a/apps/webapp/app/services/sources/deliverWebhookRequest.server.ts b/apps/webapp/app/services/sources/deliverWebhookRequest.server.ts index f821ecce4..5d21ec454 100644 --- a/apps/webapp/app/services/sources/deliverWebhookRequest.server.ts +++ b/apps/webapp/app/services/sources/deliverWebhookRequest.server.ts @@ -11,10 +11,10 @@ export class DeliverWebhookRequestService { this.#prismaClient = prismaClient; } - public async call(id: string) { - const requestDelivery = await this.#prismaClient.webhookRequestDelivery.findUniqueOrThrow({ + public async call(webhookEnvironmentId: string, requestDeliveryIds: string[], batched = false) { + const webhookEnvironment = await this.#prismaClient.webhookEnvironment.findUniqueOrThrow({ where: { - id, + id: webhookEnvironmentId, }, include: { webhook: { @@ -31,25 +31,40 @@ export class DeliverWebhookRequestService { }, }, }, - webhookEnvironment: { + environment: { include: { - environment: { - include: { - organization: true, - project: true, - }, - }, + organization: true, + project: true, }, }, endpoint: true, }, }); - if (!requestDelivery.webhookEnvironment.active) { + if (!webhookEnvironment.active) { return; } - const { secretReference } = requestDelivery.webhook.httpEndpoint; + const requestDeliveries = await this.#prismaClient.webhookRequestDelivery.findMany({ + where: { + id: { in: requestDeliveryIds }, + }, + include: { + endpoint: true, + }, + }); + + if (!requestDeliveries.length) { + throw new Error(`No request deliveries found, expected ${requestDeliveryIds.length} total.`); + } + + if (!batched && requestDeliveries.length > 1) { + throw new Error( + `Batching is disabled. Will not handle multiple deliveries. Requested ${requestDeliveryIds.length} total.` + ); + } + + const { secretReference } = webhookEnvironment.webhook.httpEndpoint; const secretStore = getSecretStore(secretReference.provider); @@ -61,36 +76,44 @@ export class DeliverWebhookRequestService { ); if (!secret) { - throw new Error(`Secret not found for ${requestDelivery.webhook.key}`); + throw new Error(`Secret not found for ${webhookEnvironment.webhook.key}`); } const clientApi = new EndpointApi( - requestDelivery.webhookEnvironment.environment.apiKey, - requestDelivery.endpoint.url + webhookEnvironment.environment.apiKey, + requestDeliveries[0].endpoint.url ); - const { response, verified, error } = await clientApi.deliverWebhookRequest({ - key: requestDelivery.webhook.key, + const requests = requestDeliveries.map((delivery) => ({ + url: delivery.url, + method: delivery.method, + headers: delivery.headers as Record, + rawBody: delivery.body, + })); + + const { response, deliveryResults } = await clientApi.deliverWebhookRequests({ + key: webhookEnvironment.webhook.key, secret: secret.secret, - params: requestDelivery.webhook.params, - request: { - url: requestDelivery.url, - method: requestDelivery.method, - headers: requestDelivery.headers as Record, - rawBody: requestDelivery.body, - }, + params: webhookEnvironment.webhook.params, + requests, + batched, }); - await this.#prismaClient.webhookRequestDelivery.update({ - where: { - id, - }, - data: { - deliveredAt: new Date(), - verified, - error, - }, - }); + const deliveredAt = new Date(); + + await Promise.allSettled( + deliveryResults.map((result, i) => { + return this.#prismaClient.webhookRequestDelivery.update({ + where: { + id: requestDeliveries[i].id, + }, + data: { + ...result, + deliveredAt, + }, + }); + }) + ); return response; } diff --git a/apps/webapp/app/services/sources/handleWebhookRequest.server.ts b/apps/webapp/app/services/sources/handleWebhookRequest.server.ts index d8265864e..23cbc4418 100644 --- a/apps/webapp/app/services/sources/handleWebhookRequest.server.ts +++ b/apps/webapp/app/services/sources/handleWebhookRequest.server.ts @@ -3,8 +3,9 @@ import { prisma } from "~/db.server"; import { workerQueue } from "../worker.server"; import { RuntimeEnvironmentType } from "@trigger.dev/database"; import { createHttpSourceRequest } from "~/utils/createHttpSourceRequest"; -import { WebhookContextMetadata } from "@trigger.dev/core"; +import { BatcherOptions, WebhookContextMetadata } from "@trigger.dev/core"; import { createHash } from "crypto"; +import { ZodWorkerBatchEnqueueOptions } from "~/platform/zodWorker.server"; export class HandleWebhookRequestService { #prismaClient: PrismaClient; @@ -19,6 +20,7 @@ export class HandleWebhookRequestService { id, }, include: { + deliveryBatcher: true, endpoint: true, environment: true, }, @@ -60,23 +62,65 @@ export class HandleWebhookRequestService { }, }); - await workerQueue.enqueue( - "deliverWebhookRequest", - { - id: delivery.id, - }, - { + if (webhookEnvironment.deliveryBatcher) { + const { maxPayloads, runAt } = this.#getBatchEnqueueOptions( + webhookEnvironment.deliveryBatcher + ); + + await workerQueue.batchEnqueue("batchWebhookDeliveryRequests", [delivery.id], { tx, maxAttempts: webhookEnvironment.environment.type === RuntimeEnvironmentType.DEVELOPMENT ? 1 : undefined, - } - ); + jobKey: webhookEnvironment.id, + maxPayloads, + runAt, + }); + } else { + await workerQueue.enqueue( + "deliverWebhookRequest", + { + webhookEnvironmentId: webhookEnvironment.id, + requestDeliveryId: delivery.id, + }, + { + tx, + maxAttempts: + webhookEnvironment.environment.type === RuntimeEnvironmentType.DEVELOPMENT + ? 1 + : undefined, + } + ); + } }); return { status: 200 }; } + + #getBatchEnqueueOptions(batcherConfig?: { + maxPayloads: number | null; + maxInterval: number | null; + }): Pick { + const DEFAULT_MAX_PAYLOADS = 500; + const DEFAULT_MAX_INTERVAL_IN_SECONDS = 10 * 60; + + const MAX_PAYLOADS = DEFAULT_MAX_PAYLOADS; + const MAX_INTERVAL_IN_SECONDS = DEFAULT_MAX_INTERVAL_IN_SECONDS; + + const maxPayloads = Math.min(batcherConfig?.maxPayloads ?? DEFAULT_MAX_PAYLOADS, MAX_PAYLOADS); + + const runAt = new Date( + Date.now() + + Math.min( + batcherConfig?.maxInterval ?? DEFAULT_MAX_INTERVAL_IN_SECONDS, + MAX_INTERVAL_IN_SECONDS + ) * + 1000 + ); + + return { maxPayloads, runAt }; + } } function webhookIdToLockId(webhookId: string): number { diff --git a/apps/webapp/app/services/sources/webhookDeliveryBatcher.server.ts b/apps/webapp/app/services/sources/webhookDeliveryBatcher.server.ts new file mode 100644 index 000000000..592ca7687 --- /dev/null +++ b/apps/webapp/app/services/sources/webhookDeliveryBatcher.server.ts @@ -0,0 +1,98 @@ +import type { PrismaClientOrTransaction } from "~/db.server"; +import { Prisma, prisma } from "~/db.server"; +import { logger } from "~/services/logger.server"; +import { workerQueue } from "../worker.server"; + +const DEFAULT_MAX_PAYLOAD_SIZE = 2 * 1024 * 1024; // 2MB + +export class WebhookDeliveryBatcherService { + #prismaClient: PrismaClientOrTransaction; + + constructor( + private maxPayloadSize = DEFAULT_MAX_PAYLOAD_SIZE, + prismaClient: PrismaClientOrTransaction = prisma + ) { + this.#prismaClient = prismaClient; + } + + public async call(id: string, eventRecordIds: string[]) { + const webhookEnvironment = await this.#prismaClient.webhookEnvironment.findUniqueOrThrow({ + where: { + id, + }, + include: { + deliveryBatcher: true, + }, + }); + + if (!webhookEnvironment.active) { + logger.debug("Webhook environment is disabled", { + webhookEnvironment, + }); + + return; + } + + const batcher = webhookEnvironment.deliveryBatcher; + + if (!batcher) { + logger.debug("Webhook environment has no batcher", { + webhookEnvironment, + }); + + return; + } + + const requestDeliveries = await prisma.$queryRaw<{ id: string; bodySize: number }[]>` + SELECT id, LENGTH(payload::text) AS bodySize FROM "WebhookRequestDelivery" + WHERE id IN (${Prisma.join(eventRecordIds)}); + `; + + let chunkSize = 0; + let chunkIndex = 0; + + const chunks: Record = { 0: [] }; + + for (const delivery of requestDeliveries) { + if (chunkSize + delivery.bodySize > this.maxPayloadSize) { + // enqueue full chunk + await this.#enqueueChunk(webhookEnvironment.id, chunks[chunkIndex], batcher.maxInterval); + + // start new chunk + chunkIndex++; + chunkSize = 0; + chunks[chunkIndex] = []; + } + + chunkSize += delivery.bodySize; + chunks[chunkIndex].push(delivery.id); + } + + if (chunks[chunkIndex].length) { + await this.#enqueueChunk(webhookEnvironment.id, chunks[chunkIndex], batcher.maxInterval); + } + } + + async #enqueueChunk( + webhookEnvironmentId: string, + requestDeliveryIds: string[], + maxInterval: number | null + ) { + logger.debug("Invoking batch webhook delivery", { + webhookEnvironmentId, + totalDeliveries: requestDeliveryIds.length, + }); + + const MAX_INTERVAL_IN_SECONDS = 10 * 60; + + const deliverAfter = maxInterval ? Math.max(maxInterval, MAX_INTERVAL_IN_SECONDS) : undefined; + + await workerQueue.enqueue( + "deliverMultipleWebhookRequests", + { webhookEnvironmentId, requestDeliveryIds }, + // { runAt: deliverAfter ? deliverAfterToDate(deliverAfter) : undefined } + ); + } +} + +const deliverAfterToDate = (seconds: number) => new Date(Date.now() + seconds * 1000); diff --git a/apps/webapp/app/services/triggers/registerWebhook.server.ts b/apps/webapp/app/services/triggers/registerWebhook.server.ts index f5e004f86..1164142ac 100644 --- a/apps/webapp/app/services/triggers/registerWebhook.server.ts +++ b/apps/webapp/app/services/triggers/registerWebhook.server.ts @@ -1,4 +1,4 @@ -import { REGISTER_WEBHOOK, WebhookMetadata } from "@trigger.dev/core"; +import { BatcherOptions, REGISTER_WEBHOOK, WebhookMetadata } from "@trigger.dev/core"; import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server"; import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server"; import { IngestSendEvent } from "../events/ingestSendEvent.server"; @@ -19,6 +19,12 @@ type ExtendedWebhook = Prisma.WebhookGetPayload<{ }; }>; +type ExtendedWebhookEnvironment = Prisma.WebhookEnvironmentGetPayload<{ + include: { + deliveryBatcher: true; + }; +}>; + export class RegisterWebhookService { #prismaClient: PrismaClientOrTransaction; @@ -126,8 +132,13 @@ export class RegisterWebhookService { update: { desiredConfig: webhookMetadata.config, }, + include: { + deliveryBatcher: true, + }, }); + await this.#registerDeliveryBatcher(tx, webhookEnvironment, webhookMetadata.batch); + return { webhook, webhookEnvironment }; }); } @@ -166,4 +177,51 @@ export class RegisterWebhookService { }, }); } + + async #registerDeliveryBatcher( + tx: PrismaClientOrTransaction, + webhookEnvironment: ExtendedWebhookEnvironment, + batchOptions?: BatcherOptions + ) { + if (batchOptions) { + let maxPayloads: number | null = null; + let maxInterval: number | null = null; + + if (typeof batchOptions !== "boolean") { + maxPayloads = batchOptions.maxPayloads ?? null; + maxInterval = batchOptions.maxInterval ?? null; + } + + await tx.webhookDeliveryBatcher.upsert({ + where: { + webhookId_webhookEnvironmentId: { + webhookId: webhookEnvironment.webhookId, + webhookEnvironmentId: webhookEnvironment.id, + }, + }, + create: { + webhookId: webhookEnvironment.webhookId, + webhookEnvironmentId: webhookEnvironment.id, + environmentId: webhookEnvironment.environmentId, + maxPayloads, + maxInterval, + }, + update: { + maxPayloads, + maxInterval, + }, + }); + } else { + if (webhookEnvironment.deliveryBatcher) { + await tx.webhookDeliveryBatcher.delete({ + where: { + webhookId_webhookEnvironmentId: { + webhookId: webhookEnvironment.webhookId, + webhookEnvironmentId: webhookEnvironment.id, + }, + }, + }); + } + } + } } diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index f2306ad3f..6466d3904 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -30,7 +30,8 @@ import { ResumeRunService } from "./runs/resumeRun.server"; import { executionRateLimiter } from "./runExecutionRateLimiter.server"; import { DeliverWebhookRequestService } from "./sources/deliverWebhookRequest.server"; import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server"; -import { DispatchChunkerService } from "./events/dispatchChunker.server"; +import { DispatchBatcherService } from "./events/dispatchBatcher.server"; +import { WebhookDeliveryBatcherService } from "./sources/webhookDeliveryBatcher.server"; const workerCatalog = { indexEndpoint: z.object({ @@ -48,7 +49,15 @@ const workerCatalog = { id: z.string(), }), deliverHttpSourceRequest: z.object({ id: z.string() }), - deliverWebhookRequest: z.object({ id: z.string() }), + batchWebhookDeliveryRequests: z.array(z.string()), + deliverWebhookRequest: z.object({ + webhookEnvironmentId: z.string(), + requestDeliveryId: z.string(), + }), + deliverMultipleWebhookRequests: z.object({ + webhookEnvironmentId: z.string(), + requestDeliveryIds: z.string().array(), + }), refreshOAuthToken: z.object({ organizationId: z.string(), connectionId: z.string(), @@ -250,7 +259,7 @@ function getWorkerQueue() { throw new Error("Job key is required for batch jobs."); } - const service = new DispatchChunkerService(); + const service = new DispatchBatcherService(); await service.call(job.key, payload); }, @@ -337,14 +346,38 @@ function getWorkerQueue() { await service.call(payload.id); }, }, - deliverWebhookRequest: { + batchWebhookDeliveryRequests: { + priority: 0, // smaller number = higher priority + maxAttempts: 6, + queueName: (payload, jobKey) => `webhooks-batch:${jobKey}`, + handler: async (payload, job) => { + if (!job.key) { + throw new Error("Job key is required for batch jobs."); + } + + const service = new WebhookDeliveryBatcherService(); + + await service.call(job.key, payload); + }, + }, + deliverMultipleWebhookRequests: { priority: 1, // smaller number = higher priority maxAttempts: 14, - queueName: (payload) => `webhooks:${payload.id}`, + queueName: (payload) => `webhooks:${payload}`, handler: async (payload, job) => { const service = new DeliverWebhookRequestService(); - await service.call(payload.id); + await service.call(payload.webhookEnvironmentId, payload.requestDeliveryIds, true); + }, + }, + deliverWebhookRequest: { + priority: 1, // smaller number = higher priority + maxAttempts: 14, + queueName: (payload) => `webhooks:${payload.webhookEnvironmentId}`, + handler: async (payload, job) => { + const service = new DeliverWebhookRequestService(); + + await service.call(payload.webhookEnvironmentId, [payload.requestDeliveryId]); }, }, startRun: { diff --git a/packages/core/src/schemas/api.ts b/packages/core/src/schemas/api.ts index 637db54bf..28428140c 100644 --- a/packages/core/src/schemas/api.ts +++ b/packages/core/src/schemas/api.ts @@ -14,7 +14,7 @@ import { ScheduleMetadataSchema, } from "./schedules"; import { CachedTaskSchema, ServerTaskSchema, TaskSchema } from "./tasks"; -import { EventSpecificationSchema, TriggerMetadataSchema } from "./triggers"; +import { BatcherOptionsSchema, EventSpecificationSchema, TriggerMetadataSchema } from "./triggers"; import { RunStatusSchema } from "./runs"; import { JobRunStatusRecordSchema } from "./statuses"; import { RequestFilterSchema } from "./requestFilter"; @@ -217,16 +217,25 @@ export const HttpEndpointRequestHeadersSchema = z.object({ export const WebhookSourceRequestHeadersSchema = z.object({ "x-ts-key": z.string(), + "x-ts-batched": z.string().transform((s) => JSON.parse(s) as boolean), "x-ts-dynamic-id": z.string().optional(), "x-ts-secret": z.string(), "x-ts-params": z.string().transform((s) => JSON.parse(s)), - "x-ts-http-url": z.string(), - "x-ts-http-method": z.string(), - "x-ts-http-headers": z.string().transform((s) => z.record(z.string()).parse(JSON.parse(s))), }); export type WebhookSourceRequestHeaders = z.output; +export const WebhookSourceRequestBodySchema = z + .object({ + url: z.string(), + method: z.string(), + headers: z.record(z.string()), + body: z.string().optional(), + }) + .array(); + +export type WebhookSourceRequestBody = z.infer; + export const PongSuccessResponseSchema = z.object({ ok: z.literal(true), triggerVersion: z.string().optional(), @@ -339,6 +348,7 @@ export const WebhookMetadataSchema = z.object({ key: z.string(), params: z.any(), config: z.record(z.array(z.string())), + batch: BatcherOptionsSchema.optional(), integration: IntegrationConfigSchema, httpEndpoint: z.object({ id: z.string(), @@ -583,6 +593,7 @@ export type AutoYieldConfig = z.infer; export const RunJobBodySchema = z.object({ event: ApiEventLogSchema, payload: z.string().nullable(), + batched: z.boolean(), job: z.object({ id: z.string(), version: z.string(), @@ -983,12 +994,18 @@ export const HttpSourceResponseSchema = z.object({ metadata: HttpSourceResponseMetadataSchema.optional(), }); -export const WebhookDeliveryResponseSchema = z.object({ - response: NormalizedResponseSchema, +export const WebhookDeliveryResultSchema = z.object({ verified: z.boolean(), error: z.string().optional(), }); +export type WebhookDeliveryResult = z.infer; + +export const WebhookDeliveryResponseSchema = z.object({ + response: NormalizedResponseSchema, + deliveryResults: WebhookDeliveryResultSchema.array(), +}); + export type WebhookDeliveryResponse = z.infer; export const RegisterTriggerBodySchemaV1 = z.object({ diff --git a/packages/core/src/schemas/triggers.ts b/packages/core/src/schemas/triggers.ts index d1022bec7..4f8c8174e 100644 --- a/packages/core/src/schemas/triggers.ts +++ b/packages/core/src/schemas/triggers.ts @@ -37,7 +37,7 @@ export const TriggerHelpSchema = z.object({ .optional(), }); -export const EventDispatchBatcherOptionsSchema = z.union([ +export const BatcherOptionsSchema = z.union([ z.boolean(), z.object({ maxPayloads: z.number().optional(), @@ -45,7 +45,7 @@ export const EventDispatchBatcherOptionsSchema = z.union([ }), ]); -export type EventDispatchBatcherOptions = z.infer; +export type BatcherOptions = z.infer; export const StaticTriggerMetadataSchema = z.object({ type: z.literal("static"), @@ -54,7 +54,7 @@ export const StaticTriggerMetadataSchema = z.object({ rule: EventRuleSchema, link: z.string().optional(), help: TriggerHelpSchema.optional(), - batch: EventDispatchBatcherOptionsSchema.optional(), + batch: BatcherOptionsSchema.optional(), }); export const InvokeTriggerMetadataSchema = z.object({ diff --git a/packages/database/prisma/migrations/20231130233325_add_webhook_delivery_batcher/migration.sql b/packages/database/prisma/migrations/20231130233325_add_webhook_delivery_batcher/migration.sql new file mode 100644 index 000000000..13e8559c8 --- /dev/null +++ b/packages/database/prisma/migrations/20231130233325_add_webhook_delivery_batcher/migration.sql @@ -0,0 +1,35 @@ +-- AlterTable +ALTER TABLE "JobRun" ADD COLUMN "batched" BOOLEAN NOT NULL DEFAULT false, +ADD COLUMN "eventIds" TEXT[]; + +-- AlterTable +ALTER TABLE "Webhook" ADD COLUMN "batched" BOOLEAN NOT NULL DEFAULT false; + +-- CreateTable +CREATE TABLE "WebhookDeliveryBatcher" ( + "id" TEXT NOT NULL, + "maxPayloads" INTEGER, + "maxInterval" INTEGER, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + "webhookId" TEXT NOT NULL, + "webhookEnvironmentId" TEXT NOT NULL, + "environmentId" TEXT NOT NULL, + + CONSTRAINT "WebhookDeliveryBatcher_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "WebhookDeliveryBatcher_webhookEnvironmentId_key" ON "WebhookDeliveryBatcher"("webhookEnvironmentId"); + +-- CreateIndex +CREATE UNIQUE INDEX "WebhookDeliveryBatcher_webhookId_webhookEnvironmentId_key" ON "WebhookDeliveryBatcher"("webhookId", "webhookEnvironmentId"); + +-- AddForeignKey +ALTER TABLE "WebhookDeliveryBatcher" ADD CONSTRAINT "WebhookDeliveryBatcher_webhookId_fkey" FOREIGN KEY ("webhookId") REFERENCES "Webhook"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "WebhookDeliveryBatcher" ADD CONSTRAINT "WebhookDeliveryBatcher_webhookEnvironmentId_fkey" FOREIGN KEY ("webhookEnvironmentId") REFERENCES "WebhookEnvironment"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "WebhookDeliveryBatcher" ADD CONSTRAINT "WebhookDeliveryBatcher_environmentId_fkey" FOREIGN KEY ("environmentId") REFERENCES "RuntimeEnvironment"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 93a3fd5ed..c0b593963 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -333,6 +333,7 @@ model RuntimeEnvironment { keyValueItems KeyValueItem[] webhookEnvironments WebhookEnvironment[] webhookRequestDeliveries WebhookRequestDelivery[] + webhookDeliveryBatchers WebhookDeliveryBatcher[] @@unique([projectId, slug, orgMemberId]) @@unique([projectId, shortcode]) @@ -673,7 +674,7 @@ model EventDispatcher { enabled Boolean @default(true) - batcher EventDispatchBatcher? + batcher EventDispatchBatcher? environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) environmentId String @@ -696,8 +697,8 @@ model EventDispatchBatcher { createdAt DateTime @default(now()) updatedAt DateTime @updatedAt - eventDispatcher EventDispatcher @relation(fields: [eventDispatcherId], references: [id], onDelete: Cascade, onUpdate: Cascade) - eventDispatcherId String @unique + eventDispatcher EventDispatcher @relation(fields: [eventDispatcherId], references: [id], onDelete: Cascade, onUpdate: Cascade) + eventDispatcherId String @unique environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) environmentId String @@ -761,7 +762,9 @@ model JobRun { id String @id @default(cuid()) number Int? internal Boolean @default(false) - payload String? + + payload String? + batched Boolean @default(false) job Job @relation(fields: [jobId], references: [id], onDelete: Cascade, onUpdate: Cascade) jobId String @@ -772,6 +775,9 @@ model JobRun { event EventRecord @relation(fields: [eventId], references: [id], onDelete: Cascade, onUpdate: Cascade) eventId String + // Will be empty unless batched + eventIds String[] + organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade) organizationId String @@ -1152,13 +1158,15 @@ model TriggerSourceOption { model Webhook { id String @id @default(cuid()) - active Boolean @default(false) + active Boolean @default(false) + batched Boolean @default(false) key String params Json? webhookEnvironments WebhookEnvironment[] requestDeliveries WebhookRequestDelivery[] + deliveryBatchers WebhookDeliveryBatcher[] httpEndpoint TriggerHttpEndpoint @relation(fields: [httpEndpointId], references: [id], onDelete: Cascade, onUpdate: Cascade) httpEndpointId String @unique @@ -1183,7 +1191,8 @@ model WebhookEnvironment { config Json? desiredConfig Json? - requestDeliveries WebhookRequestDelivery[] + requestDeliveries WebhookRequestDelivery[] + deliveryBatcher WebhookDeliveryBatcher? endpoint Endpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade, onUpdate: Cascade) endpointId String @@ -1231,6 +1240,27 @@ model WebhookRequestDelivery { deliveredAt DateTime? } +model WebhookDeliveryBatcher { + id String @id @default(cuid()) + + maxPayloads Int? + maxInterval Int? + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + webhook Webhook @relation(fields: [webhookId], references: [id], onDelete: Cascade, onUpdate: Cascade) + webhookId String + + webhookEnvironment WebhookEnvironment @relation(fields: [webhookEnvironmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) + webhookEnvironmentId String @unique + + environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade) + environmentId String + + @@unique([webhookId, webhookEnvironmentId]) +} + model WebhookDeliveryCounter { webhookId String @id lastNumber Int @default(0) diff --git a/packages/testing/src/index.ts b/packages/testing/src/index.ts index 80a8ddff0..03db0809e 100644 --- a/packages/testing/src/index.ts +++ b/packages/testing/src/index.ts @@ -1,4 +1,5 @@ import { + BatcherOptions, RunJobBody, SendEvent, SendEventBodySchema, @@ -85,9 +86,15 @@ const buildRequest = (action: TriggerAction, apiKey: string, opts: Record ({ +const buildRequestBody = ( + event: RunJobBody["event"], + job: RunJobBody["job"], + batched: boolean +): RunJobBody => ({ event, job, + payload: JSON.stringify(batched ? [event.payload] : event.payload), + batched, run: { id: String(Math.random()), isTest: false, @@ -192,8 +199,11 @@ export const createJobTester = return run(payload, io, ctx); }; + const triggerMetadata = job.trigger.toJSON(); + const batchOptions = triggerMetadata.type === "static" ? triggerMetadata.batch : undefined; + const request = buildRequest("EXECUTE_JOB", client.apiKey() ?? "", { - body: buildRequestBody(eventLog, job), + body: buildRequestBody(eventLog, job, !!batchOptions), jobId: job.id, }); const requestResult = await client.handleRequest(request); diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts index 853fa1ef5..29d632e04 100644 --- a/packages/trigger-sdk/src/triggerClient.ts +++ b/packages/trigger-sdk/src/triggerClient.ts @@ -1,5 +1,6 @@ import { API_VERSIONS, + BatcherOptions, ConnectionAuth, DELIVER_WEBHOOK_REQUEST, DeserializedJson, @@ -43,7 +44,9 @@ import { StatusUpdate, SuccessfulRunNotification, WebhookDeliveryResponse, + WebhookDeliveryResult, WebhookMetadata, + WebhookSourceRequestBodySchema, WebhookSourceRequestHeadersSchema, } from "@trigger.dev/core"; import { yellow } from "colorette"; @@ -587,24 +590,30 @@ export class TriggerClient { }; } - const sourceRequestNeedsBody = headers.data["x-ts-http-method"] !== "GET"; + const parsedBody = await request.json(); - const sourceRequestInit: RequestInit = { - method: headers.data["x-ts-http-method"], - headers: headers.data["x-ts-http-headers"], - body: sourceRequestNeedsBody ? request.body : undefined, - }; + const serializedRequests = WebhookSourceRequestBodySchema.parse(parsedBody); - if (sourceRequestNeedsBody) { - try { - // @ts-ignore - sourceRequestInit.duplex = "half"; - } catch (error) { - // ignore + const requests = serializedRequests.map((req) => { + const needsBody = req.method !== "GET"; + + const sourceRequestInit: RequestInit = { + method: req.method, + headers: req.headers, + body: needsBody ? request.body : undefined, + }; + + if (needsBody) { + try { + // @ts-ignore + sourceRequestInit.duplex = "half"; + } catch (error) { + // ignore + } } - } - const webhookRequest = new Request(headers.data["x-ts-http-url"], sourceRequestInit); + return new Request(req.url, sourceRequestInit); + }); const key = headers.data["x-ts-key"]; const secret = headers.data["x-ts-secret"]; @@ -616,14 +625,25 @@ export class TriggerClient { params, }; - const { response, verified, error } = await this.#handleWebhookRequest(webhookRequest, ctx); + const deliveryResults: WebhookDeliveryResult[] = []; + + for (const request of requests) { + const { verified, error } = await this.#handleWebhookRequest(request, ctx); + deliveryResults.push({ verified, error }); + } + + const okResponse = { + status: 200, + body: { + ok: true, + }, + }; return { status: 200, body: { - response, - verified, - error, + response: okResponse, + deliveryResults, }, headers: this.#standardResponseHeaders(timeOrigin), }; @@ -899,6 +919,7 @@ export class TriggerClient { if (!registeredWebhook) { registeredWebhook = { key: options.key, + batch: options.source.batch, params: options.params, config: options.config, integration: { @@ -1224,15 +1245,42 @@ export class TriggerClient { try { // For compatibility with old Job Runs where payload is only available on the related Event Record const payload = body.payload ? JSON.parse(body.payload) : body.event.payload; - const parsedPayload = job.trigger.event.parsePayload(payload ?? {}); - if (!context.run.isTest) { - const verified = await job.trigger.verifyPayload(parsedPayload); - if (!verified.success) { - return { - status: "ERROR", - error: { message: `Payload verification failed. ${verified.reason}` }, - }; + let parsedPayload: any = {}; + + if (body.batched) { + if (!Array.isArray(payload)) { + throw new Error("The payload for batched Runs needs to be an array."); + } + + parsedPayload = payload.map((element) => job.trigger.event.parsePayload(element)); + + if (!context.run.isTest) { + for (const parsedElement of parsedPayload) { + const verified = await job.trigger.verifyPayload(parsedElement); + + if (!verified.success) { + return { + status: "ERROR", + error: { + message: `Payload verification failed with batching enabled. ${verified.reason}`, + }, + }; + } + } + } + } else { + parsedPayload = job.trigger.event.parsePayload(payload ?? {}); + + if (!context.run.isTest) { + const verified = await job.trigger.verifyPayload(parsedPayload); + + if (!verified.success) { + return { + status: "ERROR", + error: { message: `Payload verification failed. ${verified.reason}` }, + }; + } } } @@ -1593,18 +1641,11 @@ export class TriggerClient { async #handleWebhookRequest( request: Request, ctx: WebhookDeliveryContext - ): Promise { + ): Promise { this.#internalLogger.debug("Handling webhook request", { ctx, }); - const okResponse = { - status: 200, - body: { - ok: true, - }, - }; - const handlers = this.#registeredWebhookSourceHandlers[ctx.key]; if (!handlers) { @@ -1613,7 +1654,6 @@ export class TriggerClient { }); return { - response: okResponse, verified: false, }; } @@ -1624,7 +1664,6 @@ export class TriggerClient { if (!verifyResult.success) { return { - response: okResponse, verified: false, error: verifyResult.reason, }; @@ -1633,7 +1672,6 @@ export class TriggerClient { await generateEvents(request, this, ctx); return { - response: okResponse, verified: true, }; } diff --git a/packages/trigger-sdk/src/triggers/eventTrigger.ts b/packages/trigger-sdk/src/triggers/eventTrigger.ts index fec8dc0c2..389f73e6c 100644 --- a/packages/trigger-sdk/src/triggers/eventTrigger.ts +++ b/packages/trigger-sdk/src/triggers/eventTrigger.ts @@ -1,9 +1,4 @@ -import { - EventDispatchBatcherOptions, - EventFilter, - TriggerMetadata, - deepMergeFilters, -} from "@trigger.dev/core"; +import { BatcherOptions, EventFilter, TriggerMetadata, deepMergeFilters } from "@trigger.dev/core"; import { Job } from "../job"; import { TriggerClient } from "../triggerClient"; import { @@ -23,7 +18,7 @@ type EventTriggerOptions> = source?: string; filter?: EventFilter; verify?: EventTypeFromSpecification extends Request ? VerifyCallback : never; - batch?: EventDispatchBatcherOptions; + batch?: BatcherOptions; }; export class EventTrigger> @@ -55,7 +50,7 @@ export class EventTrigger> attachToJob(triggerClient: TriggerClient, job: Job, any>): void {} batch( - options?: Exclude + options?: Exclude ): EventTrigger { const { batch, ...rest } = this.#options; @@ -112,7 +107,7 @@ type TriggerOptions = { */ filter?: EventFilter; /** Used to set batching options. */ - batch?: EventDispatchBatcherOptions; + batch?: BatcherOptions; examples?: EventSpecificationExample[]; }; diff --git a/packages/trigger-sdk/src/triggers/webhook.ts b/packages/trigger-sdk/src/triggers/webhook.ts index 85a360127..84ba28d52 100644 --- a/packages/trigger-sdk/src/triggers/webhook.ts +++ b/packages/trigger-sdk/src/triggers/webhook.ts @@ -1,4 +1,5 @@ import { + BatcherOptions, DisplayProperty, EventFilter, HandleTriggerSource, @@ -141,6 +142,7 @@ type WebhookOptions< crud: WebhookCRUD; filter?: FilterFunction; register?: RegisterFunction; + batch?: BatcherOptions; verify?: (options: { request: Request; client: TriggerClient; @@ -226,6 +228,10 @@ export class WebhookSource< return `${this.options.id}-${this.#shortHash(parts.join(""))}`; } + get batch() { + return this.options.batch; + } + get integration() { return this.options.integration; }