feat: add webhook producer to service-utils

Author: arcoraven

.changeset/slimy-tigers-trade.md

@@ -0,0 +1,5 @@
+---
+"@thirdweb-dev/service-utils": minor
+---
+
+feat: Add webhook producer to service-utils

packages/service-utils/README-shared-producer.md

@@ -0,0 +1,67 @@
+# Shared KafkaProducer Pattern
+
+When using multiple event producers in the same service (e.g., `UsageV2Producer` and `WebhookEventProducer`), it's recommended to share a single `KafkaProducer` instance to avoid multiple TCP connections.
+
+## Recommended Usage
+
+```typescript
+import { KafkaProducer, UsageV2Producer, WebhookEventProducer } from '@thirdweb-dev/service-utils/node';
+
+// Create one shared KafkaProducer
+const kafkaProducer = new KafkaProducer({
+  producerName: "my-service",
+  kafkaServers: process.env.KAFKA_SERVERS,
+  username: process.env.KAFKA_USERNAME,
+  password: process.env.KAFKA_PASSWORD,
+});
+
+// Share it between producers
+const usageProducer = new UsageV2Producer({
+  kafkaProducer,
+  source: "storage"
+});
+
+const webhookProducer = new WebhookEventProducer({
+  kafkaProducer
+});
+
+// Use the producers
+await Promise.all([
+  usageProducer.sendEvents([...]),
+  webhookProducer.sendEvents([...])
+]);
+
+// Shutdown: disconnect the shared producer directly
+await kafkaProducer.disconnect();
+```
+
+## Legacy Usage (Still Supported)
+
+Each producer can still create its own connection:
+
+```typescript
+const usageProducer = new UsageV2Producer({
+  producerName: "my-service",
+  kafkaServers: process.env.KAFKA_SERVERS,
+  source: "storage",
+  username: process.env.KAFKA_USERNAME,
+  password: process.env.KAFKA_PASSWORD,
+});
+
+const webhookProducer = new WebhookEventProducer({
+  producerName: "my-service",
+  kafkaServers: process.env.KAFKA_SERVERS,
+  username: process.env.KAFKA_USERNAME,
+  password: process.env.KAFKA_PASSWORD,
+});
+
+// Note: With legacy usage, you need to disconnect each producer's internal KafkaProducer
+// This is less efficient and not recommended for services using multiple producers
+```
+
+## Benefits of Shared Producer
+
+- **Resource efficiency**: One TCP connection instead of multiple
+- **Consistent configuration**: Single source of connection config
+- **Simplified shutdown**: One disconnect call instead of multiple
+- **Better performance**: Reduced connection overhead

packages/service-utils/src/node/index.ts

@@ -19,6 +19,7 @@ export * from "../core/usage.js";
 export * from "../core/usageV2.js";
 export * from "./kafka.js";
 export * from "./usageV2.js";
+export * from "./webhookProducer.js";
 
 type NodeServiceConfig = CoreServiceConfig;
 

packages/service-utils/src/node/usageV2.ts

@@ -6,48 +6,59 @@ import {
 } from "../core/usageV2.js";
 import { KafkaProducer } from "./kafka.js";
 
