feat(webhooks): add webhook event dispatcher (#998)

backend/src/modules/webhooks/webhooks.dispatcher.test.ts

@@ -0,0 +1,121 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+vi.mock("../../common/utils/logger.js", () => ({
+  logger: { info: vi.fn(), error: vi.fn(), warn: vi.fn(), debug: vi.fn() },
+}));
+
+vi.mock("../../db/prisma.js", () => ({
+  prisma: {
+    webhookSubscription: { findMany: vi.fn() },
+    webhookDelivery: { create: vi.fn() },
+  },
+}));
+
+vi.mock("../../jobs/webhookDelivery.js", () => ({
+  scheduleWebhookDelivery: vi.fn(),
+}));
+
+import { dispatchWebhookEvent } from "./webhooks.dispatcher.js";
+import { prisma } from "../../db/prisma.js";
+import { scheduleWebhookDelivery } from "../../jobs/webhookDelivery.js";
+
+const subA = {
+  id: "wh_sub_a",
+  ownerId: "user_01",
+  url: "https://a.example.com/webhook",
+  secret: "secret-a",
+  events: ["tip.received"],
+  status: "ACTIVE",
+  deletedAt: null,
+};
+
+const subB = {
+  id: "wh_sub_b",
+  ownerId: "user_01",
+  url: "https://b.example.com/webhook",
+  secret: "secret-b",
+  events: ["tip.received", "goal.completed"],
+  status: "ACTIVE",
+  deletedAt: null,
+};
+
+describe("dispatchWebhookEvent (issue #998)", () => {
+  beforeEach(() => vi.clearAllMocks());
+
+  it("returns zero matches when no subscription is registered for the event", async () => {
+    vi.mocked(prisma.webhookSubscription.findMany).mockResolvedValueOnce([]);
+
+    const result = await dispatchWebhookEvent("user_01", "tip.received", { tipId: "tip_1" });
+
+    expect(result).toEqual({ matched: 0, dispatched: 0 });
+    expect(prisma.webhookDelivery.create).not.toHaveBeenCalled();
+    expect(scheduleWebhookDelivery).not.toHaveBeenCalled();
+  });
+
+  it("only queries subscriptions owned by the caller, ACTIVE, not deleted, matching the event", async () => {
+    vi.mocked(prisma.webhookSubscription.findMany).mockResolvedValueOnce([]);
+
+    await dispatchWebhookEvent("user_01", "tip.received", {});
+
+    expect(prisma.webhookSubscription.findMany).toHaveBeenCalledWith({
+      where: {
+        ownerId: "user_01",
+        status: "ACTIVE",
+        deletedAt: null,
+        events: { has: "tip.received" },
+      },
+    });
+  });
+
+  it("creates a PENDING delivery and enqueues a signed job for every matching subscription", async () => {
+    vi.mocked(prisma.webhookSubscription.findMany).mockResolvedValueOnce([subA, subB] as never);
+
+    const result = await dispatchWebhookEvent("user_01", "tip.received", { tipId: "tip_1" });
+
+    expect(result).toEqual({ matched: 2, dispatched: 2 });
+    expect(prisma.webhookDelivery.create).toHaveBeenCalledTimes(2);
+    expect(prisma.webhookDelivery.create).toHaveBeenCalledWith({
+      data: { subscriptionId: "wh_sub_a" },
+    });
+    expect(prisma.webhookDelivery.create).toHaveBeenCalledWith({
+      data: { subscriptionId: "wh_sub_b" },
+    });
+
+    expect(scheduleWebhookDelivery).toHaveBeenCalledTimes(2);
+    expect(scheduleWebhookDelivery).toHaveBeenCalledWith(
+      subA.url,
+      expect.objectContaining({
+        event: "tip.received",
+        timestamp: expect.any(String),
+        data: { tipId: "tip_1" },
+      }),
+      subA.secret,
+    );
+    expect(scheduleWebhookDelivery).toHaveBeenCalledWith(
+      subB.url,
+      expect.objectContaining({ event: "tip.received", data: { tipId: "tip_1" } }),
+      subB.secret,
+    );
+  });
+
+  it("sends a signed envelope with an ISO timestamp so consumers can verify authenticity", async () => {
+    vi.mocked(prisma.webhookSubscription.findMany).mockResolvedValueOnce([subA] as never);
+
+    await dispatchWebhookEvent("user_01", "tip.received", { tipId: "tip_1" });
+
+    const [, envelope, secret] = vi.mocked(scheduleWebhookDelivery).mock.calls[0];
+    expect(secret).toBe(subA.secret);
+    expect(() => new Date((envelope as { timestamp: string }).timestamp).toISOString()).not.toThrow();
+  });
+
+  it("dispatches to remaining subscriptions when one fails to enqueue", async () => {
+    vi.mocked(prisma.webhookSubscription.findMany).mockResolvedValueOnce([subA, subB] as never);
+    vi.mocked(scheduleWebhookDelivery)
+      .mockRejectedValueOnce(new Error("queue unavailable"))
+      .mockResolvedValueOnce(undefined);
+
+    const result = await dispatchWebhookEvent("user_01", "tip.received", {});
+
+    expect(result).toEqual({ matched: 2, dispatched: 1 });
+  });
+});

backend/src/modules/webhooks/webhooks.dispatcher.ts

@@ -0,0 +1,68 @@
+import { prisma } from "../../db/prisma.js";
+import { logger } from "../../common/utils/logger.js";
+import { scheduleWebhookDelivery } from "../../jobs/webhookDelivery.js";
+import type { WebhookEventType } from "./webhooks.schema.js";
+import type { WebhookEventEnvelope, WebhookDispatchResult } from "./webhooks.types.js";
+
+/**
+ * Fans a domain event (e.g. "tip.received") out to every ACTIVE, non-deleted
+ * webhook subscription owned by `ownerId` that is registered for it.
+ *
+ * For each matching subscription this records a `PENDING` `WebhookDelivery`
+ * row (so `GET /webhooks/deliveries` has something to show) and enqueues a
+ * signed HTTP delivery job via the existing `webhook-delivery` queue. One
+ * subscription failing to enqueue never blocks the others.
+ */
+export async function dispatchWebhookEvent(
+  ownerId: string,
+  event: WebhookEventType,
+  data: Record<string, unknown>,
+): Promise<WebhookDispatchResult> {
+  const subscriptions = await prisma.webhookSubscription.findMany({
+    where: {
+      ownerId,
+      status: "ACTIVE",
+      deletedAt: null,
+      events: { has: event },
+    },
+  });
+
+  if (subscriptions.length === 0) {
+    logger.info({ ownerId, event }, "No active webhook subscriptions matched event");
+    return { matched: 0, dispatched: 0 };
+  }
+
+  const envelope: WebhookEventEnvelope = {
+    event,
+    timestamp: new Date().toISOString(),
+    data,
+  };
+
+  const outcomes = await Promise.allSettled(
+    subscriptions.map(async (subscription) => {
+      await prisma.webhookDelivery.create({
+        data: { subscriptionId: subscription.id },
+      });
+      await scheduleWebhookDelivery(subscription.url, envelope, subscription.secret);
+    }),
+  );
+
+  let dispatched = 0;
+  outcomes.forEach((outcome, index) => {
+    if (outcome.status === "fulfilled") {
+      dispatched += 1;
+    } else {
+      logger.error(
+        { ownerId, event, subscriptionId: subscriptions[index].id, err: outcome.reason },
+        "Failed to dispatch webhook event to subscription",
+      );
+    }
+  });
+
+  logger.info(
+    { ownerId, event, matched: subscriptions.length, dispatched },
+    "Dispatched webhook event",
+  );
+
+  return { matched: subscriptions.length, dispatched };
+}

backend/src/modules/webhooks/webhooks.schema.ts

@@ -9,6 +9,9 @@ export const WEBHOOK_EVENT_TYPES = [
   "credit_score.updated",
 ] as const;
 
+/** Union of valid webhook event type strings. */
+export type WebhookEventType = (typeof WEBHOOK_EVENT_TYPES)[number];
+
 export const createWebhookSubscriptionSchema = z.object({
   url: z
     .string()

backend/src/modules/webhooks/webhooks.types.ts

@@ -1,6 +1,21 @@
+import type { WebhookEventType } from "./webhooks.schema.js";
+
 /** Lifecycle status of a webhook subscription. Mirrors the Prisma enum. */
 export type WebhookSubscriptionStatus = "ACTIVE" | "DISABLED";
 
+/** The signed envelope sent as the body of every webhook delivery. */
+export interface WebhookEventEnvelope {
+  event: WebhookEventType;
+  timestamp: string;
+  data: Record<string, unknown>;
+}
+
+/** Result of fanning an event out to matching subscriptions. */
+export interface WebhookDispatchResult {
+  matched: number;
+  dispatched: number;
+}
+
 /** A registered webhook subscription (secret omitted — only returned on creation). */
 export interface WebhookSubscriptionResponse {
   id: string;