refactor: move pub/sub and stream metrics recording to command wrapper

Author: PavelPashov

packages/client/lib/client/index.ts

@@ -1027,31 +1027,6 @@ export default class RedisClient<
     } else {
       const reply = await fn();
 
-      // Record pub/sub messages for PUBLISH and SPUBLISH commands (after successful send)
-      const cmdName = parser.redisArgs[0]?.toString().toUpperCase();
-      if (cmdName === 'PUBLISH' || cmdName === 'SPUBLISH') {
-        const channel = parser.redisArgs[1]?.toString();
-        const sharded = cmdName === 'SPUBLISH';
-        OTelMetrics.instance.pubSubMetrics.recordPubSubMessage(
-          'out',
-          channel,
-          sharded,
-          this._getClientOTelAttributes()
-        );
-      }
-
-      // Record stream messages for XADD command (after successful send)
-      if (cmdName === 'XADD') {
-        const stream = parser.redisArgs[1]?.toString();
-        if (stream) {
-          OTelMetrics.instance.streamMetrics.recordStreamProduced(
-            stream,
-            1,
-            this._getClientOTelAttributes()
-          );
-        }
-      }
-
       if (transformReply) {
         return transformReply(reply, parser.preserve, commandOptions?.typeMapping);
       }

packages/client/lib/commander.ts

@@ -1,4 +1,5 @@
 import { Command, CommanderConfig, RedisArgument, RedisCommands, RedisFunction, RedisFunctions, RedisModules, RedisScript, RedisScripts, RespVersions, TransformReply } from './RESP/types';
+import { OTelMetrics } from './opentelemetry';
 
 interface AttachConfigOptions<
   M extends RedisModules,
@@ -41,7 +42,7 @@ export function attachConfig<
     if (config?.RESP == 3 && command.unstableResp3 && !config.unstableResp3) {
       Class.prototype[name] = throwResp3SearchModuleUnstableError;
     } else {
-      Class.prototype[name] = createCommand(command, RESP);
+      Class.prototype[name] = OTelMetrics.wrapWithMetrics(name, createCommand(command, RESP)) as ReturnType<typeof createCommand>;
     }
   }