webhooks.processor.ts

webhooks.processor.tsapps/worker/src/processors/webhooks.processor.tsraw ← processors19,399 B← Back to the app
1/**
2 * apps/worker — outbound webhooks (dispatcher + event fan-out), TASK-0130.
3 *
4 * Inbound webhooks (the API exposing /webhooks/incoming/…) live in apps/api —
5 * this processor is OUTBOUND only. The delivery CONTRACT (envelope,
6 * signature, transport, SSRF, backoff, 4xx/5xx split) is implemented ONCE in
7 * `@factory/shared` (`webhook-delivery.ts`) and shared with the API's
8 * synchronous test-send endpoint; this file is the BullMQ/Postgres
9 * orchestration around it.
10 *
11 * Entry points:
12 *  1. FAN-OUT (stream) — StreamConsumer on `factory:events`
13 *     (group worker-webhooks): for each catalog event, look up active
14 *     Webhook rows subscribed to it (exact name or `*`) and enqueue one
15 *     dispatch job per webhook. The stable `evt_…` id is minted ONCE here
16 *     (dedupe key — `X-Factory-Event-Id`).
17 *  2. DISPATCH (queue) — BullMQ `webhooks`, row jobs:
18 *        { kind:"row", webhookId, event, payload, deliveryId?, eventId?,
19 *          test?, resubmit? (deliveryId + originDeliveryId of the retry) }
20 *     The API enqueues the same shape for test sends + resubmits.
21 *
22 * Dispatch (row):
23 *   - load Webhook (must exist + active + subscribed; `*` = all),
24 *   - resolve the signing secret (shared `resolveSigningSecret`:
25 *     `secretEncrypted` → legacy ciphertext in `secretHash` → bare-hash),
26 *   - build the docs §3 envelope `{id: evt_…, type, workspaceId, occurredAt,
27 *     workspace:{id,name,slug}, data, dataDelta?}` (occurredAt frozen at
28 *     fan-out so retries don't mutate the payload),
29 *   - POST via the shared transport (10 s timeout, ≤3 redirects re-SSRF'd),
30 *   - 2xx → DELIVERED {statusCode, deliveredAt};
31 *     4xx → REJECTED (terminal — the endpoint's contract bug, no retry);
32 *     5xx/timeout/transport → FAILED + RE-ENQUEUE after the backoff delay
33 *     (worker-owned retry, docs §4.3 / task `min(30s·2^attempt,15min)` ±20%
34 *     jitter, env-overridable); after 8 attempts → DEAD_LETTER (terminal)
35 *     + admin notification (throttled 24 h per webhook, Redis key).
36 *
37 * BullMQ attempts are set to 1 (no built-in backoff): every failed attempt
38 * is scheduled as a DELAYED job with the exact backoff delay, so
39 * `webhook_deliveries.nextRetryAt` is always the real scheduled time and the
40 * 8-attempt cap is deterministic (C-13.8d).
41 */
42import { asJson } from "../lib/prisma.js";
43import { QUEUES, enqueue } from "../queues.js";
44import { getLogger } from "../logger.js";
45import type { WorkerEnv } from "../config.js";
46import { StreamConsumer, type StreamEntry } from "../lib/redis-stream.js";
47import { notifyWorkspaceAdmins } from "../lib/notify.js";
48import { tryParseEvent, WEBHOOK_EVENT_CATALOG } from "../events.js";
49import {
50  buildDeliveryEnvelope,
51  classifyDeliveryResult,
52  generateId,
53  newDeliveryId,
54  postWebhookDelivery,
55  resolveSigningSecret,
56  webhookBackoffMs,
57  type DeliveryEnvelopeInput
58} from "@factory/shared";
59import { Worker, type Job } from "bullmq";
60import { withJobOtel } from "../lib/job-otel.js";
61import type { Redis } from "ioredis";
62import { z } from "zod";
63import type { PrismaClient } from "@prisma/client";
64 
65const log = getLogger("webhooks");
66 
67/** 8 total attempts (1 initial + 7 retries) — docs §4.3. */
68export const WEBHOOK_MAX_ATTEMPTS = 8;
69/** 24 h admin-notification throttle per dead-lettered webhook (Redis key). */
70const NOTIFY_THROTTLE_KEY = (webhookId: string) => `webhook:dlq-notified:${webhookId}`;
71const NOTIFY_THROTTLE_SECONDS = 24 * 3600;
72 
73// ---------------------------------------------------------------------------
74// Job schema (row-based only — the send_webhook automation action does its
75// own inline fetch in the automations executor and never touches this queue)
76// ---------------------------------------------------------------------------
77 
78export const webhookJobSchema = z.object({
79  kind: z.literal("row"),
80  webhookId: z.string().min(1),
81  event: z.string().min(1),
82  payload: z.record(z.string(), z.unknown()).default({}),
83  deliveryId: z.string().min(1).optional(),
84  /** Stable envelope id (`evt_…`) — minted once at fan-out; reused on retry. */
85  eventId: z.string().min(1).optional(),
86  /** Frozen occurredAt (ISO) — set at fan-out; retries keep it. */
87  occurredAt: z.string().min(1).optional(),
88  test: z.literal(true).optional(),
89  /** Resubmit marker: the ORIGINAL delivery id this row re-delivers. */
90  originDeliveryId: z.string().min(1).optional(),
91  /** 1-based attempt counter (carried across worker-owned retries). */
92  attempt: z.number().int().min(1).optional()
93});
94export type WebhookJobData = z.infer<typeof webhookJobSchema>;
95 
96// ---------------------------------------------------------------------------
97// Subscription matching (exact name or the `*` wildcard)
98// ---------------------------------------------------------------------------
99 
100export function isWebhookSubscribed(events: string[], event: string): boolean {
101  return events.includes("*") || events.includes(event);
102}
103 
104/**
105 * Backoff for the Nth attempt (1-based) using the deployment's configured
106 * base/cap (task default 30 s / 15 min, ±20 % jitter — C-13.8d).
107 */
108function backoffFor(env: WorkerEnv, attempt: number): number {
109  return webhookBackoffMs(attempt, {
110    baseMs: env.WEBHOOK_BACKOFF_BASE_MS,
111    maxMs: env.WEBHOOK_BACKOFF_MAX_MS
112  });
113}
114 
115// ---------------------------------------------------------------------------
116// Worker
117// ---------------------------------------------------------------------------
118 
119export function createWebhooksWorker(opts: {
120  connection: Redis;
121  prisma: PrismaClient;
122  env: WorkerEnv;
123  concurrency?: number;
124  fetchImpl?: typeof fetch;
125}): Worker {
126  const { connection, prisma, env } = opts;
127  const doFetch = opts.fetchImpl ?? fetch;
128 
129  return new Worker(
130    QUEUES.webhooks,
131    withJobOtel(QUEUES.webhooks, async (job: Job) => {
132      let data: WebhookJobData;
133      try {
134        data = webhookJobSchema.parse(job.data);
135      } catch (err) {
136        log.error({ jobId: job.id, err: (err as Error).message }, "webhook job unparseable — discarding");
137        job.discard();
138        return;
139      }
140      try {
141        await dispatchRow(doFetch, prisma, connection, env, job, data);
142      } catch (err) {
143        // Unexpected crash mid-dispatch (DB outage, …): record it, dead-letter
144        // the row if it exists, and DISCARD — BullMQ's default 8-attempt
145        // backoff would re-run it with a schedule the delivery log never
146        // shows (C-13.8d: the log must reflect the real schedule).
147        log.error(
148          { jobId: job.id, webhookId: data.webhookId, event: data.event, err: (err as Error).message },
149          "webhook dispatch crashed — dead-lettering"
150        );
151        if (data.deliveryId) {
152          await prisma.webhookDelivery
153            .update({
154              where: { id: data.deliveryId },
155              data: { status: "DEAD_LETTER", error: `crashed: ${(err as Error).message}`, nextRetryAt: null }
156            })
157            .catch(() => undefined);
158        }
159        job.discard();
160      }
161    }),
162    {
163      connection,
164      concurrency: opts.concurrency ?? env.WORKER_CONCURRENCY
165      // Worker-owned retry scheduling: no BullMQ attempts/backoff — every
166      // FAILED attempt re-enqueues itself as a delayed job (see dispatchRow).
167    }
168  );
169}
170 
171/** One dispatch attempt for a row-based job. Never throws on 4xx/5xx —
172 *  terminal states are recorded, retries are re-enqueued. */
173async function dispatchRow(
174  doFetch: typeof fetch,
175  prisma: PrismaClient,
176  connection: Redis,
177  env: WorkerEnv,
178  job: Job,
179  data: WebhookJobData
180): Promise<void> {
181  const { webhookId, event } = data;
182  const isTest = Boolean(data.test);
183  const isResubmit = Boolean(data.originDeliveryId);
184  const attempt = data.attempt ?? 1;
185  const started = Date.now();
186  log.info(
187    { jobId: job.id, webhookId, event, test: isTest, resubmit: isResubmit, attempt },
188    "dispatching webhook"
189  );
190 
191  const webhook = await prisma.webhook.findUnique({ where: { id: webhookId } });
192  if (!webhook) {
193    log.warn({ jobId: job.id, webhookId }, "webhook row missing — discarding");
194    await touchDelivery(prisma, data.deliveryId, { status: "REJECTED", error: "webhook deleted" });
195    job.discard();
196    return;
197  }
198  if (!webhook.active && !isTest) {
199    log.info({ jobId: job.id, webhookId }, "webhook inactive — discarding");
200    await touchDelivery(prisma, data.deliveryId, { status: "REJECTED", error: "webhook inactive" });
201    job.discard();
202    return;
203  }
204  if (!isTest && !isWebhookSubscribed(webhook.events, event)) {
205    log.info({ jobId: job.id, webhookId, event }, "not subscribed — discarding");
206    await touchDelivery(prisma, data.deliveryId, { status: "REJECTED", error: "not subscribed" });
207    job.discard();
208    return;
209  }
210 
211  const resolved = resolveSigningSecret(env.MASTER_ENCRYPTION_KEY, webhook.secretEncrypted, webhook.secretHash);
212  if (!resolved) {
213    log.error({ jobId: job.id, webhookId }, "webhook has no resolvable signing secret — dead-lettering");
214    await touchDelivery(prisma, data.deliveryId, { status: "DEAD_LETTER", error: "signing secret unavailable" });
215    job.discard();
216    return;
217  }
218 
219  const deliveryId = data.deliveryId ?? newDeliveryId();
220  const wireEvent = isTest ? "webhook.test" : event;
221 
222  // The docs §3 envelope — `id` + `occurredAt` are frozen across retries.
223  const ws = await prisma.workspace.findUnique({
224    where: { id: webhook.workspaceId },
225    select: { id: true, name: true, slug: true }
226  });
227  const input: DeliveryEnvelopeInput = {
228    type: wireEvent,
229    workspaceId: webhook.workspaceId,
230    data: data.payload
231  };
232  if (data.eventId) input.id = data.eventId;
233  if (ws) input.workspace = ws;
234  if (data.occurredAt) input.occurredAt = data.occurredAt;
235  if (isTest) input.test = true;
236  const envelope = buildDeliveryEnvelope(input);
237 
238  // PENDING upsert: the delivery row exists across every retry; `attempts`
239  // is the attempt THIS dispatch is making (1-based).
240  await prisma.webhookDelivery.upsert({
241    where: { id: deliveryId },
242    create: {
243      id: deliveryId,
244      webhookId,
245      event: wireEvent,
246      payload: asJson(envelope),
247      status: "PENDING",
248      attempts: attempt,
249      originDeliveryId: data.originDeliveryId ?? null
250    },
251    update: { attempts: attempt, nextRetryAt: null, status: "PENDING" }
252  });
253 
254  const result = await postWebhookDelivery(
255    {
256      url: webhook.url,
257      secret: resolved.secret,
258      event: wireEvent,
259      envelope,
260      deliveryId,
261      attempt
262    },
263    { allowPrivate: env.WEBHOOK_ALLOW_PRIVATE, fetchImpl: doFetch }
264  );
265  const outcome = classifyDeliveryResult(result);
266 
267  if (outcome === "DELIVERED") {
268    await prisma.webhookDelivery.update({
269      where: { id: deliveryId },
270      data: {
271        status: "DELIVERED",
272        statusCode: result.status,
273        error: null,
274        deliveredAt: new Date(),
275        latencyMs: result.latencyMs
276      }
277    });
278    log.info(
279      {
280        jobId: job.id,
281        webhookId,
282        deliveryId,
283        status: result.status,
284        latencyMs: result.latencyMs,
285        attempt,
286        secretSource: resolved.source
287      },
288      "webhook delivered"
289    );
290    return;
291  }
292 
293  if (outcome === "REJECTED") {
294    // 4xx = the endpoint's contract bug: terminal, NO retry (docs §4.1).
295    await prisma.webhookDelivery.update({
296      where: { id: deliveryId },
297      data: {
298        status: "REJECTED",
299        statusCode: result.status,
300        error: `rejected: HTTP ${result.status}`,
301        nextRetryAt: null,
302        latencyMs: result.latencyMs
303      }
304    });
305    log.warn({ jobId: job.id, webhookId, deliveryId, status: result.status, attempt }, "webhook 4xx — rejected (no retry)");
306    return;
307  }
308 
309  // FAILED (5xx / timeout / transport): retry, or dead-letter at the cap.
310  if (attempt >= WEBHOOK_MAX_ATTEMPTS) {
311    await markDeadLetter(prisma, connection, env, webhook, deliveryId, result, attempt);
312    return;
313  }
314  const delay = backoffFor(env, attempt);
315  const nextRetryAt = new Date(Date.now() + delay);
316  await prisma.webhookDelivery.update({
317    where: { id: deliveryId },
318    data: {
319      status: "FAILED",
320      statusCode: result.status || null,
321      error: result.transportError ?? `non-2xx: HTTP ${result.status}`,
322      nextRetryAt,
323      latencyMs: result.latencyMs
324    }
325  });
326  // Re-enqueue the SAME logical delivery as a delayed job (stable jobId —
327  // BullMQ coalesces duplicates; the attempt counter rides in the payload).
328  await enqueue(connection, QUEUES.webhooks, {
329    kind: "row",
330    webhookId,
331    event,
332    payload: data.payload,
333    deliveryId,
334    eventId: data.eventId,
335    occurredAt: data.occurredAt,
336    attempt: attempt + 1,
337    originDeliveryId: data.originDeliveryId
338  // BullMQ job ids may not contain ":" ("Custom Id cannot contain :") — dash it.
339  }, { delay, jobId: `${deliveryId}-a${attempt + 1}` });
340  log.warn(
341    { jobId: job.id, webhookId, deliveryId, status: result.status, attempt, nextRetryMs: delay },
342    "webhook failed — retry scheduled"
343  );
344}
345 
346/** DEAD_LETTER terminal status + throttled (24 h) admin notification. */
347async function markDeadLetter(
348  prisma: PrismaClient,
349  connection: Redis,
350  env: WorkerEnv,
351  webhook: { id: string; workspaceId: string; url?: string },
352  deliveryId: string,
353  result: { status: number; transportError?: string; latencyMs: number },
354  attempt: number
355): Promise<void> {
356  const error = result.transportError ?? `non-2xx: HTTP ${result.status}`;
357  log.error(
358    { webhookId: webhook.id, deliveryId, status: result.status, attempt, err: error },
359    "webhook dispatch failed after all retries → DEAD_LETTER"
360  );
361  await prisma.webhookDelivery.update({
362    where: { id: deliveryId },
363    data: {
364      status: "DEAD_LETTER",
365      statusCode: result.status || null,
366      error,
367      nextRetryAt: null,
368      latencyMs: result.latencyMs
369    }
370  });
371  // Alert (in-app + email) — throttled 24 h per webhook so a persistently
372  // dead endpoint does not spam the admin inbox on every event.
373  try {
374    const key = NOTIFY_THROTTLE_KEY(webhook.id);
375    const notified = await connection.set(key, "1", "EX", NOTIFY_THROTTLE_SECONDS, "NX");
376    if (notified !== "OK") return;
377    await notifyWorkspaceAdmins(
378      prisma,
379      connection,
380      env.SMTP_HOST,
381      webhook.workspaceId,
382      "Webhook delivery failed",
383      `Webhook ${webhook.id} (${webhook.url}) could not be completed after ${attempt} attempts (${error}) and was moved to the dead letter queue. Resubmit it from the delivery log when the endpoint recovers.`,
384      { type: "webhook", id: webhook.id },
385      "INTEGRATION"
386    );
387  } catch (err) {
388    log.warn({ err: (err as Error).message }, "dead-letter notification failed (non-fatal)");
389  }
390}
391 
392async function touchDelivery(
393  prisma: PrismaClient,
394  deliveryId: string | undefined,
395  patch: Record<string, unknown>
396): Promise<void> {
397  if (!deliveryId) return;
398  await prisma.webhookDelivery
399    .update({ where: { id: deliveryId }, data: patch })
400    .catch(() => undefined);
401}
402 
403// ---------------------------------------------------------------------------
404// Fan-out consumer (domain event → per-webhook dispatch jobs)
405// ---------------------------------------------------------------------------
406 
407export function createWebhooksConsumer(opts: {
408  redis: Redis;
409  prisma: PrismaClient;
410  consumerId: string;
411}): StreamConsumer {
412  const { redis, prisma, consumerId } = opts;
413  return new StreamConsumer(
414    redis,
415    "factory:events",
416    "worker-webhooks",
417    consumerId,
418    async (entry: StreamEntry) => {
419      const parsed = tryParseEvent(entry);
420      if (!parsed) return; // poison entry: drop + XACK (logged)
421      const { envelope } = parsed;
422      if (
423        !WEBHOOK_EVENT_CATALOG.includes(
424          envelope.event as (typeof WEBHOOK_EVENT_CATALOG)[number]
425        )
426      ) {
427        return; // XACK: not a webhook-able event
428      }
429      const p = envelope.payload as Record<string, unknown>;
430      const workspaceId = p.workspaceId ? String(p.workspaceId) : null;
431      const webhooks = await prisma.webhook.findMany({
432        where: {
433          active: true,
434          OR: [{ events: { has: "*" } }, { events: { has: envelope.event } }],
435          ...(workspaceId ? { workspaceId } : {})
436        },
437        select: { id: true, events: true }
438      });
439      for (const wh of webhooks) {
440        // Defense in depth: the Prisma `has` filter is exact-match; the
441        // wildcard check lives here (and again in dispatchRow).
442        if (!isWebhookSubscribed(wh.events, envelope.event)) continue;
443        // The delivery row is created by the dispatcher (PENDING). The
444        // stable `eventId` + `occurredAt` are minted ONCE here so every
445        // retry carries the same `X-Factory-Event-Id` / envelope time.
446        await enqueue(redis, QUEUES.webhooks, {
447          kind: "row",
448          webhookId: wh.id,
449          event: envelope.event,
450          payload: envelope.payload,
451          eventId: generateId("evt"),
452          occurredAt: new Date().toISOString()
453        });
454      }
455      if (webhooks.length) {
456        log.debug({ event: envelope.event, webhooks: webhooks.length }, "fan-out enqueued");
457      }
458    }
459  );
460}
461 
462// ---------------------------------------------------------------------------
463// Final-failure hook (worker index.ts `failed` event — belt & braces:
464// dispatchRow dead-letters at the cap itself; this catches unexpected job
465// crashes so a row never stays PENDING forever).
466// ---------------------------------------------------------------------------
467 
468export function handleWebhookFinalFailure(
469  prisma: PrismaClient,
470  connection: Redis,
471  env: WorkerEnv
472): (job: Job, err: Error) => Promise<void> {
473  return async (job, err) => {
474    let data: WebhookJobData;
475    try {
476      data = webhookJobSchema.parse(job.data);
477    } catch {
478      log.error({ jobId: job.id, err: err.message }, "webhook job failed (unparseable data)");
479      return;
480    }
481    log.error(
482      {
483        jobId: job.id,
484        webhookId: data.webhookId,
485        event: data.event,
486        err: err.message
487      },
488      "webhook job crashed → dead-lettering (unexpected error)"
489    );
490    try {
491      await prisma.webhookDelivery
492        .update({
493          where: { id: data.deliveryId ?? "" },
494          data: {
495            status: "DEAD_LETTER",
496            error: `crashed: ${err.message}`,
497            nextRetryAt: null
498          }
499        })
500        .catch(() => undefined);
501      const webhook = await prisma.webhook.findUnique({
502        where: { id: data.webhookId },
503        select: { id: true, workspaceId: true }
504      });
505      if (webhook) {
506        const key = NOTIFY_THROTTLE_KEY(webhook.id);
507        const notified = await connection.set(key, "1", "EX", NOTIFY_THROTTLE_SECONDS, "NX");
508        if (notified === "OK") {
509          await notifyWorkspaceAdmins(
510            prisma,
511            connection,
512            env.SMTP_HOST,
513            webhook.workspaceId,
514            "Webhook delivery failed",
515            `A webhook delivery crashed unexpectedly (${err.message}) and was moved to the dead letter queue.`,
516            { type: "webhook", id: webhook.id },
517            "INTEGRATION"
518          );
519        }
520      }
521    } catch (inner) {
522      log.error({ err: (inner as Error).message }, "webhook crash bookkeeping failed");
523    }
524  };
525}
526