Skip to content

Commit ead2f98

Browse files
committed
fix(webhook-engine,webapp): skip Redis and worker setup when the feature is disabled
The engine singleton is constructed at boot, opening the front-gate Redis client and the worker's queue (a second Redis client plus queue-size gauges) regardless of WEBHOOK_ENABLED. Only worker.start() was flag-gated, so a deployment with webhooks off still held two Redis connections and polled queue size, and one without the webhook Redis configured would spam reconnect errors. Add an engine-level disabled option (set from WEBHOOK_ENABLED !== '1') that skips building the Redis clients and the worker entirely. The public entry points (ingest, simulateInject, replayDelivery, getJob) assert the engine is enabled and quit() no-ops, so an off deployment is genuinely inert. worker.disabled is unchanged: an ingress-only instance still builds the engine but does not start the worker loop.
1 parent c267f35 commit ead2f98

4 files changed

Lines changed: 65 additions & 3 deletions

File tree

apps/webapp/app/v3/webhookEngine.server.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ function createWebhookEngine() {
4141
const engine = new WebhookEngine({
4242
prisma: webhookPrisma,
4343
logLevel: env.WEBHOOK_ENGINE_LOG_LEVEL,
44+
disabled: env.WEBHOOK_ENABLED !== "1",
4445
redis: {
4546
host: env.WEBHOOK_WORKER_REDIS_HOST ?? "localhost",
4647
port: env.WEBHOOK_WORKER_REDIS_PORT ?? 6379,
Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
import type { WebhookDatabase } from "@trigger.dev/database";
2+
import { describe, expect, it } from "vitest";
3+
import { WebhookEngine } from "./index.js";
4+
import type { TriggerWebhookTaskCallback } from "./types.js";
5+
6+
const triggerTask: TriggerWebhookTaskCallback = async () => ({ success: true });
7+
8+
function buildDisabledEngine() {
9+
return new WebhookEngine({
10+
prisma: {} as unknown as WebhookDatabase,
11+
disabled: true,
12+
redis: { host: "127.0.0.1", port: 6379 },
13+
worker: { concurrency: 1 },
14+
triggerTask,
15+
resolveSigningSecret: async () => undefined,
16+
logLevel: "error",
17+
});
18+
}
19+
20+
describe("WebhookEngine (disabled)", () => {
21+
it("skips Redis and worker setup, and rejects public calls", async () => {
22+
const engine = buildDisabledEngine();
23+
24+
await expect(
25+
engine.ingest({
26+
opaqueId: "x",
27+
rawBytes: new TextEncoder().encode("{}"),
28+
headers: {},
29+
url: "https://example.com/webhooks/v1/ingest/x",
30+
})
31+
).rejects.toThrow(/disabled/i);
32+
33+
await expect(engine.getJob("job")).rejects.toThrow(/disabled/i);
34+
35+
await expect(engine.quit()).resolves.toBeUndefined();
36+
});
37+
});

internal-packages/webhook-engine/src/engine/index.ts

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ import { evaluateSessionKeyTemplate, walkPath as resolveBodyPath } from "./sessi
3434
const WEBHOOK_DELIVER_MAX_ATTEMPTS = webhookWorkerCatalog["webhook.deliver"].retry.maxAttempts;
3535

3636
export class WebhookEngine {
37-
private worker: Worker<typeof webhookWorkerCatalog>;
37+
private worker!: Worker<typeof webhookWorkerCatalog>;
3838
private logger: Logger;
3939
private tracer: Tracer;
4040
private meter: Meter;
@@ -50,7 +50,7 @@ export class WebhookEngine {
5050
prisma: WebhookDatabase;
5151

5252
private triggerTask: TriggerWebhookTaskCallback;
53-
private frontGate: Redis;
53+
private frontGate!: Redis;
5454
// Caches the endpoint + resolved signing secret per opaqueId so the ingest hot path skips two
5555
// Postgres reads (endpoint lookup + secret decrypt). Both are immutable per endpoint within the
5656
// TTL; a status change or secret rotation takes effect after at most the TTL.
@@ -61,7 +61,6 @@ export class WebhookEngine {
6161
options.logger ?? new Logger("WebhookEngine", (this.options.logLevel ?? "info") as any);
6262
this.prisma = options.prisma;
6363
this.triggerTask = options.triggerTask;
64-
this.frontGate = createRedisClient(options.redis);
6564

6665
this.tracer = options.tracer ?? getTracer("webhook-engine");
6766
this.meter = options.meter ?? getMeter("webhook-engine");
@@ -96,6 +95,13 @@ export class WebhookEngine {
9695
options.endpointCache?.maxSize ?? 10_000
9796
);
9897

98+
if (options.disabled) {
99+
this.logger.info("Webhook engine disabled; skipping Redis and worker setup");
100+
return;
101+
}
102+
103+
this.frontGate = createRedisClient(options.redis);
104+
99105
this.worker = new Worker({
100106
name: "webhook-engine-worker",
101107
redisOptions: {
@@ -137,8 +143,15 @@ export class WebhookEngine {
137143
}
138144
}
139145

146+
#assertEnabled(): void {
147+
if (this.options.disabled) {
148+
throw new Error('WebhookEngine is disabled: WEBHOOK_ENABLED is not "1"');
149+
}
150+
}
151+
140152
// PUBLIC ENTRY: verify inline, append-only delivery write, enqueue routing, ack.
141153
async ingest(input: IngestInput): Promise<IngestResult> {
154+
this.#assertEnabled();
142155
return startSpan(this.tracer, "webhook.ingest", async (span) => {
143156
span.setAttribute("opaqueId", input.opaqueId);
144157

@@ -317,6 +330,7 @@ export class WebhookEngine {
317330
* downstream (filter, startOn, routing, run/session) runs for real.
318331
*/
319332
async simulateInject(input: IngestInput): Promise<IngestResult> {
333+
this.#assertEnabled();
320334
return startSpan(this.tracer, "webhook.simulate", async (span) => {
321335
span.setAttribute("opaqueId", input.opaqueId);
322336

@@ -354,6 +368,7 @@ export class WebhookEngine {
354368
// fresh idempotency key is created so the run actually executes (not deduped) and the replay is
355369
// auditable; it shares the original externalDeliveryId so the two group together.
356370
async replayDelivery(input: { id: string; createdAt: Date }): Promise<ReplayResult> {
371+
this.#assertEnabled();
357372
return startSpan(this.tracer, "webhook.replay", async (span) => {
358373
span.setAttribute("deliveryId", input.id);
359374

@@ -748,10 +763,12 @@ export class WebhookEngine {
748763
}
749764

750765
async getJob(id: string) {
766+
this.#assertEnabled();
751767
return this.worker.getJob(id);
752768
}
753769

754770
async quit() {
771+
if (this.options.disabled) return;
755772
this.logger.info("Shutting down webhook engine");
756773

757774
try {

internal-packages/webhook-engine/src/engine/types.ts

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,13 @@ export interface WebhookEngineOptions {
3030
logLevel?: string;
3131
prisma: WebhookDatabase;
3232
redis: RedisOptions;
33+
/**
34+
* When true the feature is fully off: the engine skips opening its Redis clients and building the
35+
* worker, so a deployment with webhooks disabled holds no connections and does no queue polling.
36+
* Distinct from `worker.disabled`, which keeps the engine (and its front-gate Redis) for ingress
37+
* but does not start the worker loop.
38+
*/
39+
disabled?: boolean;
3340
worker: {
3441
concurrency: number;
3542
workers?: number;

0 commit comments

Comments
 (0)