/**
 * AI Software Factory — Integration: Webhooks production (TASK-0130).
 *
 * The full outbound-delivery contract through the REAL API + in-process
 * worker (same source-import convention as automation-pipeline):
 *
 *   1.  delivery happy path — create page → `page.created` fan-out →
 *       DELIVERED; envelope + ALL documented headers; in-test HMAC
 *       recompute `v1 == HMAC-SHA256(secret, "<t>.<rawBody>")`;
 *   2.  secret storage — `secretEncrypted` = AES-GCM ciphertext (NOT
 *       plaintext), `secretHash` = sha256, list/get expose no raw secret
 *       (schema test), 409 at the 20-per-workspace cap;
 *   3.  4xx target → REJECTED (terminal, no retry);
 *   4.  5xx target → FAILED ×8 → DEAD_LETTER + admin notification (24h
 *       throttled); nextRetryAt gaps match `min(30s·2^(a-1),15min)` ±20%
 *       (env-shortened: 300ms base / 2400ms cap);
 *   5.  resubmit `all_dead` → new rows linked by originDeliveryId,
 *       attempts reset, re-delivered;
 *   6.  delivery retry endpoint — FAILED → PENDING → re-delivered;
 *       DELIVERED → 409;
 *   7.  SSRF — private/loopback delivery URLs rejected (400), allowed with
 *       WEBHOOK_ALLOW_PRIVATE=true;
 *   8.  test send — synchronous `webhook.test` delivery (no retry queue);
 *   9.  secret rotation (PATCH rotateSecret) — new secret signs, old
 *       signature invalid;
 *  10.  catalog completeness — EVERY catalog event fires on its operation
 *       (or the canonical stream entry) and is delivered + signed;
 *  11.  incoming webhooks — valid → 202 + run row; bad sig → 401
 *       signature_invalid; 6-min-old ts → 401 timestamp_expired;
 *       replay → 401 replay_detected.
 *
 * Backoff env is shortened for this suite (WEBHOOK_BACKOFF_BASE_MS=300,
 * MAX=2400) so the 8-attempt run completes in ~6 s.
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import * as http from "node:http";
import { createHash, createHmac } from "node:crypto";
import { PrismaClient } from "@prisma/client";
import { Redis } from "ioredis";
import { WEBHOOK_EVENT_CATALOG, WEBHOOK_EVENT_NAMES, type WebhookCatalogEvent } from "@factory/shared";
import { getDatabaseUrl, getRedisUrl } from "./helpers/containers";
import { deploySchema } from "./helpers/api";
import { createTestApp, type TestAppHandle } from "./helpers/api";
import { authHeaders, createWorkspace, createUser } from "./helpers/seed";

// ── worker source (no build on the monorepo test path) ─────────────────────
import { pathToFileURL } from "node:url";
import path from "node:path";
import { fileURLToPath } from "node:url";
const HERE = path.dirname(fileURLToPath(import.meta.url));
const WORKER_SRC = path.resolve(HERE, "..", "..", "apps", "worker", "src");
function wurl(rel: string): string {
  return pathToFileURL(path.join(WORKER_SRC, rel)).href;
}
const { createWebhooksWorker, createWebhooksConsumer } = await import(
  /* webpackIgnore: true */ wurl("processors/webhooks.processor.ts")
);

// ── infra gate ─────────────────────────────────────────────────────────────
let dbUrl = "";
let redisUrl = "";
let infraReady = false;
try {
  [dbUrl, redisUrl] = await Promise.all([getDatabaseUrl(), getRedisUrl()]);
  infraReady = true;
} catch {
  infraReady = false; // no Docker daemon
}

