webhooks.integration.test.ts

webhooks.integration.test.tstests/integration/webhooks.integration.test.tsraw ← integration46,806 B← Back to the app
1/**
2 * AI Software Factory — Integration: Webhooks production (TASK-0130).
3 *
4 * The full outbound-delivery contract through the REAL API + in-process
5 * worker (same source-import convention as automation-pipeline):
6 *
7 *   1.  delivery happy path — create page → `page.created` fan-out →
8 *       DELIVERED; envelope + ALL documented headers; in-test HMAC
9 *       recompute `v1 == HMAC-SHA256(secret, "<t>.<rawBody>")`;
10 *   2.  secret storage — `secretEncrypted` = AES-GCM ciphertext (NOT
11 *       plaintext), `secretHash` = sha256, list/get expose no raw secret
12 *       (schema test), 409 at the 20-per-workspace cap;
13 *   3.  4xx target → REJECTED (terminal, no retry);
14 *   4.  5xx target → FAILED ×8 → DEAD_LETTER + admin notification (24h
15 *       throttled); nextRetryAt gaps match `min(30s·2^(a-1),15min)` ±20%
16 *       (env-shortened: 300ms base / 2400ms cap);
17 *   5.  resubmit `all_dead` → new rows linked by originDeliveryId,
18 *       attempts reset, re-delivered;
19 *   6.  delivery retry endpoint — FAILED → PENDING → re-delivered;
20 *       DELIVERED → 409;
21 *   7.  SSRF — private/loopback delivery URLs rejected (400), allowed with
22 *       WEBHOOK_ALLOW_PRIVATE=true;
23 *   8.  test send — synchronous `webhook.test` delivery (no retry queue);
24 *   9.  secret rotation (PATCH rotateSecret) — new secret signs, old
25 *       signature invalid;
26 *  10.  catalog completeness — EVERY catalog event fires on its operation
27 *       (or the canonical stream entry) and is delivered + signed;
28 *  11.  incoming webhooks — valid → 202 + run row; bad sig → 401
29 *       signature_invalid; 6-min-old ts → 401 timestamp_expired;
30 *       replay → 401 replay_detected.
31 *
32 * Backoff env is shortened for this suite (WEBHOOK_BACKOFF_BASE_MS=300,
33 * MAX=2400) so the 8-attempt run completes in ~6 s.
34 */
35import { afterAll, beforeAll, describe, expect, it } from "vitest";
36import * as http from "node:http";
37import { createHash, createHmac } from "node:crypto";
38import { PrismaClient } from "@prisma/client";
39import { Redis } from "ioredis";
40import { WEBHOOK_EVENT_CATALOG, WEBHOOK_EVENT_NAMES, type WebhookCatalogEvent } from "@factory/shared";
41import { getDatabaseUrl, getRedisUrl } from "./helpers/containers";
42import { deploySchema } from "./helpers/api";
43import { createTestApp, type TestAppHandle } from "./helpers/api";
44import { authHeaders, createWorkspace, createUser } from "./helpers/seed";
45 
46// ── worker source (no build on the monorepo test path) ─────────────────────
47import { pathToFileURL } from "node:url";
48import path from "node:path";
49import { fileURLToPath } from "node:url";
50const HERE = path.dirname(fileURLToPath(import.meta.url));
51const WORKER_SRC = path.resolve(HERE, "..", "..", "apps", "worker", "src");
52function wurl(rel: string): string {
53  return pathToFileURL(path.join(WORKER_SRC, rel)).href;
54}
55const { createWebhooksWorker, createWebhooksConsumer } = await import(
56  /* webpackIgnore: true */ wurl("processors/webhooks.processor.ts")
57);
58 
59// ── infra gate ─────────────────────────────────────────────────────────────
60let dbUrl = "";
61let redisUrl = "";
62let infraReady = false;
63try {
64  [dbUrl, redisUrl] = await Promise.all([getDatabaseUrl(), getRedisUrl()]);
65  infraReady = true;
66} catch {
67  infraReady = false; // no Docker daemon
68}
69 
70describe.runIf(infraReady)("Webhooks (TASK-0130)", () => {
71  let app: TestAppHandle;
72  let prisma: PrismaClient;
73  let redis: Redis;
74  let env: import("../../apps/worker/src/config").WorkerEnv;
75  const startedAt = new Date();
76 
77  function baseEnv() {
78    return {
79      NODE_ENV: "test",
80      REDIS_URL: redisUrl,
81      DATABASE_URL: dbUrl,
82      S3_ENDPOINT: "http://localhost:9000",
83      S3_REGION: "us-east-1",
84      S3_BUCKET_FILES: "factory-files",
85      S3_BUCKET_THUMBS: "factory-thumbs",
86      S3_BUCKET: "factory-files",
87      S3_ACCESS_KEY: "minioadmin",
88      S3_SECRET_KEY: "minioadmin",
89      S3_PUBLIC_BASE: "http://localhost:9001",
90      SEARCH_URL: "http://localhost:9200",
91      SEARCH_USERNAME: "admin",
92      SEARCH_PASSWORD: "changeme",
93      SEARCH_VECTOR_DIMENSION: 1536,
94      SEARCH_KNN_M: 16,
95      SEARCH_KNN_EF_CONSTRUCTION: 100,
96      MASTER_ENCRYPTION_KEY: process.env.MASTER_ENCRYPTION_KEY ?? "",
97      WEBHOOK_ALLOW_PRIVATE: "true",
98      // Short backoff: 300ms base ×2 → 2.4s cap (tests the SCHEDULE math;
99      // the production 30s/15min defaults live in worker config + shared
100      // unit tests).
101      WEBHOOK_BACKOFF_BASE_MS: "300",
102      WEBHOOK_BACKOFF_MAX_MS: "2400",
103      AI_API_KEY: "",
104      AI_PROVIDER: "openai",
105      AI_BASE_URL: "",
106      AI_MODEL: "gpt-4o-mini",
107      AI_EMBEDDING_PROVIDER: "",
108      AI_EMBEDDING_MODEL: "text-embedding-3-small",
109      SMTP_HOST: "", // no mail — the Notification row is the assertion
110      SMTP_PORT: 587,
111      SMTP_USER: "",
112      SMTP_PASS: "",
113      SMTP_FROM: "noreply@factory.local",
114      SMTP_STARTTLS: true,
115      CLAMAV_HOST: "",
116      CLAMAV_PORT: 3310,
117      CLAMD_SOCKET: "",
118      FILES_WORKER_CONCURRENCY: 4,
119      CLAMAV_TIMEOUT_MS: 30_000,
120      FILES_TEMP_DIR: "",
121      FILES_TEMP_MAX_AGE_MS: 600_000,
122      FILES_TEMP_FREE_MIN_BYTES: 1,
123      PUBLIC_URL: "http://localhost:3000",
124      WORKER_CONCURRENCY: 10,
125      WORKER_ID: "webhooks-it",
126      WORKER_RETENTION_EXECUTE: "false",
127      HEALTH_PORT: 4003,
128      AUTO_RUN_MAX_RETRIES: 8,
129      AUTOMATIONS_WORKER_CONCURRENCY: 1
130    } as import("../../apps/worker/src/config").WorkerEnv;
131  }
132 
133  beforeAll(async () => {
134    process.env.WEBHOOK_ALLOW_PRIVATE = "true"; // the API's create() SSRF guard
135    app = await createTestApp();
136    env = baseEnv();
137    prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
138    await prisma.$connect();
139    redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
140  }, 180_000);
141 
142  afterAll(async () => {
143    // Wipe the webhooks queue (delayed retry jobs parked with real delays).
144    try {
145      const { Queue } = await import("bullmq");
146      const q = new Queue("webhooks", { connection: redis });
147      await q.obliterate({ force: true });
148      await q.close();
149    } catch {
150      /* best-effort */
151    }
152    await redis?.quit().catch(() => {});
153    await prisma?.$disconnect().catch(() => {});
154    await app?.close();
155  });
156 
157  // ── fixtures ────────────────────────────────────────────────────────────
158 
159  interface Captured {
160    rawBody: Buffer;
161    headers: http.IncomingHttpHeaders;
162    /** Epoch ms the receiver captured the request (backoff-schedule proof). */
163    at: number;
164  }
165  function startReceiver(statusCode: number, requests?: Captured[]): Promise<{ url: string; close: () => Promise<void> }> {
166    const list = requests ?? [];
167    const server = http.createServer((req, res) => {
168      const chunks: Buffer[] = [];
169      req.on("data", (c) => chunks.push(c));
170      req.on("end", () => {
171        list.push({ rawBody: Buffer.concat(chunks), headers: req.headers, at: Date.now() });
172        res.writeHead(statusCode, { "Content-Type": "application/json" });
173        res.end("{}");
174      });
175    });
176    return new Promise((resolve) => {
177      server.listen(0, "127.0.0.1", () => {
178        const addr = server.address();
179        const port = typeof addr === "object" && addr ? addr.port : 0;
180        resolve({
181          url: `http://127.0.0.1:${port}/hook`,
182          close: () =>
183            new Promise<void>((r) => {
184              server.closeAllConnections?.();
185              server.close(() => r());
186            })
187        });
188      });
189    });
190  }
191 
192  interface Hook {
193    headers: Record<string, string>;
194    workspaceId: string;
195    webhookId: string;
196    secret: string;
197  }
198  async function makeHook(events: string[], url: string, opts?: { secret?: string; active?: boolean; eventsWildcard?: boolean }): Promise<Hook> {
199    const user = await createUser(app.request, { name: `Hook ${Math.random().toString(36).slice(2, 6)}` });
200    const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), `WS ${Math.random().toString(36).slice(2, 6)}`);
201    const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
202      headers: authHeaders(user.accessToken!),
203      body: {
204        url,
205        events: opts?.eventsWildcard ? ["*"] : events,
206        ...(opts?.secret ? { secret: opts.secret } : {}),
207        ...(opts?.active !== undefined ? { active: opts.active } : {})
208      }
209    });
210    expect(res.status).toBe(201);
211    const body = await res.json();
212    const hook = body.data ?? body;
213    return {
214      headers: authHeaders(user.accessToken!),
215      workspaceId,
216      webhookId: hook.id,
217      secret: opts?.secret ?? hook.secret
218    };
219  }
220 
221  /** The worker: in-process consumer (stream) + BullMQ worker (queue). */
222  async function withWorker<T>(fn: () => Promise<T>): Promise<T> {
223    const consumerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
224    const workerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
225    const consumer = createWebhooksConsumer({ redis: consumerRedis, prisma, consumerId: `it-${Date.now()}` });
226    const worker = createWebhooksWorker({ connection: workerRedis, prisma, env, concurrency: 2 });
227    try {
228      await consumer.start();
229      await worker.waitUntilReady();
230      return await fn();
231    } finally {
232      await worker.close().catch(() => {});
233      await consumer.stop().catch(() => {});
234      await consumerRedis.quit().catch(() => {});
235      await workerRedis.quit().catch(() => {});
236    }
237  }
238 
239  async function waitFor<T>(
240    probe: () => Promise<T | null | undefined>,
241    pred: (v: T) => boolean,
242    timeoutMs: number,
243    what: string
244  ): Promise<T> {
245    const deadline = Date.now() + timeoutMs;
246    for (;;) {
247      const v = await probe();
248      if (v !== null && v !== undefined && pred(v as T)) return v as T;
249      if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);
250      await new Promise((r) => setTimeout(r, 250));
251    }
252  }
253 
254  /** All deliveries of a webhook newer than `since`. */
255  async function deliveriesOf(webhookId: string, since?: Date) {
256    return app.prisma.webhookDelivery.findMany({
257      where: { webhookId, ...(since ? { createdAt: { gt: since } } : {}) },
258      orderBy: { createdAt: "asc" }
259    });
260  }
261 
262  /** Recompute + verify the signature of a captured request. */
263  function expectSigned(req: Captured, secret: string, event: string) {
264    const header = req.headers["x-factory-signature"];
265    expect(typeof header, "X-Factory-Signature header present").toBe("string");
266    const m = String(header).match(/^t=(\d+),v1=([0-9a-f]{64})$/);
267    expect(m, `canonical header shape (got ${header})`).not.toBeNull();
268    const t = Number(m![1]);
269    const v1 = m![2];
270    const expected = createHmac("sha256", secret).update(`${t}.` + req.rawBody.toString("utf8")).digest("hex");
271    expect(v1, "v1 == HMAC-SHA256(secret, `<t>.<rawBody>`)").toBe(expected);
272    expect(req.headers["x-factory-event"]).toBe(event);
273    expect(String(req.headers["user-agent"])).toContain("factory-webhooks/1.0");
274  }
275 
276  // ── 1. delivery happy path ───────────────────────────────────────────────
277 
278  it("page.created → DELIVERED: envelope, headers, in-test HMAC", async () => {
279    const reqs: Captured[] = [];
280    const receiver = await startReceiver(200, reqs);
281    try {
282      await withWorker(async () => {
283        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_it_happy_0123456789abcdef" });
284        const pg = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
285          headers: hook.headers,
286          body: { title: "Hook Trigger Page" }
287        });
288        expect(pg.status).toBe(201);
289        const pageId = ((await pg.json()) as { id: string }).id;
290 
291        const d = await waitFor(
292          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
293          (row) => row.status === "DELIVERED",
294          30_000,
295          "DELIVERED row"
296        );
297        expect(d.attempts).toBe(1);
298        expect(d.statusCode).toBe(200);
299        expect(d.latencyMs).toBeGreaterThanOrEqual(0);
300 
301        // The receiver got the envelope.
302        await waitFor(() => (reqs.length ? reqs : null), (r) => r.length >= 1, 10_000, "captured request");
303        const cap = reqs[0]!;
304        const envBody = JSON.parse(cap.rawBody.toString("utf8")) as Record<string, unknown>;
305        expect(envBody.type).toBe("page.created");
306        expect(envBody.id).toMatch(/^evt_/);
307        expect(envBody.workspaceId).toBe(hook.workspaceId);
308        expect(typeof envBody.occurredAt).toBe("string");
309        const ws = envBody.workspace as Record<string, unknown>;
310        expect(ws.id).toBe(hook.workspaceId);
311        const data = envBody.data as Record<string, unknown>;
312        expect(data.pageId).toBe(pageId);
313        expect(data.title).toBe("Hook Trigger Page");
314 
315        // Signature + headers.
316        expectSigned(cap, "whsec_it_happy_0123456789abcdef", "page.created");
317        expect(cap.headers["x-factory-event-id"]).toBe(envBody.id);
318        expect(cap.headers["x-factory-delivery-id"]).toBe(d.id);
319        expect(cap.headers["x-factory-attempt"]).toBe("1");
320 
321        // The stored payload matches the wire bytes (minus re-serialization).
322        const stored = JSON.parse(JSON.stringify(d.payload)) as Record<string, unknown>;
323        expect(stored.id).toBe(envBody.id);
324        expect(stored.data).toEqual(envBody.data);
325      });
326    } finally {
327      await receiver.close();
328    }
329  }, 90_000);
330 
331  // ── 2. secret storage + schema (no raw secret on the wire) ──────────────
332 
333  it("secret: AES-GCM at rest, sha256 bookkeeping, list/get expose prefix only, 409 at 20", async () => {
334    const receiver = await startReceiver(200);
335    try {
336      const secret = "whsec_storage_test_00000000000000";
337      const hook = await makeHook(["*"], receiver.url, { secret });
338      const row = await app.prisma.webhook.findUniqueOrThrow({ where: { id: hook.webhookId } });
339 
340      // At rest: ciphertext (never plaintext), sha256 bookkeeping.
341      expect(row.secretEncrypted).toBeInstanceOf(Uint8Array);
342      expect(row.secretEncrypted!.length).toBeGreaterThan(27); // iv+tag+ct
343      expect(Buffer.from(row.secretEncrypted).toString("utf8")).not.toContain(secret);
344      expect(row.secretHash).toBe(createHash("sha256").update(secret).digest("hex"));
345      expect(row.secretHash).not.toContain(secret);
346 
347      // Wire: list + get never leak the raw secret (schema test).
348      const listRes = await app.request("GET", `/api/v1/workspaces/${hook.workspaceId}/webhooks`, { headers: hook.headers });
349      expect(listRes.status).toBe(200);
350      const listBody = await listRes.json();
351      const listText = JSON.stringify(listBody);
352      expect(listText).not.toContain(secret);
353      const items = (listBody.data ?? listBody) as Array<Record<string, unknown>>;
354      const me = items.find((i) => i.id === hook.webhookId)!;
355      expect(me.secretPrefix).toMatch(/^whsec_/);
356      expect(me).not.toHaveProperty("secret");
357      expect(me).not.toHaveProperty("secretEncrypted");
358      expect(me).not.toHaveProperty("secretHash");
359 
360      const getRes = await app.request("GET", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}`, {
361        headers: hook.headers
362      });
363      const gotBody = await getRes.json();
364      const got = gotBody.data ?? gotBody;
365      expect(JSON.stringify(got)).not.toContain(secret);
366      expect(got).not.toHaveProperty("secret");
367      expect(got.secretPrefix).toBe(me.secretPrefix);
368 
369      // 409 at the 20-per-workspace cap.
370      for (let i = 0; i < 19; i++) {
371        const r = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/webhooks`, {
372          headers: hook.headers,
373          body: { url: receiver.url, events: ["*"] }
374        });
375        expect(r.status).toBe(201);
376      }
377      const over = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/webhooks`, {
378        headers: hook.headers,
379        body: { url: receiver.url, events: ["*"] }
380      });
381      expect(over.status).toBe(409);
382      const overBody = await over.json();
383      expect((overBody.error as { code: string }).code).toBe("CONFLICT");
384    } finally {
385      await receiver.close();
386    }
387  }, 120_000);
388 
389  // ── 3. 4xx → REJECTED (terminal) ─────────────────────────────────────────
390 
391  it("4xx target → REJECTED, no retry", async () => {
392    const reqs: Captured[] = [];
393    const receiver = await startReceiver(422, reqs);
394    try {
395      await withWorker(async () => {
396        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_reject_0000000000000000" });
397        const t0 = new Date();
398        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
399          headers: hook.headers,
400          body: { title: "422 page" }
401        });
402        const d = await waitFor(
403          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
404          (row) => row.status === "REJECTED",
405          30_000,
406          "REJECTED row"
407        );
408        expect(d.attempts).toBe(1);
409        expect(d.statusCode).toBe(422);
410        expect(String(d.error)).toContain("422");
411        expect(d.nextRetryAt).toBeNull();
412        // Exactly ONE attempt — a 4xx is terminal (the endpoint's contract bug).
413        await new Promise((r) => setTimeout(r, 2_000));
414        expect(reqs.length).toBe(1);
415        expect((await deliveriesOf(hook.webhookId, t0)).length).toBe(1);
416        expectSigned(reqs[0]!, "whsec_reject_0000000000000000", "page.created");
417      });
418    } finally {
419      await receiver.close();
420    }
421  }, 90_000);
422 
423  // ── 4. 5xx ×8 → DEAD_LETTER + notification + backoff gaps ───────────────
424 
425  it("5xx ×8 → DEAD_LETTER + admin notification; nextRetryAt gaps match backoff ±20%", async () => {
426    const reqs: Captured[] = [];
427    const receiver = await startReceiver(500, reqs);
428    try {
429      await withWorker(async () => {
430        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_dlq_00000000000000000000" });
431        const t0 = new Date();
432        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
433          headers: hook.headers,
434          body: { title: "DLQ page" }
435        });
436        const d = await waitFor(
437          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
438          (row) => row.status === "DEAD_LETTER",
439          60_000,
440          "DEAD_LETTER row"
441        );
442        expect(d.attempts).toBe(8);
443        expect(d.statusCode).toBe(500);
444        expect(d.nextRetryAt).toBeNull();
445        // One row per logical delivery (attempts increment 1→8); the retries
446        // are PROVEN visible by the 8 captured attempts + their timing.
447        expect(reqs.length).toBe(8);
448 
449        // Admin notification (in-app row; no SMTP in the test env) — the
450        // workspace owner gets exactly one (24h throttle).
451        const note = await waitFor(
452          () =>
453            app.prisma.notification.findFirst({
454              where: { workspaceId: hook.workspaceId, title: "Webhook delivery failed" },
455              orderBy: { createdAt: "desc" }
456            }),
457          (n) => n !== null,
458          10_000,
459          "admin notification"
460        );
461        expect(note.type).toBe("INTEGRATION");
462        expect(String(note.body)).toContain(hook.webhookId);
463 
464        // The backoff SCHEDULE, proven from the 8 captured attempts: the gap
465        // between attempt a and a+1 is ≈ base·2^(a-1) (±20% jitter + a few
466        // seconds of BullMQ/queue/DB slack). ≥3 retries are visibly spaced.
467        const base = 300;
468        const cap = 2400;
469        for (let a = 1; a < reqs.length; a++) {
470          const gap = reqs[a]!.at - reqs[a - 1]!.at;
471          const expected = Math.min(base * 2 ** (a - 1), cap);
472          // Lower bound: never faster than 60% of the nominal delay (jitter
473          // floor −20% − slack). Upper bound: nominal + jitter + queue slack.
474          expect(gap, `gap after attempt ${a}`).toBeGreaterThanOrEqual(expected * 0.6 - 500);
475          expect(gap, `gap after attempt ${a}`).toBeLessThanOrEqual(expected * 1.2 + 5000);
476        }
477        // The final row is terminal; earlier FAILED states are subsumed into
478        // the single row's attempt counter (worker-owned retry, C-13.8d).
479        const rows = (await deliveriesOf(hook.webhookId, t0)).filter((r) => r.status === "FAILED");
480        for (const row of rows) {
481          expect(row.nextRetryAt).toBeNull(); // only the live row carries it
482        }
483      });
484    } finally {
485      await receiver.close();
486    }
487  }, 120_000);
488 
489  // ── 5. resubmit all_dead ─────────────────────────────────────────────────
490 
491  it("resubmit all_dead → new rows (originDeliveryId), attempts reset, re-delivered", async () => {
492    const reqs: Captured[] = [];
493    const receiver = await startReceiver(200, reqs);
494    try {
495      await withWorker(async () => {
496        // A dead-lettered webhook (from a previous failing receiver is gone —
497        // craft the state directly: webhook + a DEAD_LETTER delivery row).
498        const user = await createUser(app.request, { name: "Resubmit owner" });
499        const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "Resubmit WS");
500        const hookRes = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
501          headers: authHeaders(user.accessToken!),
502          body: { url: receiver.url, events: ["*"], secret: "whsec_resubmit_000000000000" }
503        });
504        const hookBody = await hookRes.json();
505        const hook = (hookBody.data ?? hookBody) as { id: string };
506        const webhookId = hook.id;
507 
508        const deadId = `dlv_dead_${Math.random().toString(36).slice(2, 10)}`;
509        const envelope = {
510          id: `evt_${Math.random().toString(36).slice(2, 10)}`,
511          type: "page.created",
512          workspaceId,
513          occurredAt: new Date().toISOString(),
514          data: { pageId: "pg_resubmit", title: "resubmit me" }
515        };
516        await app.prisma.webhookDelivery.create({
517          data: { id: deadId, webhookId, event: "page.created", payload: envelope as object, status: "DEAD_LETTER", attempts: 8, statusCode: 503, error: "non-2xx: HTTP 503" }
518        });
519 
520        const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks/${webhookId}/resubmit`, {
521          headers: authHeaders(user.accessToken!),
522          body: { all_dead: true }
523        });
524        expect(res.status).toBe(200);
525        const body = await res.json();
526        const out = body.data ?? body;
527        expect(out.resubmitted).toBe(1);
528        const newId = out.deliveryIds[0] as string;
529        expect(out.originDeliveryIds[0]).toBe(deadId);
530 
531        const newRow = await waitFor(
532          () => app.prisma.webhookDelivery.findUnique({ where: { id: newId } }),
533          (row) => row.status === "DELIVERED",
534          30_000,
535          "resubmitted delivery"
536        );
537        expect(newRow.originDeliveryId).toBe(deadId);
538        expect(newRow.attempts).toBe(1);
539        // The ORIGINAL stays put (the log keeps the history).
540        const original = await app.prisma.webhookDelivery.findUnique({ where: { id: deadId } });
541        expect(original.status).toBe("DEAD_LETTER");
542        expect(reqs.length).toBe(1);
543        expectSigned(reqs[0]!, "whsec_resubmit_000000000000", "page.created");
544        // Retry of an already-DELIVERED row → 409.
545        const dup = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks/${webhookId}/deliveries/${newId}/retry`, {
546          headers: authHeaders(user.accessToken!)
547        });
548        expect(dup.status).toBe(409);
549      });
550    } finally {
551      await receiver.close();
552    }
553  }, 90_000);
554 
555  // ── 6. per-delivery retry ─────────────────────────────────────────────────
556 
557  it("delivery retry: FAILED → PENDING → re-delivered (attempts reset)", async () => {
558    const reqs: Captured[] = [];
559    const receiver = await startReceiver(500, reqs);
560    const okReceiver = await startReceiver(200, reqs);
561    try {
562      await withWorker(async () => {
563        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_retry_0000000000000000" });
564        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
565          headers: hook.headers,
566          body: { title: "retry page" }
567        });
568        const failed = await waitFor(
569          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
570          (row) => row.status === "FAILED",
571          30_000,
572          "FAILED row"
573        );
574        // Point the webhook at a healthy receiver, then retry.
575        const patch = await app.request("PATCH", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}`, {
576          headers: hook.headers,
577          body: { url: okReceiver.url }
578        });
579        expect(patch.status).toBe(200);
580        const retry = await app.request(
581          "POST",
582          `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}/deliveries/${failed.id}/retry`,
583          { headers: hook.headers }
584        );
585        expect(retry.status).toBe(200);
586        const retried = await waitFor(
587          () => app.prisma.webhookDelivery.findUnique({ where: { id: failed.id } }),
588          (row) => row.status === "DELIVERED",
589          30_000,
590          "retried delivery"
591        );
592        // The suite runs a 300 ms backoff, so the worker's own auto-retry
593        // may race the manual one — the row is delivered either way; only
594        // the final attempt count is non-deterministic.
595        expect(retried.attempts).toBeGreaterThanOrEqual(1);
596        expect(reqs.length).toBeGreaterThanOrEqual(2); // 500 attempt + retry
597        expectSigned(reqs[reqs.length - 1]!, "whsec_retry_0000000000000000", "page.created");
598      });
599    } finally {
600      await receiver.close();
601      await okReceiver.close();
602    }
603  }, 90_000);
604 
605  // ── 7. SSRF ──────────────────────────────────────────────────────────────
606 
607  it("SSRF: private/loopback/non-http rejected; WEBHOOK_ALLOW_PRIVATE override works", async () => {
608    const user = await createUser(app.request, { name: "SSRF owner" });
609    const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "SSRF WS");
610    const H = authHeaders(user.accessToken!);
611 
612    // Flip the override OFF (the suite runs with it ON for the receivers)
613    // and confirm every private/loopback/non-http target is rejected. The
614    // API's loadEnv is cached — reset it after flipping.
615    const { resetEnvCache } = await import("@factory/api/src/config/env");
616    process.env.WEBHOOK_ALLOW_PRIVATE = "false";
617    resetEnvCache();
618    for (const url of [
619      "http://127.0.0.1:1/dead",
620      "http://10.1.2.3/x",
621      "http://192.168.0.1/x",
622      "http://169.254.169.254/latest/meta-data",
623      "http://localhost/hook",
624      "ftp://example.com/hook",
625      "not-a-url"
626    ]) {
627      const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
628        headers: H,
629        body: { url, events: ["*"] }
630      });
631      expect(res.status, url).toBe(400);
632      const body = await res.json();
633      expect((body.error as { code: string }).code, url).toBe("VALIDATION_ERROR");
634    }
635 
636    // Restore the override: the SAME loopback url is now accepted (that is
637    // what every receiver-based scenario in this suite relies on).
638    process.env.WEBHOOK_ALLOW_PRIVATE = "true";
639    resetEnvCache();
640    const allowed = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
641      headers: H,
642      body: { url: "http://127.0.0.1:1/dead", events: ["*"] }
643    });
644    expect(allowed.status).toBe(201);
645  }, 60_000);
646 
647  // ── 8. test send ─────────────────────────────────────────────────────────
648 
649  it("test send: synchronous webhook.test delivery (no retry queue)", async () => {
650    const reqs: Captured[] = [];
651    const receiver = await startReceiver(200, reqs);
652    try {
653      const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_testsend_0000000000000000" });
654      const res = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}/test`, {
655        headers: hook.headers
656      });
657      expect(res.status).toBe(200);
658      const body = await res.json();
659      const out = body.data ?? body;
660      expect(out.status).toBe("DELIVERED");
661      expect(out.statusCode).toBe(200);
662      expect(out.attempts).toBe(1);
663      expect(out.latencyMs).toBeGreaterThanOrEqual(0);
664      const cap = reqs[0]!;
665      const envBody = JSON.parse(cap.rawBody.toString("utf8")) as Record<string, unknown>;
666      expect(envBody.type).toBe("webhook.test");
667      expect(envBody.test).toBe(true);
668      expectSigned(cap, "whsec_testsend_0000000000000000", "webhook.test");
669      // The test delivery appears in the log (the inspector tab asserts it).
670      const logRes = await app.request("GET", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}/deliveries?event=webhook.test`, {
671        headers: hook.headers
672      });
673      const logBody = await logRes.json();
674      const items = (logBody.data ?? []) as Array<Record<string, unknown>>;
675      expect(items.some((i) => i.id === out.deliveryId)).toBe(true);
676    } finally {
677      await receiver.close();
678    }
679  }, 60_000);
680 
681  // ── 9. rotation ───────────────────────────────────────────────────────────
682 
683  it("rotateSecret (PATCH): new secret signs deliveries, old signature invalid", async () => {
684    const reqs: Captured[] = [];
685    const receiver = await startReceiver(200, reqs);
686    try {
687      await withWorker(async () => {
688        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_rotate_OLD_000000000000000000" });
689        const rot = await app.request("PATCH", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}`, {
690          headers: hook.headers,
691          body: { rotateSecret: true }
692        });
693        expect(rot.status).toBe(200);
694        const rotBody = await rot.json();
695        const rotOut = rotBody.data ?? rotBody;
696        expect(typeof rotOut.secret).toBe("string");
697        expect(rotOut.secret.length).toBeGreaterThanOrEqual(40);
698        const newSecret = rotOut.secret as string;
699        expect(newSecret).not.toBe(hook.secret);
700        expect(String(rotOut.warning)).toContain("cannot be retrieved again");
701 
702        // The old secret no longer decrypts — the row now carries the new one.
703        const row = await app.prisma.webhook.findUniqueOrThrow({ where: { id: hook.webhookId } });
704        expect(row.secretHash).toBe(createHash("sha256").update(newSecret).digest("hex"));
705 
706        // The next delivery signs with the NEW secret.
707        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
708          headers: hook.headers,
709          body: { title: "rotate page" }
710        });
711        await waitFor(
712          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
713          (r) => r.status === "DELIVERED",
714          30_000,
715          "post-rotation delivery"
716        );
717        expectSigned(reqs[0]!, newSecret, "page.created");
718        // The same bytes do NOT verify under the old secret.
719        const m = String(reqs[0]!.headers["x-factory-signature"])!.match(/^t=(\d+),v1=([0-9a-f]{64})$/)!;
720        const withOld = createHmac("sha256", "whsec_rotate_OLD_000000000000000000").update(`${m[1]}.` + reqs[0]!.rawBody.toString("utf8")).digest("hex");
721        expect(m[2]).not.toBe(withOld);
722      });
723    } finally {
724      await receiver.close();
725    }
726  }, 90_000);
727 
728  // ── 10. catalog completeness ─────────────────────────────────────────────
729 
730  it("catalog completeness: every catalog event fires on its operation and delivers signed", async () => {
731    const reqs: Captured[] = [];
732    const receiver = await startReceiver(200, reqs);
733    const t0 = new Date();
734    try {
735      await withWorker(async () => {
736        // The hook must live in the SAME workspace the operations fire in
737        // (the fan-out is workspace-scoped), so create the workspace first
738        // and the webhook inside it (not via makeHook's own workspace).
739        const user = await createUser(app.request, { name: "Catalog owner" });
740        const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "Catalog WS");
741        const H = authHeaders(user.accessToken!);
742        const hookRes = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
743          headers: H,
744          body: { url: receiver.url, events: ["*"], secret: "whsec_catalog_000000000000000000" }
745        });
746        expect(hookRes.status).toBe(201);
747        // Create returns a BARE object ({…shape, secret, warning}) — no wrapper.
748        const hookBody = (await hookRes.json()) as { id: string; data?: { id: string } };
749        const hook = { webhookId: hookBody.data?.id ?? hookBody.id };
750        expect(hook.webhookId).toBeTruthy();
751 
752        const seen = new Set<string>();
753        const observe = (timeoutMs: number, what: string) =>
754          waitFor(
755            async () => {
756              const rows = await app.prisma.webhookDelivery.findMany({ where: { webhookId: hook.webhookId, createdAt: { gt: t0 } } });
757              rows.forEach((r) => seen.add(r.event));
758              return seen.has(what) ? true : null;
759            },
760            (v) => v === true,
761            timeoutMs,
762            `delivery of ${what}`
763          );
764 
765        // page.created / database.created — a page with a database fires both.
766        const pg = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
767          headers: H,
768          body: { title: "Catalog page", type: "database" }
769        });
770        expect(pg.status).toBe(201);
771        const page = (await pg.json()) as { id: string; databaseId?: string };
772        await observe(20_000, "page.created");
773        await observe(20_000, "database.created");
774        const dbId = page.databaseId as string;
775 
776        // page.updated (PATCH title) — record ops come next on the db.
777        await app.request("PATCH", `/api/v1/pages/${page.id}`, { headers: H, body: { title: "Catalog page v2" } });
778        await observe(20_000, "page.updated");
779 
780        // page.deleted — trash a fresh page (page.deleted fires on trash).
781        const pgDel = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
782          headers: H,
783          body: { title: "Catalog delete me" }
784        });
785        const delPage = (await pgDel.json()) as { id: string };
786        await app.request("DELETE", `/api/v1/pages/${delPage.id}`, { headers: H });
787        await observe(20_000, "page.deleted");
788 
789        // record.created / record.updated (db seeded with Title + Status;
790        // the title is a first-class field — the event fires on any change).
791        const rec = await app.request("POST", `/api/v1/databases/${dbId}/records`, {
792          headers: H,
793          body: { title: "Catalog record" }
794        });
795        expect([200, 201]).toContain(rec.status);
796        const recDto = (await rec.json()) as { id: string };
797        expect(recDto.id).toBeTruthy();
798        await observe(20_000, "record.created");
799        const up = await app.request("PATCH", `/api/v1/records/${recDto.id}`, {
800          headers: H,
801          body: { title: "Catalog record v2" }
802        });
803        expect(up.status).toBe(200);
804        await observe(20_000, "record.updated");
805 
806        // database.updated (rename).
807        await app.request("PATCH", `/api/v1/databases/${dbId}`, { headers: H, body: { name: "Catalog db v2" } });
808        await observe(20_000, "database.updated");
809 
810        // page.shared (link share).
811        await app.request("POST", `/api/v1/shares`, {
812          headers: H,
813          body: { workspaceId, targetType: "page", targetId: page.id, kind: "link", level: "VIEW" }
814        });
815        await observe(20_000, "page.shared");
816 
817        // comment.created (page comment).
818        await app.request("POST", `/api/v1/comments`, {
819          headers: H,
820          body: { targetType: "page", targetId: page.id, body: "Catalog comment" }
821        });
822        await observe(20_000, "comment.created");
823 
824        // api_key.created (workspace key).
825        await app.request("POST", `/api/v1/workspaces/${workspaceId}/api-keys`, {
826          headers: H,
827          body: { name: "Catalog key", scopes: ["READ"] }
828        });
829        await observe(20_000, "api_key.created");
830 
831        // form.submitted (public form on the db).
832        const formRes = await app.request("POST", `/api/v1/databases/${dbId}/forms`, {
833          headers: H,
834          body: {
835            name: "Catalog form",
836            slug: `catalog-${Math.random().toString(36).slice(2, 8)}`,
837            questions: [{ type: "SHORT_TEXT", label: "Feedback" }]
838          }
839        });
840        expect(formRes.status).toBe(201);
841        const form = (await formRes.json()) as { id: string; slug: string };
842        // Public submit requires a PUBLISHED form (DRAFT → 404).
843        const pub = await app.request("PATCH", `/api/v1/forms/${form.id}`, {
844          headers: H,
845          body: { status: "PUBLISHED" }
846        });
847        expect(pub.status).toBe(200);
848        const sub = await app.request("POST", `/api/v1/public/forms/${form.slug}/submit`, { body: { answers: {} } });
849        expect(sub.status).toBeGreaterThanOrEqual(200);
850        expect(sub.status).toBeLessThan(300);
851        await observe(20_000, "form.submitted");
852 
853        // file.uploaded + automation.executed — their emitters live in
854        // pipelines this suite doesn't run (MinIO / a full automation run);
855        // drive them through the SAME stream entry shape the emitters
856        // publish (the catalog contract is about the event, not the upload).
857        for (const [event, payload] of [
858          ["file.uploaded", { workspaceId, fileId: "file_catalog", name: "catalog.bin", size: 10, mimeType: "application/octet-stream" }],
859          ["automation.executed", { workspaceId, automationId: "auto_catalog", runId: "run_catalog", status: "SUCCEEDED" }]
860        ]) {
861          await redis.xadd(
862            "factory:events",
863            "*",
864            "event",
865            event,
866            "payload",
867            JSON.stringify(payload),
868            "ts",
869            new Date().toISOString()
870          );
871          await observe(20_000, event);
872        }
873 
874        // Every catalog event observed AND delivered (the AC: a new catalog
875        // event with no emitter fails this build — the table is the catalog).
876        expect(WEBHOOK_EVENT_NAMES.length).toBe(WEBHOOK_EVENT_CATALOG.length);
877        for (const name of WEBHOOK_EVENT_NAMES) {
878          expect(seen.has(name), `catalog event ${name} fired + delivered`).toBe(true);
879        }
880 
881        // Every delivered envelope carries a valid signature (in-test
882        // recompute) — spot-check all captured requests.
883        for (const cap of reqs) {
884          const m = String(cap.headers["x-factory-signature"])!.match(/^t=(\d+),v1=([0-9a-f]{64})$/);
885          expect(m, "header shape").not.toBeNull();
886          const expected = createHmac("sha256", "whsec_catalog_000000000000000000").update(`${m![1]}.` + cap.rawBody.toString("utf8")).digest("hex");
887          expect(m![2], "v1 matches").toBe(expected);
888        }
889      });
890    } finally {
891      await receiver.close();
892    }
893  }, 240_000);
894 
895  // ── 11. incoming webhooks ─────────────────────────────────────────────────
896 
897  describe("incoming webhooks (signature + replay)", () => {
898    let incoming: { url: string; secret: string; workspaceId: string; automationId: string; headers: Record<string, string> };
899 
900    async function signedBody(secret: string, body: unknown, ts = Math.floor(Date.now() / 1000)): Promise<{ raw: string; header: string }> {
901      const raw = JSON.stringify(body);
902      const v1 = createHmac("sha256", secret).update(`${ts}.` + raw).digest("hex");
903      return { raw, header: `t=${ts},v1=${v1}` };
904    }
905    async function postIncoming(raw: string, header: string | undefined): Promise<{ status: number; body: Record<string, unknown> }> {
906      const res = await fetch(`${app.baseUrl}/api/v1${incoming.url}`, {
907        method: "POST",
908        headers: {
909          "Content-Type": "application/json",
910          ...(header !== undefined ? { "X-Factory-Signature": header } : {})
911        },
912        body: raw
913      });
914      return { status: res.status, body: (await res.json().catch(() => ({}))) as Record<string, unknown> };
915    }
916 
917    beforeAll(async () => {
918      const user = await createUser(app.request, { name: "Incoming owner" });
919      const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "Incoming WS");
920      const H = authHeaders(user.accessToken!);
921      const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/automations`, {
922        headers: H,
923        body: { name: "Incoming automation", triggers: [{ type: "WEBHOOK", config: {} }], actions: [] }
924      });
925      expect([200, 201]).toContain(res.status);
926      const body = await res.json();
927      const autom = body.data ?? body;
928      const automationId = autom.id ?? autom.automationId;
929      // Activate: a created automation starts ACTIVE (Phase 11 default) —
930      // findActiveWebhook requires ACTIVE.
931      const st = await app.request("POST", `/api/v1/automations/${automationId}/status`, {
932        headers: H,
933        body: { status: "ACTIVE" }
934      });
935      expect(st.status).toBe(200);
936      // The trigger PUT returns the generated webhook pair (shown once).
937      const trig = await app.request("PUT", `/api/v1/automations/${automationId}/triggers`, {
938        headers: H,
939        body: { triggers: [{ type: "WEBHOOK", config: {} }] }
940      });
941      expect(trig.status).toBe(200);
942      const trigBody = await trig.json();
943      const trigOut = trigBody.data ?? trigBody;
944      const wh = (trigOut.webhooks as Array<{ webhookId: string; secret: string }>)?.[0] ??
945        (trigOut.triggers as Array<{ config?: { secret?: string } }>)?.[0]?.config ?? null;
946      expect(wh, "incoming webhook pair generated").not.toBeNull();
947      const url = `/webhooks/incoming/${(wh as { webhookId: string }).webhookId}`;
948      incoming = {
949        url,
950        secret: (wh as { secret: string }).secret,
951        workspaceId,
952        automationId: String(automationId),
953        headers: H
954      };
955    });
956 
957    it("valid signature → 202 {data:{runId}} + run row + INTEGRATION audit", async () => {
958      const { raw, header } = await signedBody(incoming.secret, { hello: "factory", n: 1 });
959      const { status, body } = await postIncoming(raw, header);
960      expect(status).toBe(202);
961      expect((body.data as { runId: string }).runId).toMatch(/^[a-z0-9]+$/);
962      const runId = (body.data as { runId: string }).runId;
963      const run = await waitFor(
964        () => app.prisma.automationRun.findUnique({ where: { id: runId } }),
965        (r) => r !== null,
966        10_000,
967        "run row"
968      );
969      expect(run.workspaceId).toBe(incoming.workspaceId);
970      const audit = await app.prisma.auditLog.findFirst({
971        where: { workspaceId: incoming.workspaceId, action: "webhook.received", actorType: "INTEGRATION" }
972      });
973      expect(audit).not.toBeNull();
974    }, 60_000);
975 
976    it("bad signature → 401 signature_invalid", async () => {
977      const { raw } = await signedBody("whsec_WRONG_SECRET_00000000000000000", { hello: "x" });
978      const { status, body } = await postIncoming(raw, `t=${Math.floor(Date.now() / 1000)},v1=${"0".repeat(64)}`);
979      expect(status).toBe(401);
980      const err = body.error as { code: string; details?: Array<{ reason?: string }> };
981      expect(err.code).toBe("WEBHOOK_SIGNATURE_INVALID");
982      expect(err.details?.[0]?.reason).toBe("signature_invalid");
983    }, 30_000);
984 
985    it("6-min-old timestamp → 401 timestamp_expired", async () => {
986      const staleTs = Math.floor(Date.now() / 1000) - 6 * 60;
987      const raw = JSON.stringify({ hello: "stale" });
988      const v1 = createHmac("sha256", incoming.secret).update(`${staleTs}.` + raw).digest("hex");
989      const { status, body } = await postIncoming(raw, `t=${staleTs},v1=${v1}`);
990      expect(status).toBe(401);
991      const err = body.error as { code: string; details?: Array<{ reason?: string }> };
992      expect(err.code).toBe("WEBHOOK_TIMESTAMP_STALE");
993      expect(err.details?.[0]?.reason).toBe("timestamp_expired");
994    }, 30_000);
995 
996    it("replay (same t+body twice) → second is 401 replay_detected", async () => {
997      const { raw, header } = await signedBody(incoming.secret, { replay: "me" });
998      const first = await postIncoming(raw, header);
999      expect(first.status).toBe(202);
1000      const second = await postIncoming(raw, header);
1001      expect(second.status).toBe(401);
1002      const err = second.body.error as { code: string; details?: Array<{ reason?: string }> };
1003      expect(err.code).toBe("WEBHOOK_REPLAY_DETECTED");
1004      expect(err.details?.[0]?.reason).toBe("replay_detected");
1005    }, 30_000);
1006  });
1007});
1008 
1009describe.skipIf(!infraReady)("Webhooks infra gate", () => {
1010  it("skips when Docker/Testcontainers are unavailable", () => {});
1011});
1012