-const TEAM_ID_PREFIX = "team_";
-const PROJECT_ID_PREFIX = "prj_";
-
 /**
- * Creates a UsageV2Producer which opens a persistent TCP connection.
- * This class is thread-safe so your service should re-use one instance.
+ * Creates a producer for usage events.
  *
  * Example:
  * ```ts
- * usageV2 = new UsageV2Producer(..)
- * await usageV2.sendEvents(events)
- * // Non-blocking:
- * // void usageV2.sendEvents(events).catch((e) => console.error(e))
+ * const kafkaProducer = new KafkaProducer({...});
+ * const usageV2 = new UsageV2Producer({ kafkaProducer, source: "storage" });
+ * await usageV2.sendEvents(events);
  * ```
  */
 export class UsageV2Producer {
   private kafkaProducer: KafkaProducer;
   private topic: string;
 
-  constructor(config: {
-    /**
-     * A descriptive name for your service. Example: "storage-server"
-     */
-    producerName: string;
-    /**
-     * A comma-separated list of `host[:port]` Kafka servers.
-     */
-    kafkaServers: string;
-    /**
-     * The product where usage is coming from.
-     */
-    source: UsageV2Source;
-
-    username: string;
-    password: string;
-  }) {
-    this.kafkaProducer = new KafkaProducer({
-      kafkaServers: config.kafkaServers,
-      password: config.password,
-      producerName: config.producerName,
-      username: config.username,
-    });
+  constructor(
+    config:
+      | {
+          /**
+           * Shared KafkaProducer instance.
+           */
+          kafkaProducer: KafkaProducer;
+          /**
+           * The product where usage is coming from.
+           */
+          source: UsageV2Source;
+        }
+      | {
+          /**
+           * A descriptive name for your service. Example: "storage-server"
+           */
+          producerName: string;
+          /**
+           * A comma-separated list of `host[:port]` Kafka servers.
+           */
+          kafkaServers: string;
+          /**
+           * The product where usage is coming from.
+           */
+          source: UsageV2Source;
+          username: string;
+          password: string;
+        },
+  ) {
+    if ("kafkaProducer" in config) {
+      this.kafkaProducer = config.kafkaProducer;
+    } else {
+      this.kafkaProducer = new KafkaProducer({
+        kafkaServers: config.kafkaServers,
+        password: config.password,
+        producerName: config.producerName,
+        username: config.username,
+      });
+    }
     this.topic = getTopicName(config.source);
   }
 
@@ -61,29 +72,21 @@ export class UsageV2Producer {
    * @param events
    */
   async sendEvents(events: UsageV2Event[]): Promise<void> {
-    const parsedEvents = events.map((event) => ({
+    const parsedEvents: UsageV2Event[] = events.map((event) => ({
       ...event,
       // Default to now.
       created_at: event.created_at ?? new Date(),
       // Default to a generated UUID.
       id: event.id ?? randomUUID(),
-      // Remove the "prj_" prefix, if any.
-      project_id: event.project_id?.startsWith(PROJECT_ID_PREFIX)
-        ? event.project_id.slice(PROJECT_ID_PREFIX.length)
+      // Remove the "prj_" prefix.
+      project_id: event.project_id?.startsWith("prj_")
+        ? event.project_id.slice(4)
         : event.project_id,
-      // Remove the "team_" prefix, if any.
-      team_id: event.team_id.startsWith(TEAM_ID_PREFIX)
-        ? event.team_id.slice(TEAM_ID_PREFIX.length)
+      // Remove the "team_" prefix.
+      team_id: event.team_id.startsWith("team_")
+        ? event.team_id.slice(5)
         : event.team_id,
     }));
     await this.kafkaProducer.send(this.topic, parsedEvents);
   }
-
-  /**
-   * Disconnects UsageV2Producer.
-   * Useful when shutting down the service to flush in-flight events.
-   */
-  async disconnect() {
-    await this.kafkaProducer.disconnect();
-  }
 }

packages/service-utils/src/node/webhookProducer.ts

@@ -0,0 +1,86 @@
+import { randomUUID } from "node:crypto";
+import { KafkaProducer } from "./kafka.js";
+
+interface WebhookEvent extends Record<string, unknown> {
+  id?: `evt_${string}`;
+  teamId: string;
+  projectId?: string;
+  createdAt?: Date;
+  /**
+   * This should match your model defined in api-server.
+   */
+  payload: Record<string, unknown>;
+}
+
+/**
+ * Creates a producer for webhook events.
+ *
+ * Example:
+ * ```ts
+ * const kafkaProducer = new KafkaProducer({...});
+ * const webhookProducer = new WebhookEventProducer({ kafkaProducer });
+ * await webhookProducer.sendEvents(events);
+ * ```
+ */
+export class WebhookEventProducer {
+  private kafkaProducer: KafkaProducer;
+
+  constructor(
+    config:
+      | {
+          /**
+           * Shared KafkaProducer instance.
+           */
+          kafkaProducer: KafkaProducer;
+        }
+      | {
+          /**
+           * A descriptive name for your service. Example: "webhook-service"
+           */
+          producerName: string;
+          /**
+           * A comma-separated list of `host[:port]` Kafka servers.
+           */
+          kafkaServers: string;
+          username: string;
+          password: string;
+        },
+  ) {
+    if ("kafkaProducer" in config) {
+      this.kafkaProducer = config.kafkaProducer;
+    } else {
+      this.kafkaProducer = new KafkaProducer({
+        kafkaServers: config.kafkaServers,
+        password: config.password,
+        producerName: config.producerName,
+        username: config.username,
+      });
+    }
+  }
+
+  /**
+   * Emit a webhook event.
+   * This method may throw. To call this non-blocking:
+   * ```ts
+   * void webhookProducer.sendEvents(events).catch((e) => console.error(e))
+   * ```
+   */
+  async sendEvents(topic: string, events: WebhookEvent[]): Promise<void> {
+    const parsedEvents: WebhookEvent[] = events.map((event) => ({
+      // Default to now.
+      created_at: event.createdAt ?? new Date(),
+      // Default to a generated UUID.
+      id: event.id ?? `evt_${randomUUID()}`,
+      payload: event.payload,
+      // Add the "prj_" prefix.
+      projectId: event.projectId?.startsWith("prj_")
+        ? event.projectId
+        : `prj_${event.projectId}`,
+      // Add the "team_" prefix.
+      teamId: event.teamId.startsWith("team_")
+        ? event.teamId
+        : `team_${event.teamId}`,
+    }));
+    await this.kafkaProducer.send(topic, parsedEvents);
+  }
+}