add webhook event bus for async processing

Author: cb-srinaths

packages/better-auth/src/index.ts

@@ -212,8 +212,15 @@ export { CHARGEBEE_ERROR_CODES } from "./error-codes";
 export type {
 	ChargebeeOptions,
 	ChargebeePlan,
+	ChargebeeWebhookEventBus,
 	Subscription,
 	SubscriptionOptions,
 	SubscriptionStatus,
+	WebhookEvent,
 	WithChargebeeCustomerId,
 } from "./types";
+export {
+	type ChargebeeWebhookProcessor,
+	type ChargebeeWebhookProcessorSource,
+	createChargebeeWebhookProcessor,
+} from "./webhook-processor";

packages/better-auth/src/routes.ts

@@ -23,7 +23,10 @@ import {
 	isActiveOrTrialing,
 	isPendingCancel,
 } from "./utils";
-import { createWebhookHandler } from "./webhook-handler";
+import {
+	createWebhookHandler,
+	createWebhookPublishHandler,
+} from "./webhook-handler";
 
 export function getWebhookEndpoint(options: ChargebeeOptions) {
 	return createAuthEndpoint(
@@ -33,22 +36,30 @@ export function getWebhookEndpoint(options: ChargebeeOptions) {
 			metadata: { isAction: false },
 		},
 		async (ctx) => {
-			// Create webhook handler with better-auth context
-			const handler = createWebhookHandler(
-				options,
-				{
-					context: ctx.context as Record<string, unknown>,
-					adapter: ctx.context.adapter as unknown as {
-						findOne: <T = unknown>(params: unknown) => Promise<T | null>;
-						findMany: <T = unknown>(params: unknown) => Promise<T[]>;
-						update: (params: unknown) => Promise<unknown>;
-						deleteMany: (params: unknown) => Promise<void>;
-						create: (params: unknown) => Promise<unknown>;
-					},
-					logger: ctx.context.logger,
-				},
-				ctx as any,
-			);
+			// When an event bus is configured, validate + parse the event and
+			// forward it to the bus (e.g. an application queue) instead of running
+			// the DB-sync hooks inline. Otherwise process the event synchronously.
+			const handler = options.webhookEventBus
+				? createWebhookPublishHandler(
+						options,
+						options.webhookEventBus,
+						ctx.context.logger,
+					)
+				: createWebhookHandler(
+						options,
+						{
+							context: ctx.context as Record<string, unknown>,
+							adapter: ctx.context.adapter as unknown as {
+								findOne: <T = unknown>(params: unknown) => Promise<T | null>;
+								findMany: <T = unknown>(params: unknown) => Promise<T[]>;
+								update: (params: unknown) => Promise<unknown>;
+								deleteMany: (params: unknown) => Promise<void>;
+								create: (params: unknown) => Promise<unknown>;
+							},
+							logger: ctx.context.logger,
+						},
+						ctx as any,
+					);
 
 			// Let user register custom event listeners on the handler
 			options.webhookHandler?.(handler);

packages/better-auth/src/types.ts

@@ -2,8 +2,8 @@ import type { Session, User } from "better-auth";
 import type { Organization } from "better-auth/plugins/organization";
 import type Chargebee from "chargebee";
 import type {
-	Event as ChargebeeEvent,
 	Subscription as ChargebeeSubscription,
+	WebhookEvent as ChargebeeWebhookEvent,
 	Customer,
 	WebhookHandler,
 } from "chargebee";
@@ -136,43 +136,68 @@ export type SubscriptionOptions = {
 	) => Promise<boolean>;
 };
 
-export type WebhookEvent = ChargebeeEvent;
+export type WebhookEvent = ChargebeeWebhookEvent;
+
+/**
+ * Event bus seam used to decouple webhook ingestion from processing.
+ *
+ * When provided via {@link ChargebeeOptions.webhookEventBus}, the webhook
+ * endpoint validates and parses each incoming Chargebee event and then calls
+ * `publish` instead of running the DB-sync hooks inline. The application is
+ * expected to push the event onto its own queue and later process it from a
+ * consumer using `createChargebeeWebhookProcessor`.
+ */
+export interface ChargebeeWebhookEventBus {
+	/** Called at the HTTP endpoint for every validated, parsed event. */
+	publish(event: WebhookEvent): Promise<void> | void;
+}
 
 // Use native Chargebee customer creation params
 export type ChargebeeCustomerCreateParams = Partial<Customer.CreateInputParam>;
 
 export interface ChargebeeOptions {
-	chargebeeClient: InstanceType<typeof Chargebee>;
-	webhookUsername?: string;
-	webhookPassword?: string;
-	createCustomerOnSignUp?: boolean;
-	/**
-	 * Return additional params to pass to `cb.customer.create` for user customers.
-	 * Use this to pass fields like `first_name`, `last_name`, or any other
-	 * Chargebee customer params. The `ctx` argument is only available when the
-	 * customer is created on-demand (e.g. at subscription time), not during sign-up.
-	 */
-	getCustomerCreateParams?: (
-		user: User,
-		ctx?: Record<string, unknown>,
-	) =>
-		| Promise<Partial<ChargebeeCustomerCreateParams>>
-		| Partial<ChargebeeCustomerCreateParams>;
-	onCustomerCreate?: (params: CustomerCreateParams) => Promise<void> | void;
-	webhookHandler?: (handler: WebhookHandler) => void;
-	subscription?: SubscriptionOptions;
-	organization?: {
-		enabled: boolean;
+		chargebeeClient: InstanceType<typeof Chargebee>;
+		webhookUsername?: string;
+		webhookPassword?: string;
+		createCustomerOnSignUp?: boolean;
+		/**
+		 * Return additional params to pass to `cb.customer.create` for user customers.
+		 * Use this to pass fields like `first_name`, `last_name`, or any other
+		 * Chargebee customer params. The `ctx` argument is only available when the
+		 * customer is created on-demand (e.g. at subscription time), not during sign-up.
+		 */
 		getCustomerCreateParams?: (
-			organization: Organization & WithChargebeeCustomerId,
-			ctx: Record<string, unknown>,
-		) => Promise<Partial<ChargebeeCustomerCreateParams>>;
-		onCustomerCreate?: (
-			params: OrganizationCustomerCreateParams,
-			ctx: Record<string, unknown>,
-		) => Promise<void> | void;
-	};
-}
+			user: User,
+			ctx?: Record<string, unknown>,
+		) =>
+			| Promise<Partial<ChargebeeCustomerCreateParams>>
+			| Partial<ChargebeeCustomerCreateParams>;
+		onCustomerCreate?: (params: CustomerCreateParams) => Promise<void> | void;
+		webhookHandler?: (handler: WebhookHandler) => void;
+		/**
+		 * Optional event bus used to decouple webhook ingestion from processing.
+		 *
+		 * When set, the webhook endpoint in the app is exptected to validate and
+		 * parses each event and calls `webhookEventBus.publish(event)` (typically pushing it onto an application
+		 * queue) instead of running the DB-sync hooks inline. Process queued events
+		 * later with `createChargebeeWebhookProcessor`.
+		 *
+		 * When not set, events are processed synchronously within the request.
+		 */
+		webhookEventBus?: ChargebeeWebhookEventBus;
+		subscription?: SubscriptionOptions;
+		organization?: {
+			enabled: boolean;
+			getCustomerCreateParams?: (
+				organization: Organization & WithChargebeeCustomerId,
+				ctx: Record<string, unknown>,
+			) => Promise<Partial<ChargebeeCustomerCreateParams>>;
+			onCustomerCreate?: (
+				params: OrganizationCustomerCreateParams,
+				ctx: Record<string, unknown>,
+			) => Promise<void> | void;
+		};
+	}
 
 export interface Subscription {
 	id: string;