Merge pull request #12 from ARyaskov/dev

v2.3.2

Author: ARyaskov

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@riaskov/nevo-messaging",
-  "version": "2.3.1",
+  "version": "2.3.2",
   "description": "Microservices messaging framework for NestJS with NATS/Kafka/SocketIO/HTTP transport, MessagePack, retry/circuit-breaker, OTel, schema validation, DLQ, idempotency, saga, outbox, metrics, health, DevTools",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
@@ -50,7 +50,6 @@
   "license": "MIT",
   "peerDependencies": {
     "@msgpack/msgpack": "^3.1.3",
-    "@napi-rs/zstd": "^1.0.0",
     "@nats-io/jetstream": "^3.0.0",
     "@nats-io/nats-core": "^3.0.0",
     "@nats-io/transport-node": "^3.0.0",
@@ -145,9 +144,6 @@
     },
     "cacheable-lookup": {
       "optional": true
-    },
-    "@napi-rs/zstd": {
-      "optional": true
     }
   },
   "devDependencies": {

pnpm-lock.yaml

@@ -2,10 +2,13 @@
 import { promises as fs } from "node:fs"
 import * as path from "node:path"
 import { randomUUID } from "node:crypto"
+import { createRequire } from "node:module"
 import { generateContractModule } from "./generate"
 import { NEVO_CONTRACT_METHOD } from "../common/contract"
 import type { ServiceContract } from "../common/contract"
 
+const nodeRequire = createRequire(__filename)
+
 interface CliOptions {
   transport: "http" | "nats"
   service: string
@@ -111,7 +114,7 @@ async function fetchContractNats(opts: CliOptions): Promise<ServiceContract> {
   let connect: any
   try {
     // nats.js v3 split the legacy `nats` package: `connect` lives in transport-node.
-    connect = require("@nats-io/transport-node").connect
+    connect = nodeRequire("@nats-io/transport-node").connect
   } catch {
     throw new Error("Missing optional dependency '@nats-io/transport-node'. Install with: npm i @nats-io/transport-node")
   }

src/cli/contract.ts

@@ -1,6 +1,5 @@
 import { randomUUID } from "node:crypto"
 import {
-  MessagePayload,
   MessageRequest,
   MicroserviceConfig,
   TransportClientOptions,

src/common/base.client.ts

@@ -280,7 +280,7 @@ export abstract class BaseMessageController {
     const nowMs = Date.now()
     const startMs = nowMs
     const { method, uuid, params, meta } = this.extractMessageData(data)
-    let success = true
+    const success = true
 
     // Establish a chain context for the duration of this handler so any
     // outbound calls picks up the same chain id via AsyncLocalStorage.

src/common/base.controller.ts

@@ -1,9 +1,12 @@
+import { createRequire } from "node:module"
 import { CircuitOpenError, MessagingError } from "./errors"
 import { ErrorCode } from "./error-code"
 import type { CircuitBreakerOptions } from "./types"
 import { getDevToolsBus, DevToolsBus } from "./devtools"
 import type { DevToolsRegistry } from "./devtools-registry"
 
+const nodeRequire = createRequire(__filename)
+
 export type CircuitState = "closed" | "open" | "half-open"
 
 export interface ResolvedCircuitOptions {
@@ -41,7 +44,7 @@ export class CircuitBreakerRegistry {
     if (deps?.registry) {
       this.registry = deps.registry
     } else {
-      const { getDevToolsRegistry } = require("./devtools-registry") as typeof import("./devtools-registry")
+      const { getDevToolsRegistry } = nodeRequire("./devtools-registry") as typeof import("./devtools-registry")
       this.registry = getDevToolsRegistry()
     }
   }

src/common/circuit-breaker.ts

@@ -1,7 +1,10 @@
+import { createRequire } from "node:module"
 import { bigIntReplacer, makeBigIntReviver } from "./bigint.utils"
 import { MessagingError } from "./errors"
 import { ErrorCode } from "./error-code"
 
+const nodeRequire = createRequire(__filename)
+
 const legacyReviver = makeBigIntReviver({ acceptLegacy: true })
 
 export type CodecName = "msgpack" | "json" | "json-fast" | string
@@ -109,7 +112,7 @@ let sharedMsgpack: { encoder: { encode(v: unknown): Uint8Array }; decoder: { dec
 function getOrCreateSharedMsgpack(): { encoder: { encode(v: unknown): Uint8Array }; decoder: { decode(b: Uint8Array): unknown } } {
   if (sharedMsgpack) return sharedMsgpack
   try {
-    const mp = require("@msgpack/msgpack") as typeof import("@msgpack/msgpack")
+    const mp = nodeRequire("@msgpack/msgpack") as typeof import("@msgpack/msgpack")
     const EncoderCtor = (mp as unknown as { Encoder?: new (opts: object) => { encode(v: unknown): Uint8Array } }).Encoder
     const DecoderCtor = (mp as unknown as { Decoder?: new (opts: object) => { decode(b: Uint8Array): unknown } }).Decoder
     const opts = { useBigInt64: true }

src/common/codec.ts

@@ -1,7 +1,6 @@
 import { Worker } from "node:worker_threads"
-import * as path from "node:path"
 import * as os from "node:os"
-import { getDefaultLogger, NevoLogger } from "./logger"
+import { NevoLogger } from "./logger"
 
 const WORKER_INLINE_SOURCE = `
 const { parentPort } = require("node:worker_threads")
@@ -84,12 +83,10 @@ export interface CompressionWorkerOptions {
   logger?: NevoLogger
 }
 
-let logger: NevoLogger | null = null
 let cfg: CompressionWorkerOptions = {}
 
 export function configureCompressionWorker(opts: CompressionWorkerOptions): void {
   cfg = opts
-  logger = opts.logger ?? getDefaultLogger().child({ component: "compression-worker" })
   if (opts.enabled) getPool(opts.poolSize ?? Math.max(1, Math.min(4, os.cpus().length - 1)))
 }
 

src/common/compression-worker.ts

@@ -6,7 +6,11 @@ import {
   constants as zlibConstants
 } from "node:zlib"
 import { gzip, gunzip, deflate, inflate } from "node:zlib"
+import * as nodeZlib from "node:zlib"
+import { createRequire } from "node:module"
 import { promisify } from "node:util"
+
+const nodeRequire = createRequire(__filename)
 import type { CompressionOptions } from "./types"
 import { PayloadTooLargeError } from "./errors"
 
@@ -42,22 +46,7 @@ let zstdAsync: { compress: (b: Uint8Array, opts?: object) => Promise<Buffer>; de
 function tryLoadZstd(): boolean {
   if (zstdSync && zstdAsync) return true
   try {
-    const napi: any = require("@napi-rs/zstd")
-    if (napi && typeof napi.compressSync === "function") {
-      zstdSync = {
-        compress: (b) => Buffer.from(napi.compressSync(Buffer.from(b))),
-        decompress: (b) => Buffer.from(napi.decompressSync(Buffer.from(b)))
-      }
-      zstdAsync = {
-        compress: async (b) => Buffer.from(await napi.compress(Buffer.from(b))),
-        decompress: async (b) => Buffer.from(await napi.decompress(Buffer.from(b)))
-      }
-      return true
-    }
-  } catch {}
-  try {
-    const zlib = require("node:zlib") as typeof import("node:zlib")
-    const z: any = zlib
+    const z: any = nodeZlib
     if (typeof z.zstdCompressSync === "function") {
       zstdSync = {
         compress: (b, opts) => z.zstdCompressSync(b, opts),
@@ -91,7 +80,7 @@ export async function maybeCompressAsync(buf: Uint8Array, opts: ResolvedCompress
     return { data: buf, encoding: "identity" }
   }
   try {
-    const { isCompressionWorkerEnabled, compressionWorkerThreshold, workerCompress } = require("./compression-worker") as typeof import("./compression-worker")
+    const { isCompressionWorkerEnabled, compressionWorkerThreshold, workerCompress } = nodeRequire("./compression-worker") as typeof import("./compression-worker")
     if (isCompressionWorkerEnabled() && buf.byteLength >= compressionWorkerThreshold()) {
       const algo: "gzip" | "deflate" | "zstd" = opts.algorithm === "zstd" && tryLoadZstd() ? "zstd" : opts.algorithm === "deflate" ? "deflate" : "gzip"
       const data = await workerCompress(buf, algo, opts.level)
@@ -219,7 +208,7 @@ export async function maybeDecompressAsync(buf: Uint8Array, encoding?: string, m
 async function tryWorkerDecompress(buf: Uint8Array, encoding: "gzip" | "deflate", maxOutputBytes?: number): Promise<Uint8Array | null> {
   let mod: typeof import("./compression-worker")
   try {
-    mod = require("./compression-worker") as typeof import("./compression-worker")
+    mod = nodeRequire("./compression-worker") as typeof import("./compression-worker")
   } catch {
     return null
   }

src/common/compression.ts

@@ -1,4 +1,3 @@
-import { setTimeout as sleep } from "node:timers/promises"
 import { NEVO_CONTRACT_METHOD, ServiceContract, SchemaDescriptor } from "./contract"
 
 export const NEVO_CONTRACT_CHANGED_METHOD = "nevo.contract.changed"