describe.runIf(infraReady)("Webhooks (TASK-0130)", () => {
  let app: TestAppHandle;
  let prisma: PrismaClient;
  let redis: Redis;
  let env: import("../../apps/worker/src/config").WorkerEnv;
  const startedAt = new Date();

  function baseEnv() {
    return {
      NODE_ENV: "test",
      REDIS_URL: redisUrl,
      DATABASE_URL: dbUrl,
      S3_ENDPOINT: "http://localhost:9000",
      S3_REGION: "us-east-1",
      S3_BUCKET_FILES: "factory-files",
      S3_BUCKET_THUMBS: "factory-thumbs",
      S3_BUCKET: "factory-files",
      S3_ACCESS_KEY: "minioadmin",
      S3_SECRET_KEY: "minioadmin",
      S3_PUBLIC_BASE: "http://localhost:9001",
      SEARCH_URL: "http://localhost:9200",
      SEARCH_USERNAME: "admin",
      SEARCH_PASSWORD: "changeme",
      SEARCH_VECTOR_DIMENSION: 1536,
      SEARCH_KNN_M: 16,
      SEARCH_KNN_EF_CONSTRUCTION: 100,
      MASTER_ENCRYPTION_KEY: process.env.MASTER_ENCRYPTION_KEY ?? "",
      WEBHOOK_ALLOW_PRIVATE: "true",
      // Short backoff: 300ms base ×2 → 2.4s cap (tests the SCHEDULE math;
      // the production 30s/15min defaults live in worker config + shared
      // unit tests).
      WEBHOOK_BACKOFF_BASE_MS: "300",
      WEBHOOK_BACKOFF_MAX_MS: "2400",
      AI_API_KEY: "",
      AI_PROVIDER: "openai",
      AI_BASE_URL: "",
      AI_MODEL: "gpt-4o-mini",
      AI_EMBEDDING_PROVIDER: "",
      AI_EMBEDDING_MODEL: "text-embedding-3-small",
      SMTP_HOST: "", // no mail — the Notification row is the assertion
      SMTP_PORT: 587,
      SMTP_USER: "",
      SMTP_PASS: "",
      SMTP_FROM: "noreply@factory.local",
      SMTP_STARTTLS: true,
      CLAMAV_HOST: "",
      CLAMAV_PORT: 3310,
      CLAMD_SOCKET: "",
      FILES_WORKER_CONCURRENCY: 4,
      CLAMAV_TIMEOUT_MS: 30_000,
      FILES_TEMP_DIR: "",
      FILES_TEMP_MAX_AGE_MS: 600_000,
      FILES_TEMP_FREE_MIN_BYTES: 1,
      PUBLIC_URL: "http://localhost:3000",
      WORKER_CONCURRENCY: 10,
      WORKER_ID: "webhooks-it",
      WORKER_RETENTION_EXECUTE: "false",
      HEALTH_PORT: 4003,
      AUTO_RUN_MAX_RETRIES: 8,
      AUTOMATIONS_WORKER_CONCURRENCY: 1
    } as import("../../apps/worker/src/config").WorkerEnv;
  }

  beforeAll(async () => {
    process.env.WEBHOOK_ALLOW_PRIVATE = "true"; // the API's create() SSRF guard
    app = await createTestApp();
    env = baseEnv();
    prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
    await prisma.$connect();
    redis = new Redis(redisUrl, { maxRetriesPerRequest: null });
  }, 180_000);

  afterAll(async () => {
    // Wipe the webhooks queue (delayed retry jobs parked with real delays).
    try {
      const { Queue } = await import("bullmq");
      const q = new Queue("webhooks", { connection: redis });
      await q.obliterate({ force: true });
      await q.close();
    } catch {
      /* best-effort */
    }
    await redis?.quit().catch(() => {});
    await prisma?.$disconnect().catch(() => {});
    await app?.close();
  });

  // ── fixtures ────────────────────────────────────────────────────────────

  interface Captured {
    rawBody: Buffer;
    headers: http.IncomingHttpHeaders;
    /** Epoch ms the receiver captured the request (backoff-schedule proof). */
    at: number;
  }
  function startReceiver(statusCode: number, requests?: Captured[]): Promise<{ url: string; close: () => Promise<void> }> {
    const list = requests ?? [];
    const server = http.createServer((req, res) => {
      const chunks: Buffer[] = [];
      req.on("data", (c) => chunks.push(c));
      req.on("end", () => {
        list.push({ rawBody: Buffer.concat(chunks), headers: req.headers, at: Date.now() });
        res.writeHead(statusCode, { "Content-Type": "application/json" });
        res.end("{}");
      });
    });
    return new Promise((resolve) => {
      server.listen(0, "127.0.0.1", () => {
        const addr = server.address();
        const port = typeof addr === "object" && addr ? addr.port : 0;
        resolve({
          url: `http://127.0.0.1:${port}/hook`,
          close: () =>
            new Promise<void>((r) => {
              server.closeAllConnections?.();
              server.close(() => r());
            })
        });
      });
    });
  }

  interface Hook {
    headers: Record<string, string>;
    workspaceId: string;
    webhookId: string;
    secret: string;
  }
  async function makeHook(events: string[], url: string, opts?: { secret?: string; active?: boolean; eventsWildcard?: boolean }): Promise<Hook> {
    const user = await createUser(app.request, { name: `Hook ${Math.random().toString(36).slice(2, 6)}` });
    const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), `WS ${Math.random().toString(36).slice(2, 6)}`);
    const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
      headers: authHeaders(user.accessToken!),
      body: {
        url,
        events: opts?.eventsWildcard ? ["*"] : events,
        ...(opts?.secret ? { secret: opts.secret } : {}),
        ...(opts?.active !== undefined ? { active: opts.active } : {})
      }
    });
    expect(res.status).toBe(201);
    const body = await res.json();
    const hook = body.data ?? body;
    return {
      headers: authHeaders(user.accessToken!),
      workspaceId,
      webhookId: hook.id,
      secret: opts?.secret ?? hook.secret
    };
  }

  /** The worker: in-process consumer (stream) + BullMQ worker (queue). */
  async function withWorker<T>(fn: () => Promise<T>): Promise<T> {
    const consumerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
    const workerRedis = new Redis(redisUrl, { maxRetriesPerRequest: null });
    const consumer = createWebhooksConsumer({ redis: consumerRedis, prisma, consumerId: `it-${Date.now()}` });
    const worker = createWebhooksWorker({ connection: workerRedis, prisma, env, concurrency: 2 });
    try {
      await consumer.start();
      await worker.waitUntilReady();
      return await fn();
    } finally {
      await worker.close().catch(() => {});
      await consumer.stop().catch(() => {});
      await consumerRedis.quit().catch(() => {});
      await workerRedis.quit().catch(() => {});
    }
  }

  async function waitFor<T>(
    probe: () => Promise<T | null | undefined>,
    pred: (v: T) => boolean,
    timeoutMs: number,
    what: string
  ): Promise<T> {
    const deadline = Date.now() + timeoutMs;
    for (;;) {
      const v = await probe();
      if (v !== null && v !== undefined && pred(v as T)) return v as T;
      if (Date.now() > deadline) throw new Error(`timed out waiting for ${what}`);
      await new Promise((r) => setTimeout(r, 250));
    }
  }

  /** All deliveries of a webhook newer than `since`. */
  async function deliveriesOf(webhookId: string, since?: Date) {
    return app.prisma.webhookDelivery.findMany({
      where: { webhookId, ...(since ? { createdAt: { gt: since } } : {}) },
      orderBy: { createdAt: "asc" }
    });
  }

  /** Recompute + verify the signature of a captured request. */
  function expectSigned(req: Captured, secret: string, event: string) {
    const header = req.headers["x-factory-signature"];
    expect(typeof header, "X-Factory-Signature header present").toBe("string");
    const m = String(header).match(/^t=(\d+),v1=([0-9a-f]{64})$/);
    expect(m, `canonical header shape (got ${header})`).not.toBeNull();
    const t = Number(m![1]);
    const v1 = m![2];
    const expected = createHmac("sha256", secret).update(`${t}.` + req.rawBody.toString("utf8")).digest("hex");
    expect(v1, "v1 == HMAC-SHA256(secret, `<t>.<rawBody>`)").toBe(expected);
    expect(req.headers["x-factory-event"]).toBe(event);
    expect(String(req.headers["user-agent"])).toContain("factory-webhooks/1.0");
  }

  // ── 1. delivery happy path ───────────────────────────────────────────────

  it("page.created → DELIVERED: envelope, headers, in-test HMAC", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(200, reqs);
    try {
      await withWorker(async () => {
        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_it_happy_0123456789abcdef" });
        const pg = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
          headers: hook.headers,
          body: { title: "Hook Trigger Page" }
        });
        expect(pg.status).toBe(201);
        const pageId = ((await pg.json()) as { id: string }).id;

        const d = await waitFor(
          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
          (row) => row.status === "DELIVERED",
          30_000,
          "DELIVERED row"
        );
        expect(d.attempts).toBe(1);
        expect(d.statusCode).toBe(200);
        expect(d.latencyMs).toBeGreaterThanOrEqual(0);

        // The receiver got the envelope.
        await waitFor(() => (reqs.length ? reqs : null), (r) => r.length >= 1, 10_000, "captured request");
        const cap = reqs[0]!;
        const envBody = JSON.parse(cap.rawBody.toString("utf8")) as Record<string, unknown>;
        expect(envBody.type).toBe("page.created");
        expect(envBody.id).toMatch(/^evt_/);
        expect(envBody.workspaceId).toBe(hook.workspaceId);
        expect(typeof envBody.occurredAt).toBe("string");
        const ws = envBody.workspace as Record<string, unknown>;
        expect(ws.id).toBe(hook.workspaceId);
        const data = envBody.data as Record<string, unknown>;
        expect(data.pageId).toBe(pageId);
        expect(data.title).toBe("Hook Trigger Page");

        // Signature + headers.
        expectSigned(cap, "whsec_it_happy_0123456789abcdef", "page.created");
        expect(cap.headers["x-factory-event-id"]).toBe(envBody.id);
        expect(cap.headers["x-factory-delivery-id"]).toBe(d.id);
        expect(cap.headers["x-factory-attempt"]).toBe("1");

        // The stored payload matches the wire bytes (minus re-serialization).
        const stored = JSON.parse(JSON.stringify(d.payload)) as Record<string, unknown>;
        expect(stored.id).toBe(envBody.id);
        expect(stored.data).toEqual(envBody.data);
      });
    } finally {
      await receiver.close();
    }
  }, 90_000);

  // ── 2. secret storage + schema (no raw secret on the wire) ──────────────

  it("secret: AES-GCM at rest, sha256 bookkeeping, list/get expose prefix only, 409 at 20", async () => {
    const receiver = await startReceiver(200);
    try {
      const secret = "whsec_storage_test_00000000000000";
      const hook = await makeHook(["*"], receiver.url, { secret });
      const row = await app.prisma.webhook.findUniqueOrThrow({ where: { id: hook.webhookId } });

      // At rest: ciphertext (never plaintext), sha256 bookkeeping.
      expect(row.secretEncrypted).toBeInstanceOf(Uint8Array);
      expect(row.secretEncrypted!.length).toBeGreaterThan(27); // iv+tag+ct
      expect(Buffer.from(row.secretEncrypted).toString("utf8")).not.toContain(secret);
      expect(row.secretHash).toBe(createHash("sha256").update(secret).digest("hex"));
      expect(row.secretHash).not.toContain(secret);

      // Wire: list + get never leak the raw secret (schema test).
      const listRes = await app.request("GET", `/api/v1/workspaces/${hook.workspaceId}/webhooks`, { headers: hook.headers });
      expect(listRes.status).toBe(200);
      const listBody = await listRes.json();
      const listText = JSON.stringify(listBody);
      expect(listText).not.toContain(secret);
      const items = (listBody.data ?? listBody) as Array<Record<string, unknown>>;
      const me = items.find((i) => i.id === hook.webhookId)!;
      expect(me.secretPrefix).toMatch(/^whsec_/);
      expect(me).not.toHaveProperty("secret");
      expect(me).not.toHaveProperty("secretEncrypted");
      expect(me).not.toHaveProperty("secretHash");

      const getRes = await app.request("GET", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}`, {
        headers: hook.headers
      });
      const gotBody = await getRes.json();
      const got = gotBody.data ?? gotBody;
      expect(JSON.stringify(got)).not.toContain(secret);
      expect(got).not.toHaveProperty("secret");
      expect(got.secretPrefix).toBe(me.secretPrefix);

      // 409 at the 20-per-workspace cap.
      for (let i = 0; i < 19; i++) {
        const r = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/webhooks`, {
          headers: hook.headers,
          body: { url: receiver.url, events: ["*"] }
        });
        expect(r.status).toBe(201);
      }
      const over = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/webhooks`, {
        headers: hook.headers,
        body: { url: receiver.url, events: ["*"] }
      });
      expect(over.status).toBe(409);
      const overBody = await over.json();
      expect((overBody.error as { code: string }).code).toBe("CONFLICT");
    } finally {
      await receiver.close();
    }
  }, 120_000);

  // ── 3. 4xx → REJECTED (terminal) ─────────────────────────────────────────

  it("4xx target → REJECTED, no retry", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(422, reqs);
    try {
      await withWorker(async () => {
        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_reject_0000000000000000" });
        const t0 = new Date();
        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
          headers: hook.headers,
          body: { title: "422 page" }
        });
        const d = await waitFor(
          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
          (row) => row.status === "REJECTED",
          30_000,
          "REJECTED row"
        );
        expect(d.attempts).toBe(1);
        expect(d.statusCode).toBe(422);
        expect(String(d.error)).toContain("422");
        expect(d.nextRetryAt).toBeNull();
        // Exactly ONE attempt — a 4xx is terminal (the endpoint's contract bug).
        await new Promise((r) => setTimeout(r, 2_000));
        expect(reqs.length).toBe(1);
        expect((await deliveriesOf(hook.webhookId, t0)).length).toBe(1);
        expectSigned(reqs[0]!, "whsec_reject_0000000000000000", "page.created");
      });
    } finally {
      await receiver.close();
    }
  }, 90_000);

  // ── 4. 5xx ×8 → DEAD_LETTER + notification + backoff gaps ───────────────

  it("5xx ×8 → DEAD_LETTER + admin notification; nextRetryAt gaps match backoff ±20%", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(500, reqs);
    try {
      await withWorker(async () => {
        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_dlq_00000000000000000000" });
        const t0 = new Date();
        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
          headers: hook.headers,
          body: { title: "DLQ page" }
        });
        const d = await waitFor(
          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
          (row) => row.status === "DEAD_LETTER",
          60_000,
          "DEAD_LETTER row"
        );
        expect(d.attempts).toBe(8);
        expect(d.statusCode).toBe(500);
        expect(d.nextRetryAt).toBeNull();
        // One row per logical delivery (attempts increment 1→8); the retries
        // are PROVEN visible by the 8 captured attempts + their timing.
        expect(reqs.length).toBe(8);

        // Admin notification (in-app row; no SMTP in the test env) — the
        // workspace owner gets exactly one (24h throttle).
        const note = await waitFor(
          () =>
            app.prisma.notification.findFirst({
              where: { workspaceId: hook.workspaceId, title: "Webhook delivery failed" },
              orderBy: { createdAt: "desc" }
            }),
          (n) => n !== null,
          10_000,
          "admin notification"
        );
        expect(note.type).toBe("INTEGRATION");
        expect(String(note.body)).toContain(hook.webhookId);

        // The backoff SCHEDULE, proven from the 8 captured attempts: the gap
        // between attempt a and a+1 is ≈ base·2^(a-1) (±20% jitter + a few
        // seconds of BullMQ/queue/DB slack). ≥3 retries are visibly spaced.
        const base = 300;
        const cap = 2400;
        for (let a = 1; a < reqs.length; a++) {
          const gap = reqs[a]!.at - reqs[a - 1]!.at;
          const expected = Math.min(base * 2 ** (a - 1), cap);
          // Lower bound: never faster than 60% of the nominal delay (jitter
          // floor −20% − slack). Upper bound: nominal + jitter + queue slack.
          expect(gap, `gap after attempt ${a}`).toBeGreaterThanOrEqual(expected * 0.6 - 500);
          expect(gap, `gap after attempt ${a}`).toBeLessThanOrEqual(expected * 1.2 + 5000);
        }
        // The final row is terminal; earlier FAILED states are subsumed into
        // the single row's attempt counter (worker-owned retry, C-13.8d).
        const rows = (await deliveriesOf(hook.webhookId, t0)).filter((r) => r.status === "FAILED");
        for (const row of rows) {
          expect(row.nextRetryAt).toBeNull(); // only the live row carries it
        }
      });
    } finally {
      await receiver.close();
    }
  }, 120_000);

  // ── 5. resubmit all_dead ─────────────────────────────────────────────────

  it("resubmit all_dead → new rows (originDeliveryId), attempts reset, re-delivered", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(200, reqs);
    try {
      await withWorker(async () => {
        // A dead-lettered webhook (from a previous failing receiver is gone —
        // craft the state directly: webhook + a DEAD_LETTER delivery row).
        const user = await createUser(app.request, { name: "Resubmit owner" });
        const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "Resubmit WS");
        const hookRes = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
          headers: authHeaders(user.accessToken!),
          body: { url: receiver.url, events: ["*"], secret: "whsec_resubmit_000000000000" }
        });
        const hookBody = await hookRes.json();
        const hook = (hookBody.data ?? hookBody) as { id: string };
        const webhookId = hook.id;

        const deadId = `dlv_dead_${Math.random().toString(36).slice(2, 10)}`;
        const envelope = {
          id: `evt_${Math.random().toString(36).slice(2, 10)}`,
          type: "page.created",
          workspaceId,
          occurredAt: new Date().toISOString(),
          data: { pageId: "pg_resubmit", title: "resubmit me" }
        };
        await app.prisma.webhookDelivery.create({
          data: { id: deadId, webhookId, event: "page.created", payload: envelope as object, status: "DEAD_LETTER", attempts: 8, statusCode: 503, error: "non-2xx: HTTP 503" }
        });

        const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks/${webhookId}/resubmit`, {
          headers: authHeaders(user.accessToken!),
          body: { all_dead: true }
        });
        expect(res.status).toBe(200);
        const body = await res.json();
        const out = body.data ?? body;
        expect(out.resubmitted).toBe(1);
        const newId = out.deliveryIds[0] as string;
        expect(out.originDeliveryIds[0]).toBe(deadId);

        const newRow = await waitFor(
          () => app.prisma.webhookDelivery.findUnique({ where: { id: newId } }),
          (row) => row.status === "DELIVERED",
          30_000,
          "resubmitted delivery"
        );
        expect(newRow.originDeliveryId).toBe(deadId);
        expect(newRow.attempts).toBe(1);
        // The ORIGINAL stays put (the log keeps the history).
        const original = await app.prisma.webhookDelivery.findUnique({ where: { id: deadId } });
        expect(original.status).toBe("DEAD_LETTER");
        expect(reqs.length).toBe(1);
        expectSigned(reqs[0]!, "whsec_resubmit_000000000000", "page.created");
        // Retry of an already-DELIVERED row → 409.
        const dup = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks/${webhookId}/deliveries/${newId}/retry`, {
          headers: authHeaders(user.accessToken!)
        });
        expect(dup.status).toBe(409);
      });
    } finally {
      await receiver.close();
    }
  }, 90_000);

  // ── 6. per-delivery retry ─────────────────────────────────────────────────

  it("delivery retry: FAILED → PENDING → re-delivered (attempts reset)", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(500, reqs);
    const okReceiver = await startReceiver(200, reqs);
    try {
      await withWorker(async () => {
        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_retry_0000000000000000" });
        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
          headers: hook.headers,
          body: { title: "retry page" }
        });
        const failed = await waitFor(
          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
          (row) => row.status === "FAILED",
          30_000,
          "FAILED row"
        );
        // Point the webhook at a healthy receiver, then retry.
        const patch = await app.request("PATCH", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}`, {
          headers: hook.headers,
          body: { url: okReceiver.url }
        });
        expect(patch.status).toBe(200);
        const retry = await app.request(
          "POST",
          `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}/deliveries/${failed.id}/retry`,
          { headers: hook.headers }
        );
        expect(retry.status).toBe(200);
        const retried = await waitFor(
          () => app.prisma.webhookDelivery.findUnique({ where: { id: failed.id } }),
          (row) => row.status === "DELIVERED",
          30_000,
          "retried delivery"
        );
        // The suite runs a 300 ms backoff, so the worker's own auto-retry
        // may race the manual one — the row is delivered either way; only
        // the final attempt count is non-deterministic.
        expect(retried.attempts).toBeGreaterThanOrEqual(1);
        expect(reqs.length).toBeGreaterThanOrEqual(2); // 500 attempt + retry
        expectSigned(reqs[reqs.length - 1]!, "whsec_retry_0000000000000000", "page.created");
      });
    } finally {
      await receiver.close();
      await okReceiver.close();
    }
  }, 90_000);

  // ── 7. SSRF ──────────────────────────────────────────────────────────────

  it("SSRF: private/loopback/non-http rejected; WEBHOOK_ALLOW_PRIVATE override works", async () => {
    const user = await createUser(app.request, { name: "SSRF owner" });
    const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "SSRF WS");
    const H = authHeaders(user.accessToken!);

    // Flip the override OFF (the suite runs with it ON for the receivers)
    // and confirm every private/loopback/non-http target is rejected. The
    // API's loadEnv is cached — reset it after flipping.
    const { resetEnvCache } = await import("@factory/api/src/config/env");
    process.env.WEBHOOK_ALLOW_PRIVATE = "false";
    resetEnvCache();
    for (const url of [
      "http://127.0.0.1:1/dead",
      "http://10.1.2.3/x",
      "http://192.168.0.1/x",
      "http://169.254.169.254/latest/meta-data",
      "http://localhost/hook",
      "ftp://example.com/hook",
      "not-a-url"
    ]) {
      const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
        headers: H,
        body: { url, events: ["*"] }
      });
      expect(res.status, url).toBe(400);
      const body = await res.json();
      expect((body.error as { code: string }).code, url).toBe("VALIDATION_ERROR");
    }

    // Restore the override: the SAME loopback url is now accepted (that is
    // what every receiver-based scenario in this suite relies on).
    process.env.WEBHOOK_ALLOW_PRIVATE = "true";
    resetEnvCache();
    const allowed = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
      headers: H,
      body: { url: "http://127.0.0.1:1/dead", events: ["*"] }
    });
    expect(allowed.status).toBe(201);
  }, 60_000);

  // ── 8. test send ─────────────────────────────────────────────────────────

  it("test send: synchronous webhook.test delivery (no retry queue)", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(200, reqs);
    try {
      const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_testsend_0000000000000000" });
      const res = await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}/test`, {
        headers: hook.headers
      });
      expect(res.status).toBe(200);
      const body = await res.json();
      const out = body.data ?? body;
      expect(out.status).toBe("DELIVERED");
      expect(out.statusCode).toBe(200);
      expect(out.attempts).toBe(1);
      expect(out.latencyMs).toBeGreaterThanOrEqual(0);
      const cap = reqs[0]!;
      const envBody = JSON.parse(cap.rawBody.toString("utf8")) as Record<string, unknown>;
      expect(envBody.type).toBe("webhook.test");
      expect(envBody.test).toBe(true);
      expectSigned(cap, "whsec_testsend_0000000000000000", "webhook.test");
      // The test delivery appears in the log (the inspector tab asserts it).
      const logRes = await app.request("GET", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}/deliveries?event=webhook.test`, {
        headers: hook.headers
      });
      const logBody = await logRes.json();
      const items = (logBody.data ?? []) as Array<Record<string, unknown>>;
      expect(items.some((i) => i.id === out.deliveryId)).toBe(true);
    } finally {
      await receiver.close();
    }
  }, 60_000);

  // ── 9. rotation ───────────────────────────────────────────────────────────

  it("rotateSecret (PATCH): new secret signs deliveries, old signature invalid", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(200, reqs);
    try {
      await withWorker(async () => {
        const hook = await makeHook(["page.created"], receiver.url, { secret: "whsec_rotate_OLD_000000000000000000" });
        const rot = await app.request("PATCH", `/api/v1/workspaces/${hook.workspaceId}/webhooks/${hook.webhookId}`, {
          headers: hook.headers,
          body: { rotateSecret: true }
        });
        expect(rot.status).toBe(200);
        const rotBody = await rot.json();
        const rotOut = rotBody.data ?? rotBody;
        expect(typeof rotOut.secret).toBe("string");
        expect(rotOut.secret.length).toBeGreaterThanOrEqual(40);
        const newSecret = rotOut.secret as string;
        expect(newSecret).not.toBe(hook.secret);
        expect(String(rotOut.warning)).toContain("cannot be retrieved again");

        // The old secret no longer decrypts — the row now carries the new one.
        const row = await app.prisma.webhook.findUniqueOrThrow({ where: { id: hook.webhookId } });
        expect(row.secretHash).toBe(createHash("sha256").update(newSecret).digest("hex"));

        // The next delivery signs with the NEW secret.
        await app.request("POST", `/api/v1/workspaces/${hook.workspaceId}/pages`, {
          headers: hook.headers,
          body: { title: "rotate page" }
        });
        await waitFor(
          () => app.prisma.webhookDelivery.findFirst({ where: { webhookId: hook.webhookId }, orderBy: { createdAt: "desc" } }),
          (r) => r.status === "DELIVERED",
          30_000,
          "post-rotation delivery"
        );
        expectSigned(reqs[0]!, newSecret, "page.created");
        // The same bytes do NOT verify under the old secret.
        const m = String(reqs[0]!.headers["x-factory-signature"])!.match(/^t=(\d+),v1=([0-9a-f]{64})$/)!;
        const withOld = createHmac("sha256", "whsec_rotate_OLD_000000000000000000").update(`${m[1]}.` + reqs[0]!.rawBody.toString("utf8")).digest("hex");
        expect(m[2]).not.toBe(withOld);
      });
    } finally {
      await receiver.close();
    }
  }, 90_000);

  // ── 10. catalog completeness ─────────────────────────────────────────────

  it("catalog completeness: every catalog event fires on its operation and delivers signed", async () => {
    const reqs: Captured[] = [];
    const receiver = await startReceiver(200, reqs);
    const t0 = new Date();
    try {
      await withWorker(async () => {
        // The hook must live in the SAME workspace the operations fire in
        // (the fan-out is workspace-scoped), so create the workspace first
        // and the webhook inside it (not via makeHook's own workspace).
        const user = await createUser(app.request, { name: "Catalog owner" });
        const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "Catalog WS");
        const H = authHeaders(user.accessToken!);
        const hookRes = await app.request("POST", `/api/v1/workspaces/${workspaceId}/webhooks`, {
          headers: H,
          body: { url: receiver.url, events: ["*"], secret: "whsec_catalog_000000000000000000" }
        });
        expect(hookRes.status).toBe(201);
        // Create returns a BARE object ({…shape, secret, warning}) — no wrapper.
        const hookBody = (await hookRes.json()) as { id: string; data?: { id: string } };
        const hook = { webhookId: hookBody.data?.id ?? hookBody.id };
        expect(hook.webhookId).toBeTruthy();

        const seen = new Set<string>();
        const observe = (timeoutMs: number, what: string) =>
          waitFor(
            async () => {
              const rows = await app.prisma.webhookDelivery.findMany({ where: { webhookId: hook.webhookId, createdAt: { gt: t0 } } });
              rows.forEach((r) => seen.add(r.event));
              return seen.has(what) ? true : null;
            },
            (v) => v === true,
            timeoutMs,
            `delivery of ${what}`
          );

        // page.created / database.created — a page with a database fires both.
        const pg = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
          headers: H,
          body: { title: "Catalog page", type: "database" }
        });
        expect(pg.status).toBe(201);
        const page = (await pg.json()) as { id: string; databaseId?: string };
        await observe(20_000, "page.created");
        await observe(20_000, "database.created");
        const dbId = page.databaseId as string;

        // page.updated (PATCH title) — record ops come next on the db.
        await app.request("PATCH", `/api/v1/pages/${page.id}`, { headers: H, body: { title: "Catalog page v2" } });
        await observe(20_000, "page.updated");

        // page.deleted — trash a fresh page (page.deleted fires on trash).
        const pgDel = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
          headers: H,
          body: { title: "Catalog delete me" }
        });
        const delPage = (await pgDel.json()) as { id: string };
        await app.request("DELETE", `/api/v1/pages/${delPage.id}`, { headers: H });
        await observe(20_000, "page.deleted");

        // record.created / record.updated (db seeded with Title + Status;
        // the title is a first-class field — the event fires on any change).
        const rec = await app.request("POST", `/api/v1/databases/${dbId}/records`, {
          headers: H,
          body: { title: "Catalog record" }
        });
        expect([200, 201]).toContain(rec.status);
        const recDto = (await rec.json()) as { id: string };
        expect(recDto.id).toBeTruthy();
        await observe(20_000, "record.created");
        const up = await app.request("PATCH", `/api/v1/records/${recDto.id}`, {
          headers: H,
          body: { title: "Catalog record v2" }
        });
        expect(up.status).toBe(200);
        await observe(20_000, "record.updated");

        // database.updated (rename).
        await app.request("PATCH", `/api/v1/databases/${dbId}`, { headers: H, body: { name: "Catalog db v2" } });
        await observe(20_000, "database.updated");

        // page.shared (link share).
        await app.request("POST", `/api/v1/shares`, {
          headers: H,
          body: { workspaceId, targetType: "page", targetId: page.id, kind: "link", level: "VIEW" }
        });
        await observe(20_000, "page.shared");

        // comment.created (page comment).
        await app.request("POST", `/api/v1/comments`, {
          headers: H,
          body: { targetType: "page", targetId: page.id, body: "Catalog comment" }
        });
        await observe(20_000, "comment.created");

        // api_key.created (workspace key).
        await app.request("POST", `/api/v1/workspaces/${workspaceId}/api-keys`, {
          headers: H,
          body: { name: "Catalog key", scopes: ["READ"] }
        });
        await observe(20_000, "api_key.created");

        // form.submitted (public form on the db).
        const formRes = await app.request("POST", `/api/v1/databases/${dbId}/forms`, {
          headers: H,
          body: {
            name: "Catalog form",
            slug: `catalog-${Math.random().toString(36).slice(2, 8)}`,
            questions: [{ type: "SHORT_TEXT", label: "Feedback" }]
          }
        });
        expect(formRes.status).toBe(201);
        const form = (await formRes.json()) as { id: string; slug: string };
        // Public submit requires a PUBLISHED form (DRAFT → 404).
        const pub = await app.request("PATCH", `/api/v1/forms/${form.id}`, {
          headers: H,
          body: { status: "PUBLISHED" }
        });
        expect(pub.status).toBe(200);
        const sub = await app.request("POST", `/api/v1/public/forms/${form.slug}/submit`, { body: { answers: {} } });
        expect(sub.status).toBeGreaterThanOrEqual(200);
        expect(sub.status).toBeLessThan(300);
        await observe(20_000, "form.submitted");

        // file.uploaded + automation.executed — their emitters live in
        // pipelines this suite doesn't run (MinIO / a full automation run);
        // drive them through the SAME stream entry shape the emitters
        // publish (the catalog contract is about the event, not the upload).
        for (const [event, payload] of [
          ["file.uploaded", { workspaceId, fileId: "file_catalog", name: "catalog.bin", size: 10, mimeType: "application/octet-stream" }],
          ["automation.executed", { workspaceId, automationId: "auto_catalog", runId: "run_catalog", status: "SUCCEEDED" }]
        ]) {
          await redis.xadd(
            "factory:events",
            "*",
            "event",
            event,
            "payload",
            JSON.stringify(payload),
            "ts",
            new Date().toISOString()
          );
          await observe(20_000, event);
        }

        // Every catalog event observed AND delivered (the AC: a new catalog
        // event with no emitter fails this build — the table is the catalog).
        expect(WEBHOOK_EVENT_NAMES.length).toBe(WEBHOOK_EVENT_CATALOG.length);
        for (const name of WEBHOOK_EVENT_NAMES) {
          expect(seen.has(name), `catalog event ${name} fired + delivered`).toBe(true);
        }

        // Every delivered envelope carries a valid signature (in-test
        // recompute) — spot-check all captured requests.
        for (const cap of reqs) {
          const m = String(cap.headers["x-factory-signature"])!.match(/^t=(\d+),v1=([0-9a-f]{64})$/);
          expect(m, "header shape").not.toBeNull();
          const expected = createHmac("sha256", "whsec_catalog_000000000000000000").update(`${m![1]}.` + cap.rawBody.toString("utf8")).digest("hex");
          expect(m![2], "v1 matches").toBe(expected);
        }
      });
    } finally {
      await receiver.close();
    }
  }, 240_000);

  // ── 11. incoming webhooks ─────────────────────────────────────────────────

  describe("incoming webhooks (signature + replay)", () => {
    let incoming: { url: string; secret: string; workspaceId: string; automationId: string; headers: Record<string, string> };

    async function signedBody(secret: string, body: unknown, ts = Math.floor(Date.now() / 1000)): Promise<{ raw: string; header: string }> {
      const raw = JSON.stringify(body);
      const v1 = createHmac("sha256", secret).update(`${ts}.` + raw).digest("hex");
      return { raw, header: `t=${ts},v1=${v1}` };
    }
    async function postIncoming(raw: string, header: string | undefined): Promise<{ status: number; body: Record<string, unknown> }> {
      const res = await fetch(`${app.baseUrl}/api/v1${incoming.url}`, {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          ...(header !== undefined ? { "X-Factory-Signature": header } : {})
        },
        body: raw
      });
      return { status: res.status, body: (await res.json().catch(() => ({}))) as Record<string, unknown> };
    }

    beforeAll(async () => {
      const user = await createUser(app.request, { name: "Incoming owner" });
      const { workspaceId } = await createWorkspace(app.request, authHeaders(user.accessToken!), "Incoming WS");
      const H = authHeaders(user.accessToken!);
      const res = await app.request("POST", `/api/v1/workspaces/${workspaceId}/automations`, {
        headers: H,
        body: { name: "Incoming automation", triggers: [{ type: "WEBHOOK", config: {} }], actions: [] }
      });
      expect([200, 201]).toContain(res.status);
      const body = await res.json();
      const autom = body.data ?? body;
      const automationId = autom.id ?? autom.automationId;
      // Activate: a created automation starts ACTIVE (Phase 11 default) —
      // findActiveWebhook requires ACTIVE.
      const st = await app.request("POST", `/api/v1/automations/${automationId}/status`, {
        headers: H,
        body: { status: "ACTIVE" }
      });
      expect(st.status).toBe(200);
      // The trigger PUT returns the generated webhook pair (shown once).
      const trig = await app.request("PUT", `/api/v1/automations/${automationId}/triggers`, {
        headers: H,
        body: { triggers: [{ type: "WEBHOOK", config: {} }] }
      });
      expect(trig.status).toBe(200);
      const trigBody = await trig.json();
      const trigOut = trigBody.data ?? trigBody;
      const wh = (trigOut.webhooks as Array<{ webhookId: string; secret: string }>)?.[0] ??
        (trigOut.triggers as Array<{ config?: { secret?: string } }>)?.[0]?.config ?? null;
      expect(wh, "incoming webhook pair generated").not.toBeNull();
      const url = `/webhooks/incoming/${(wh as { webhookId: string }).webhookId}`;
      incoming = {
        url,
        secret: (wh as { secret: string }).secret,
        workspaceId,
        automationId: String(automationId),
        headers: H
      };
    });

    it("valid signature → 202 {data:{runId}} + run row + INTEGRATION audit", async () => {
      const { raw, header } = await signedBody(incoming.secret, { hello: "factory", n: 1 });
      const { status, body } = await postIncoming(raw, header);
      expect(status).toBe(202);
      expect((body.data as { runId: string }).runId).toMatch(/^[a-z0-9]+$/);
      const runId = (body.data as { runId: string }).runId;
      const run = await waitFor(
        () => app.prisma.automationRun.findUnique({ where: { id: runId } }),
        (r) => r !== null,
        10_000,
        "run row"
      );
      expect(run.workspaceId).toBe(incoming.workspaceId);
      const audit = await app.prisma.auditLog.findFirst({
        where: { workspaceId: incoming.workspaceId, action: "webhook.received", actorType: "INTEGRATION" }
      });
      expect(audit).not.toBeNull();
    }, 60_000);

    it("bad signature → 401 signature_invalid", async () => {
      const { raw } = await signedBody("whsec_WRONG_SECRET_00000000000000000", { hello: "x" });
      const { status, body } = await postIncoming(raw, `t=${Math.floor(Date.now() / 1000)},v1=${"0".repeat(64)}`);
      expect(status).toBe(401);
      const err = body.error as { code: string; details?: Array<{ reason?: string }> };
      expect(err.code).toBe("WEBHOOK_SIGNATURE_INVALID");
      expect(err.details?.[0]?.reason).toBe("signature_invalid");
    }, 30_000);

    it("6-min-old timestamp → 401 timestamp_expired", async () => {
      const staleTs = Math.floor(Date.now() / 1000) - 6 * 60;
      const raw = JSON.stringify({ hello: "stale" });
      const v1 = createHmac("sha256", incoming.secret).update(`${staleTs}.` + raw).digest("hex");
      const { status, body } = await postIncoming(raw, `t=${staleTs},v1=${v1}`);
      expect(status).toBe(401);
      const err = body.error as { code: string; details?: Array<{ reason?: string }> };
      expect(err.code).toBe("WEBHOOK_TIMESTAMP_STALE");
      expect(err.details?.[0]?.reason).toBe("timestamp_expired");
    }, 30_000);

    it("replay (same t+body twice) → second is 401 replay_detected", async () => {
      const { raw, header } = await signedBody(incoming.secret, { replay: "me" });
      const first = await postIncoming(raw, header);
      expect(first.status).toBe(202);
      const second = await postIncoming(raw, header);
      expect(second.status).toBe(401);
      const err = second.body.error as { code: string; details?: Array<{ reason?: string }> };
      expect(err.code).toBe("WEBHOOK_REPLAY_DETECTED");
      expect(err.details?.[0]?.reason).toBe("replay_detected");
    }, 30_000);
  });
});

describe.skipIf(!infraReady)("Webhooks infra gate", () => {
  it("skips when Docker/Testcontainers are unavailable", () => {});
});