[service-utils] fix: use lz4js for usageV2 (#6130)

Author: arcoraven

packages/service-utils/package.json

@@ -47,11 +47,12 @@
   "dependencies": {
     "aws4fetch": "1.0.20",
     "kafkajs": "2.2.4",
-    "kafkajs-lz4": "2.0.0-beta.0",
+    "lz4js": "^0.2.0",
     "zod": "3.24.1"
   },
   "devDependencies": {
     "@cloudflare/workers-types": "4.20250129.0",
+    "@types/lz4js": "^0.2.1",
     "@types/node": "22.10.10",
     "typescript": "5.7.3",
     "vitest": "3.0.4"

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

@@ -6,7 +6,7 @@ import {
   type Producer,
   type ProducerConfig,
 } from "kafkajs";
-import LZ4Codec from "kafkajs-lz4";
+import { compress, decompress } from "lz4js";
 import type { ServiceName } from "../core/services.js";
 import { type UsageV2Event, getTopicName } from "../core/usageV2.js";
 
@@ -91,7 +91,20 @@ export class UsageV2Producer {
    */
   async init(configOverrides?: ProducerConfig) {
     if (this.compression === CompressionTypes.LZ4) {
-      CompressionCodecs[CompressionTypes.LZ4] = new LZ4Codec().codec;
+      CompressionCodecs[CompressionTypes.LZ4] = () => ({
+        // biome-ignore lint/style/noRestrictedGlobals: kafkajs expects a Buffer
+        compress: (encoder: { buffer: Buffer }) => {
+          const compressed = compress(encoder.buffer);
+          // biome-ignore lint/style/noRestrictedGlobals: kafkajs expects a Buffer
+          return Buffer.from(compressed);
+        },
+        // biome-ignore lint/style/noRestrictedGlobals: kafkajs expects a Buffer
+        decompress: (buffer: Buffer) => {
+          const decompressed = decompress(buffer);
+          // biome-ignore lint/style/noRestrictedGlobals: kafkajs expects a Buffer
+          return Buffer.from(decompressed);
+        },
+      });
     }
 
     this.producer = this.kafka.producer({