github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Webhook delivery batching and some fixes

Browse Source
This commit is contained in:
nicktrn
2023-11-30 23:35:09 +00:00
parent 531b9e7e99
commit ccb620807c
28 changed files with 647 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/platform/zodWorker.server.ts
+2 -1
View File
@@ -573,7 +573,8 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
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 ?? []),
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.external.$triggerParam_.runs.$runParam.trigger/route.tsx
+1
View File
@@ -28,6 +28,7 @@ export default function Page() {
<TriggerDetail
trigger={trigger}
event={{ icon: "register-source", title: "Register external source" }}
payload={trigger.payload}
properties={[]}
/>
);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.$runParam.trigger/route.tsx
+1
View File
@@ -28,6 +28,7 @@ export default function Page() {
<TriggerDetail
trigger={trigger}
event={{ icon: "webhook", title: "Register Webhook" }}
payload={trigger.payload}
properties={[]}
/>
);
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.triggers_.webhooks.$triggerParam_.runs.delivery.$runParam.trigger/route.tsx
+1
View File
@@ -28,6 +28,7 @@ export default function Page() {
<TriggerDetail
trigger={trigger}
event={{ icon: "mail-fast", title: "Deliver Webhook" }}
payload={trigger.payload}
properties={[]}
/>
);
apps/webapp/app/services/endpointApi.server.ts
+14 -6
View File
@@ -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) {
apps/webapp/app/services/events/deliverEvent.server.ts
+29 -8
View File
@@ -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<ZodWorkerBatchEnqueueOptions, "maxPayloads" | "runAt"> {
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 {
apps/webapp/app/services/events/dispatchChunker.server.ts → apps/webapp/app/services/events/dispatchBatcher.server.ts
+18 -11
View File
@@ -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<number, string[]> = { 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 }
);
}
}
apps/webapp/app/services/events/invokeDispatcher.server.ts
+4 -1
View File
@@ -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,
apps/webapp/app/services/jobs/invokeJob.server.ts
+1
View File
@@ -131,6 +131,7 @@ export class InvokeJobService {
eventIds: [eventLog.id],
job: job,
version,
batched: false
},
{
callbackUrl: options.callbackUrl,
apps/webapp/app/services/jobs/registerJob.server.ts
+60 -37
View File
@@ -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
apps/webapp/app/services/jobs/testJob.server.ts
+1
View File
@@ -105,6 +105,7 @@ export class TestJobService {
eventIds: [eventLog.id],
job: version.job,
version,
batched: false
});
},
{ timeout: 10000 }
apps/webapp/app/services/runs/createRun.server.ts
+4
View File
@@ -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) ?? [{}]
apps/webapp/app/services/runs/performRunExecutionV3.server.ts
+2
View File
@@ -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,
apps/webapp/app/services/runs/reRun.server.ts
+8 -3
View File
@@ -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) {
apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts
+1 -1
View File
@@ -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,
apps/webapp/app/services/sources/deliverWebhookRequest.server.ts
+57 -34
View File
@@ -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<string, string>,
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<string, string>,
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;
}
apps/webapp/app/services/sources/handleWebhookRequest.server.ts
+53 -9
View File
@@ -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<ZodWorkerBatchEnqueueOptions, "maxPayloads" | "runAt"> {
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 {
apps/webapp/app/services/sources/webhookDeliveryBatcher.server.ts
+98
View File
@@ -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<number, string[]> = { 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);
apps/webapp/app/services/triggers/registerWebhook.server.ts
+59 -1
View File
@@ -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,
},
},
});
}
}
}
}
apps/webapp/app/services/worker.server.ts
+39 -6
View File
@@ -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: {
packages/core/src/schemas/api.ts
+23 -6
View File
@@ -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<typeof WebhookSourceRequestHeadersSchema>;
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<typeof WebhookSourceRequestBodySchema>;
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<typeof AutoYieldConfigSchema>;
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<typeof WebhookDeliveryResultSchema>;
export const WebhookDeliveryResponseSchema = z.object({
response: NormalizedResponseSchema,
deliveryResults: WebhookDeliveryResultSchema.array(),
});
export type WebhookDeliveryResponse = z.infer<typeof WebhookDeliveryResponseSchema>;
export const RegisterTriggerBodySchemaV1 = z.object({
packages/core/src/schemas/triggers.ts
+3 -3
View File
@@ -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<typeof EventDispatchBatcherOptionsSchema>;
export type BatcherOptions = z.infer<typeof BatcherOptionsSchema>;
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({
packages/database/prisma/migrations/20231130233325_add_webhook_delivery_batcher/migration.sql
+35
View File
@@ -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;
packages/database/prisma/schema.prisma
+36 -6
View File
@@ -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)
packages/testing/src/index.ts
+12 -2
View File
@@ -1,4 +1,5 @@
import {
BatcherOptions,
RunJobBody,
SendEvent,
SendEventBodySchema,
@@ -85,9 +86,15 @@ const buildRequest = (action: TriggerAction, apiKey: string, opts: Record<string
});
};
const buildRequestBody = (event: RunJobBody["event"], job: RunJobBody["job"]): RunJobBody => ({
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);
packages/trigger-sdk/src/triggerClient.ts
+75 -37
View File
@@ -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<WebhookDeliveryResponse> {
): Promise<WebhookDeliveryResult> {
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,
};
}
packages/trigger-sdk/src/triggers/eventTrigger.ts
+4 -9
View File
@@ -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<TEventSpecification extends EventSpecification<any>> =
source?: string;
filter?: EventFilter;
verify?: EventTypeFromSpecification<TEventSpecification> extends Request ? VerifyCallback : never;
batch?: EventDispatchBatcherOptions;
batch?: BatcherOptions;
};
export class EventTrigger<TEventSpecification extends EventSpecification<any>>
@@ -55,7 +50,7 @@ export class EventTrigger<TEventSpecification extends EventSpecification<any>>
attachToJob(triggerClient: TriggerClient, job: Job<Trigger<TEventSpecification>, any>): void {}
batch(
options?: Exclude<EventDispatchBatcherOptions, boolean>
options?: Exclude<BatcherOptions, boolean>
): EventTrigger<TEventSpecification> {
const { batch, ...rest } = this.#options;
@@ -112,7 +107,7 @@ type TriggerOptions<TEvent> = {
*/
filter?: EventFilter;
/** Used to set batching options. */
batch?: EventDispatchBatcherOptions;
batch?: BatcherOptions;
examples?: EventSpecificationExample[];
};
packages/trigger-sdk/src/triggers/webhook.ts
+6
View File
@@ -1,4 +1,5 @@
import {
BatcherOptions,
DisplayProperty,
EventFilter,
HandleTriggerSource,
@@ -141,6 +142,7 @@ type WebhookOptions<
crud: WebhookCRUD<TIntegration, TParams, TConfig>;
filter?: FilterFunction<TParams, TConfig>;
register?: RegisterFunction<TIntegration, TParams, TConfig>;
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;
}