add optional compression

Author: arcoraven

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

@@ -11,8 +11,6 @@ import LZ4Codec from "kafkajs-lz4";
 import type { ServiceName } from "../core/services.js";
 import { type UsageV2Event, getTopicName } from "../core/usageV2.js";
 
-CompressionCodecs[CompressionTypes.LZ4] = new LZ4Codec().codec;
-
 /**
  * Creates a UsageV2Producer which opens a persistent TCP connection.
  * This class is thread-safe so your service should re-use one instance.
@@ -30,6 +28,7 @@ export class UsageV2Producer {
   private kafka: Kafka;
   private producer: Producer | null = null;
   private topic: string;
+  private compression: CompressionTypes;
 
   constructor(config: {
     /**
@@ -44,14 +43,27 @@ export class UsageV2Producer {
      * The product "source" where usage is coming from.
      */
     productName: ServiceName;
+    /**
+     * The compression algorithm to use.
+     */
+    compression?: CompressionTypes;
 
     username: string;
     password: string;
   }) {
+    const {
+      producerName,
+      environment,
+      productName,
+      compression = CompressionTypes.LZ4,
+      username,
+      password,
+    } = config;
+
     this.kafka = new Kafka({
-      clientId: `${config.producerName}-${config.environment}`,
+      clientId: `${producerName}-${environment}`,
       brokers:
-        config.environment === "production"
+        environment === "production"
           ? ["warpstream.thirdweb.xyz:9092"]
           : ["warpstream-dev.thirdweb.xyz:9092"],
       ssl: {
@@ -61,12 +73,16 @@ export class UsageV2Producer {
       },
       sasl: {
         mechanism: "plain",
-        username: config.username,
-        password: config.password,
+        username,
+        password,
       },
     });
 
-    this.topic = getTopicName(config.productName);
+    this.topic = getTopicName(productName);
+    this.compression = compression;
+    if (compression === CompressionTypes.LZ4) {
+      CompressionCodecs[CompressionTypes.LZ4] = new LZ4Codec().codec;
+    }
   }
 
   /**
@@ -99,7 +115,6 @@ export class UsageV2Producer {
     configOverrides?: {
       acks?: number;
       timeout?: number;
-      compression?: CompressionTypes;
     },
   ): Promise<void> {
     if (!this.producer) {
@@ -125,7 +140,7 @@ export class UsageV2Producer {
       })),
       acks: -1, // All brokers must acknowledge
       timeout: 10_000, // 10 seconds
-      compression: CompressionTypes.LZ4,
+      compression: this.compression,
       ...configOverrides,
     });
   }