chore: change to per-product usageV2 topic

Author: arcoraven

packages/service-utils/src/cf-worker/usageV2.ts

@@ -1,4 +1,5 @@
-import type { UsageV2Event } from "../core/usageV2.js";
+import type { ServiceName } from "src/core/services.js";
+import { type UsageV2Event, getTopicName } from "../core/usageV2.js";
 
 /**
  * Send events to Kafka.
@@ -19,6 +20,7 @@ export async function sendUsageV2Events(
   events: UsageV2Event[],
   options: {
     environment: "development" | "production";
+    productName: ServiceName;
     serviceKey: string;
   },
 ): Promise<void> {
@@ -27,7 +29,8 @@ export async function sendUsageV2Events(
       ? "https://u.thirdweb.com"
       : "https://u.thirdweb-dev.com";
 
-  const resp = await fetch(`${baseUrl}/usage-v2/raw-events`, {
+  const topic = getTopicName(options.productName);
+  const resp = await fetch(`${baseUrl}/usage-v2/${topic}`, {
     method: "POST",
     headers: {
       "Content-Type": "application/json",

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

@@ -1,3 +1,5 @@
+import type { ServiceName } from "src/node/index.js";
+
 export interface UsageV2Event {
   /**
    * A unique identifier for the event. Defaults to a random UUID.
@@ -8,10 +10,6 @@ export interface UsageV2Event {
    * The event timestamp. Defaults to now().
    */
   created_at?: Date;
-  /**
-   * The source of the event. Example: "storage"
-   */
-  source: string;
   /**
    * The action of the event. Example: "upload"
    */
@@ -49,8 +47,12 @@ export interface UsageV2Event {
    */
   product_version?: string;
   /**
-   * An object of service-specific data. Example: "file_size_bytes"
-   * It is safe to pass any new JSON-serializable data here before updating the usageV2 schema.
+   * An object of arbitrary key-value pairs.
+   * Values can be boolean, number, string, Date, or null.
    */
-  data: Record<string, unknown>;
+  [key: string]: boolean | number | string | Date | null | undefined;
+}
+
+export function getTopicName(productName: ServiceName) {
+  return `usage_v2.raw_${productName}`;
 }

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

@@ -1,9 +1,8 @@
 import { randomUUID } from "node:crypto";
 import { checkServerIdentity } from "node:tls";
 import { Kafka, type Producer } from "kafkajs";
-import type { UsageV2Event } from "../core/usageV2.js";
-
-const TOPIC_USAGE_V2 = "usage_v2.raw_events";
+import type { ServiceName } from "src/core/services.js";
+import { type UsageV2Event, getTopicName } from "../core/usageV2.js";
 
 /**
  * Creates a UsageV2Producer which opens a persistent TCP connection.
@@ -21,6 +20,7 @@ const TOPIC_USAGE_V2 = "usage_v2.raw_events";
 export class UsageV2Producer {
   private kafka: Kafka;
   private producer: Producer | null = null;
+  private productName: ServiceName;
 
   constructor(config: {
     /**
@@ -31,6 +31,10 @@ export class UsageV2Producer {
      * The environment the service is running in.
      */
     environment: "development" | "production";
+    /**
+     * The product "source" where usage is coming from.
+     */
+    productName: ServiceName;
 
     username: string;
     password: string;
@@ -52,6 +56,8 @@ export class UsageV2Producer {
         password: config.password,
       },
     });
+
+    this.productName = config.productName;
   }
 
   /**
@@ -82,27 +88,18 @@ export class UsageV2Producer {
 
     const parsedEvents = events.map((event) => {
       return {
+        ...event,
         id: event.id ?? randomUUID(),
         created_at: event.created_at ?? new Date(),
-        source: event.source,
-        action: event.action,
         // Remove the "team_" prefix, if any.
         team_id: event.team_id.startsWith("team_")
           ? event.team_id.slice(5)
           : event.team_id,
-        project_id: event.project_id,
-        sdk_name: event.sdk_name,
-        sdk_platform: event.sdk_platform,
-        sdk_version: event.sdk_version,
-        sdk_os: event.sdk_os,
-        product_name: event.product_name,
-        product_version: event.product_version,
-        data: JSON.stringify(event.data),
       };
     });
 
     await this.producer.send({
-      topic: TOPIC_USAGE_V2,
+      topic: getTopicName(this.productName),
       messages: parsedEvents.map((event) => ({
         value: JSON.stringify(event),
       })),