703 Handle Workstep Execution Failures

examples/bri-3/src/bri/communication/models/bpiMessageType.enum.ts

@@ -1,4 +1,5 @@
 export enum BpiMessageType {
   Info,
   Transaction,
+  Err,
 }

examples/bri-3/src/bri/vsm/capabilites/executeVsmCycle/executeVsmCycleCommand.handler.ts

@@ -1,10 +1,10 @@
-import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
-import { LoggingService } from '../../../../shared/logging/logging.service';
+import { CommandHandler, ICommandHandler, EventBus } from '@nestjs/cqrs';
 import { TransactionStorageAgent } from '../../../transactions/agents/transactionStorage.agent';
 import { TransactionAgent } from '../../../transactions/agents/transactions.agent';
 import { TransactionStatus } from '../../../transactions/models/transactionStatus.enum';
 import { WorkstepStorageAgent } from '../../../workgroup/worksteps/agents/workstepsStorage.agent';
 import { ExecuteVsmCycleCommand } from './executeVsmCycle.command';
+import { WorkstepExecutionFailuresEvent } from '../handleWorkstepFailuresEvents/workstepExecutionFailures.event';
 
 @CommandHandler(ExecuteVsmCycleCommand)
 export class ExecuteVsmCycleCommandHandler
@@ -14,10 +14,10 @@
     private agent: TransactionAgent,
     private workstepStorageAgent: WorkstepStorageAgent,
     private txStorageAgent: TransactionStorageAgent,
-    private readonly logger: LoggingService,
+    private eventBus: EventBus,
   ) {}
 
   async execute(command: ExecuteVsmCycleCommand) {
     const executionCandidates =
       await this.txStorageAgent.getTopNTransactionsByStatus(
         Number(process.env.VSM_CYCLE_TX_BATCH_SIZE),
@@ -29,6 +29,9 @@
       await this.txStorageAgent.updateTransactionStatus(tx);
 
       if (!this.agent.validateTransactionForExecution(tx)) {
+        this.eventBus.publish(
+          new WorkstepExecutionFailuresEvent(tx, 'Validation Error'),
+        );
         tx.updateStatusToInvalid();
         await this.txStorageAgent.updateTransactionStatus(tx);
         return;
@@ -39,15 +42,13 @@
       );
 
       try {
         const txResult = await this.agent.executeTransaction(tx, workstep!);
         // TODO: When do we update the nonce on the BpiAccount? // Whenever a transaction is initiated
         // TODO: #702 Update relevant Bpi Account state with txResult sucess
         tx.updateStatusToExecuted();
         this.txStorageAgent.updateTransactionStatus(tx);
       } catch (error) {
-        this.logger.logError(
-          `Error executing transaction with id ${tx.id}: ${error}`,
-        );
+        this.eventBus.publish(new WorkstepExecutionFailuresEvent(tx, error));
         tx.updateStatusToAborted();
         this.txStorageAgent.updateTransactionStatus(tx);
         return;

examples/bri-3/src/bri/vsm/capabilites/handleWorkstepFailuresEvents/workstepExecutionFailures.event.ts

@@ -0,0 +1,5 @@
+import { Transaction } from 'src/bri/transactions/models/transaction';
+
+export class WorkstepExecutionFailuresEvent {
+  constructor(public readonly tx: Transaction, public readonly err: string) {}
+}

examples/bri-3/src/bri/vsm/capabilites/handleWorkstepFailuresEvents/workstepExecutionFailures.handler.ts

@@ -0,0 +1,42 @@
+import { IEventHandler } from '@nestjs/cqrs';
+import { EventsHandler } from '@nestjs/cqrs/dist/decorators/events-handler.decorator';
+import { WorkstepExecutionFailuresEvent } from './workstepExecutionFailures.event';
+import { LoggingService } from '../../../../shared/logging/logging.service';
+import { BpiMessage } from '../../../communication/models/bpiMessage';
+import { BpiMessageType } from '../../../communication/models/bpiMessageType.enum';
+import { MessagingAgent } from '../../../communication/agents/messaging.agent';
+
+@EventsHandler(WorkstepExecutionFailuresEvent)
+export class WorkstepExecutionFailuresHandler
+  implements IEventHandler<WorkstepExecutionFailuresEvent>
+{
+  constructor(
+    private readonly logger: LoggingService,
+    private readonly messagingAgent: MessagingAgent,
+  ) {}
+
+  handle(event: WorkstepExecutionFailuresEvent) {
+    const message = `Failed execution of transaction with id ${event.tx.id}. Error: ${event.err}`;
+
+    this.logger.logError(message);
+
+    const errPayload = {
+      errorId: 'xxx',
+      errorMessage: message,
+    };
+
+    const errorBpiMessage = new BpiMessage(
+      event.tx.id,
+      event.tx.fromBpiSubjectAccountId,
+      event.tx.toBpiSubjectAccountId,
+      JSON.stringify(errPayload),
+      event.tx.signature,
+      BpiMessageType.Err,
+    );
+
+    this.messagingAgent.publishMessage(
+      event.tx.fromBpiSubjectAccount.ownerBpiSubject.publicKey,
+      JSON.stringify(errorBpiMessage),
+    );
+  }
+}

examples/bri-3/src/bri/vsm/vsm.module.ts

@@ -6,8 +6,14 @@ import { TransactionModule } from '../transactions/transactions.module';
 import { WorkstepModule } from '../workgroup/worksteps/worksteps.module';
 import { VsmTasksSchedulerAgent } from './agents/vsmTaskScheduler.agent';
 import { ExecuteVsmCycleCommandHandler } from './capabilites/executeVsmCycle/executeVsmCycleCommand.handler';
+import { MessagingAgent } from '../communication/agents/messaging.agent';
+import { WorkstepExecutionFailuresHandler } from './capabilites/handleWorkstepFailuresEvents/workstepExecutionFailures.handler';
+import { NatsMessagingClient } from '../communication/messagingClients/natsMessagingClient';
 
-export const CommandHandlers = [ExecuteVsmCycleCommandHandler];
+export const CommandHandlers = [
+  ExecuteVsmCycleCommandHandler,
+  WorkstepExecutionFailuresHandler,
+];
 
 export const QueryHandlers = [];
 
@@ -19,6 +25,15 @@ export const QueryHandlers = [];
     LoggingModule,
     WorkstepModule,
   ],
-  providers: [VsmTasksSchedulerAgent, ...CommandHandlers, ...QueryHandlers],
+  providers: [
+    VsmTasksSchedulerAgent,
+    ...CommandHandlers,
+    ...QueryHandlers,
+    MessagingAgent,
+    {
+      provide: 'IMessagingClient',
+      useClass: NatsMessagingClient,
+    },
+  ],
 })
 export class VsmModule {}