fix/cleaning callback maps in nats and workers service[4626] (#4677)

ihar-tsykala · Ihar · envision-ci-agent · web-flow · commit 806087bf1a9c · 2025-02-25T18:40:25.000+03:00
* fix/responseCallbacksMap by balance subscribtion[4626]

* refactor code

* refactor code

* feat/increase hashgraph/sdk version[4626]

* refactor code

* [skip ci] Add swagger.yaml

* fix: hashgraph/sdk issue and refactor[4626]

* fix: callback maps in workers, analitics, guardian, policy[4626]

* fix: lint errors

---------

Co-authored-by: Ihar &lt;igor.tsykala@envisionblockchain.com&gt;
Co-authored-by: envision-ci-agent &lt;envision-ci-agent@users.noreply.github.com&gt;
diff --git a/common/src/helpers/workers.ts b/common/src/helpers/workers.ts
@@ -121,8 +121,9 @@ export class Workers extends NatsService {
      * @param task
      * @param priority
      * @param userId
+     * @param registerCallback
      */
-    public addNonRetryableTask(task: ITask, priority: number, userId?: string | null): Promise<any> {
+    public addNonRetryableTask(task: ITask, priority: number, userId?: string | null, registerCallback: boolean = true): Promise<any> {
         if (!task.data.network) {
             task.data.network = Environment.network;
         }
@@ -138,7 +139,7 @@ export class Workers extends NatsService {
         if (!task.data.localNodeProtocol) {
             task.data.localNodeProtocol = Environment.localNodeProtocol;
         }
-        return this.addTask(task, priority, false, 0, true, userId);
+        return this.addTask(task, priority, false, 0, registerCallback, userId);
     }
 
     /**
@@ -147,8 +148,9 @@ export class Workers extends NatsService {
      * @param priority
      * @param attempts
      * @param userId
+     * @param registerCallback
      */
-    public addRetryableTask(task: ITask, priority: number, attempts: number = 0, userId: string = null): Promise<any> {
+    public addRetryableTask(task: ITask, priority: number, attempts: number = 0, userId: string = null, registerCallback: boolean = true): Promise<any> {
         if (!task.data.network) {
             task.data.network = Environment.network;
         }
@@ -164,7 +166,7 @@ export class Workers extends NatsService {
         if (!task.data.localNodeProtocol) {
             task.data.localNodeProtocol = Environment.localNodeProtocol;
         }
-        return this.addTask(task, priority, true, attempts, true, userId);
+        return this.addTask(task, priority, true, attempts, registerCallback, userId);
     }
 
     /**
@@ -189,6 +191,7 @@ export class Workers extends NatsService {
             if (this.tasksCallbacks.has(data.id)) {
                 const activeTask = this.tasksCallbacks.get(data.id);
                 activeTask.callback(data.data, data.error, data.isTimeoutError);
+                this.tasksCallbacks.delete(data.id)
             }
 
         })
diff --git a/common/src/mq/nats-service.ts b/common/src/mq/nats-service.ts
@@ -1,12 +1,13 @@
-import { NatsConnection, headers, Subscription } from 'nats';
-import { GenerateUUIDv4 } from '@guardian/interfaces';
-import { ZipCodec } from './zip-codec.js';
-import { IMessageResponse } from '../models/index.js';
+import {NatsConnection, headers, Subscription} from 'nats';
+import {GenerateUUIDv4} from '@guardian/interfaces';
+import {ZipCodec} from './zip-codec.js';
+import {IMessageResponse} from '../models/index.js';
 
 type CallbackFunction = (body: any, error?: string, code?: number) => void;
 
 class MessageError extends Error {
     public code: number;
+
     constructor(message: any, code?: number) {
         super(message);
         this.code = code;
@@ -107,6 +108,7 @@ export abstract class NatsService {
      */
     public subscribe(subject: string, cb: Function): Subscription {
         const sub = this.connection.subscribe(subject);
+
         const fn = async (_sub: Subscription) => {
             for await (const m of _sub) {
                 try {
@@ -124,19 +126,24 @@ export abstract class NatsService {
      * Send message
      * @param subject
      * @param data
+     * @param isResponseCallback
      */
-    public sendMessage<T>(subject: string, data?: unknown): Promise<T> {
+    public sendMessage<T>(subject: string, data?: unknown, isResponseCallback: boolean = true): Promise<T> {
         const messageId = GenerateUUIDv4();
         return new Promise(async (resolve, reject) => {
             const head = headers();
             head.append('messageId', messageId);
-            this.responseCallbacksMap.set(messageId, (body: T, error?: string, code?: number) => {
-                if (error) {
-                    reject(new MessageError(error, code));
-                } else {
-                    resolve(body);
-                }
-            })
+            if (isResponseCallback) {
+                this.responseCallbacksMap.set(messageId, (body: T, error?: string, code?: number) => {
+                    if (error) {
+                        reject(new MessageError(error, code));
+                    } else {
+                        resolve(body);
+                    }
+                })
+            } else {
+                resolve(null);
+            }
 
             this.connection.publish(subject, await this.codec.encode(data), {
                 reply: this.replySubject,
@@ -155,7 +162,9 @@ export abstract class NatsService {
         return Promise.race([
             this.sendMessage<T>(subject, data),
             new Promise<T>((_, reject) => {
-                setTimeout(() => { reject(new Error('Timeout exceed')) }, timeout)
+                setTimeout(() => {
+                    reject(new Error('Timeout exceed'))
+                }, timeout)
             })
         ])
     }
@@ -206,7 +215,7 @@ export abstract class NatsService {
                     }
                     // head.append('rawMessage', isRaw);
                     if (!noRespond) {
-                        msg.respond(await this.codec.encode(await cb(await this.codec.decode(msg.data), msg.headers)), { headers: head });
+                        msg.respond(await this.codec.encode(await cb(await this.codec.decode(msg.data), msg.headers)), {headers: head});
                     } else {
                         cb(await this.codec.decode(msg.data), msg.headers);
                     }
diff --git a/guardian-service/src/policy-engine/policy-engine.service.ts b/guardian-service/src/policy-engine/policy-engine.service.ts
@@ -200,7 +200,7 @@ export class PolicyEngineService {
 
         PolicyComponentsUtils.ExternalEventFn = async (...args: any[]) => {
             try {
-                this.channel.sendMessage(ExternalMessageEvents.BLOCK_EVENTS, args);
+                this.channel.sendMessage(ExternalMessageEvents.BLOCK_EVENTS, args, false);
             } catch (error) {
                 console.error(error);
             }
diff --git a/policy-service/src/policy-engine/policy-components-utils.ts b/policy-service/src/policy-engine/policy-components-utils.ts
@@ -108,7 +108,7 @@ export type PolicyActionMap = Map<string, Map<PolicyInputEventType, EventCallbac
  */
 export function updateBlockEvent(blocks: string[], user: PolicyUser): void {
     const type = 'update';
-    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [blocks, user.toJson()] });
+    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [blocks, user.toJson()] }, false);
 }
 
 /**
@@ -118,7 +118,7 @@ export function updateBlockEvent(blocks: string[], user: PolicyUser): void {
  */
 export function errorBlockEvent(blockType: string, message: any, user: PolicyUser): void {
     const type = 'error';
-    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [blockType, message, user.toJson()] });
+    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [blockType, message, user.toJson()] }, false);
 }
 
 /**
@@ -128,7 +128,7 @@ export function errorBlockEvent(blockType: string, message: any, user: PolicyUse
  */
 export function infoBlockEvent(user: PolicyUser, policy: Policy): void {
     const type = 'update-user';
-    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [user.toJson(), policy] });
+    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [user.toJson(), policy] }, false);
 }
 
 /**
@@ -138,7 +138,7 @@ export function infoBlockEvent(user: PolicyUser, policy: Policy): void {
  */
 export function externalBlockEvent(event: ExternalEvent<any>): void {
     const type = 'external';
-    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [event] });
+    new BlockTreeGenerator().sendMessage(PolicyEvents.BLOCK_UPDATE_BROADCAST, { type, data: [event] }, false);
 }
 
 /**
diff --git a/worker-service/package.json b/worker-service/package.json
@@ -4,7 +4,7 @@
         "@filebase/client": "^0.0.5",
         "@guardian/common": "^3.1.0",
         "@guardian/interfaces": "^3.1.0",
-        "@hashgraph/sdk": "2.52.0",
+        "@hashgraph/sdk": "2.59.0",
         "@nestjs/common": "^9.4.1",
         "@nestjs/core": "^9.4.1",
         "@nestjs/microservices": "^9.4.1",
diff --git a/worker-service/src/api/helpers/hedera-sdk-helper.ts b/worker-service/src/api/helpers/hedera-sdk-helper.ts
@@ -949,9 +949,13 @@ export class HederaSDKHelper {
                     const signedMessage = tx.signedMessages[0];
                     if (signedMessage) {
                         const pubKey = PublicKey.fromStringED25519(signedMessage.publicKey);
-                        const signature = Buffer.from(signedMessage.signature.fullSig, 'hex');
+                        const signatureBuffer = Buffer.from(signedMessage.signature.fullSig, 'hex');
+                        const signature = new Uint8Array(signatureBuffer);
+
                         try {
-                            messageTransaction.addSignature(pubKey, signature);
+                            await messageTransaction.signWith(pubKey, async () => {
+                                return signature;
+                            });
                         } catch (error) {
                             throw new Error(error);
                         }
diff --git a/worker-service/src/api/worker.ts b/worker-service/src/api/worker.ts
@@ -1,4 +1,12 @@
-import { MessageBrokerChannel, MessageResponse, NatsService, NotificationHelper, PinoLogger, SecretManager, Users } from '@guardian/common';
+import {
+    MessageBrokerChannel,
+    MessageResponse,
+    NatsService,
+    NotificationHelper,
+    PinoLogger,
+    SecretManager,
+    Users
+} from '@guardian/common';
 import { ExternalMessageEvents, GenerateUUIDv4, ISignOptions, ITask, ITaskResult, WorkerEvents, WorkerTaskType } from '@guardian/interfaces';
 import { HederaSDKHelper, NetworkOptions } from './helpers/hedera-sdk-helper.js';
 import { IpfsClientClass } from './ipfs-client-class.js';
@@ -138,7 +146,6 @@ export class Worker extends NatsService {
     public async init(): Promise<void> {
         await super.init();
         this.channel = new MessageBrokerChannel(this.connection, 'worker');
-
         try {
             await this.ipfsClient.createClient()
         } catch (e) {
@@ -226,7 +233,7 @@ export class Worker extends NatsService {
                     balance,
                     unit: 'Hbar',
                     operatorAccountId
-                });
+                }, false);
             } catch (error) {
                 throw new Error(`Worker (${['api-gateway', 'update-user-balance'].join('.')}) send: ` + error);
             }
@@ -542,7 +549,6 @@ export class Worker extends NatsService {
                         operatorKey,
                         adminKey,
                     } = task.data;
-
                     client = new HederaSDKHelper(operatorId, operatorKey, null, networkOptions);
                     result.data = await client.deleteToken(
                         TokenId.fromString(tokenId),

Original file line number	Diff line number	Diff line change
`@@ -200,7 +200,7 @@ export class PolicyEngineService {`
`200`	`200`
`201`	`201`	`PolicyComponentsUtils.ExternalEventFn = async (...args: any[]) => {`
`202`	`202`	`try {`
`203`		`- this.channel.sendMessage(ExternalMessageEvents.BLOCK_EVENTS, args);`
	`203`	`+ this.channel.sendMessage(ExternalMessageEvents.BLOCK_EVENTS, args, false);`
`204`	`204`	`} catch (error) {`
`205`	`205`	`console.error(error);`
`206`	`206`	`}`