diff --git a/backend/src/modules/webhooks/webhooks.dispatcher.test.ts b/backend/src/modules/webhooks/webhooks.dispatcher.test.ts new file mode 100644 index 00000000..e6a92ac5 --- /dev/null +++ b/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 }); + }); +}); diff --git a/backend/src/modules/webhooks/webhooks.dispatcher.ts b/backend/src/modules/webhooks/webhooks.dispatcher.ts new file mode 100644 index 00000000..cf65cc62 --- /dev/null +++ b/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, +): Promise { + 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 }; +} diff --git a/backend/src/modules/webhooks/webhooks.schema.ts b/backend/src/modules/webhooks/webhooks.schema.ts index afeae7bc..1cc0c006 100644 --- a/backend/src/modules/webhooks/webhooks.schema.ts +++ b/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() diff --git a/backend/src/modules/webhooks/webhooks.types.ts b/backend/src/modules/webhooks/webhooks.types.ts index 1ecb50d8..012f8f01 100644 --- a/backend/src/modules/webhooks/webhooks.types.ts +++ b/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; +} + +/** 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;