Add webhook event bus for async processing

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,7 +136,21 @@ 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>;
@@ -160,6 +174,17 @@ export interface ChargebeeOptions {
 		| 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;