feat: implement agent events for basic messages

apps/agent/deployment/Dockerfile

@@ -12,5 +12,6 @@ RUN yarn install
 
 EXPOSE 8080
 EXPOSE 8001
+EXPOSE 6001
 
 CMD ["node", "main.js"]

apps/attestation-manager/src/app/app.controller.ts

@@ -2,12 +2,14 @@ import { Body, Controller, Logger } from "@nestjs/common";
 
 import { MessagePattern, RpcException } from "@nestjs/microservices";
 import {
+  BasicMessageEvent,
   CloudEventDto,
   CreateCredentialDefinitionRequsetDto,
   CreateSchemaRequestDto,
   CredentialEvent,
   GatewayAcceptedResponseDto,
   IssueCredentialRequestDto,
+  MakeBasicMessageRequestDto,
   makeEvent,
   SchemaEvent,
 } from "@ocm-engine/dtos";
@@ -27,7 +29,7 @@ export class AppController {
       type: SchemaEvent;
       source: string;
     },
-  ): Promise<{ id: string }> {
+  ): Promise<GatewayAcceptedResponseDto> {
     this.logger.debug(JSON.stringify(payload, null, 2));
 
     try {
@@ -63,7 +65,40 @@ export class AppController {
       type: CredentialEvent;
       source: string;
     },
-  ): Promise<{ id: string }> {
+  ): Promise<GatewayAcceptedResponseDto> {
+    this.logger.debug(JSON.stringify(payload, null, 2));
+
+    try {
+      const event = makeEvent(payload);
+      this.logger.debug(JSON.stringify(event, null, 2));
+      await this.producerService.publish<typeof payload.data>(
+        payload.type,
+        event as CloudEventDto<typeof payload.data>,
+      );
+
+      const response = new GatewayAcceptedResponseDto();
+      response.id = event.id;
+
+      return response;
+    } catch (e) {
+      this.logger.debug(JSON.stringify(e, null, 2));
+      if (e instanceof Error) {
+        throw new RpcException(e.message);
+      }
+
+      throw new RpcException("Internal server error");
+    }
+  }
+
+  @MessagePattern("messages")
+  async sendMeesage(
+    @Body()
+    payload: {
+      data: MakeBasicMessageRequestDto;
+      type: BasicMessageEvent;
+      source: string;
+    },
+  ): Promise<GatewayAcceptedResponseDto> {
     this.logger.debug(JSON.stringify(payload, null, 2));
 
     try {

apps/connection-manager/src/app/app.controller.ts

@@ -25,7 +25,7 @@ export class AppController {
       type: ConnectionEvent;
       source: string;
     },
-  ): Promise<{ id: string }> {
+  ): Promise<GatewayAcceptedResponseDto> {
     this.logger.debug(JSON.stringify(payload, null, 2));
 
     try {

apps/gateway/src/app/managers/attestation.controller.ts

@@ -21,6 +21,8 @@ import {
   GatewayAcceptedResponseDto,
   GetSchemaRequestDto,
   IssueCredentialRequestDto,
+  MakeBasicMessageRequestDto,
+  MESSAGE_MAKE,
   SCHEMA_CREATE,
   SCHEMA_GET,
   SCHEMA_LIST,
@@ -312,4 +314,38 @@ export class AttestationController {
       },
     });
   }
+
+  @Post("/messages")
+  @ApiResponse({
+    status: 200,
+    description:
+      "Request is accepted for execution, the response id will match the event id received from the web socket",
+    type: GatewayAcceptedResponseDto,
+  })
+  @ApiBadRequestResponse({
+    status: 400,
+    description:
+      "Error in sending data to attestation manager. This error shows that attestation manager could not convert request to event or attestation manager could not send the event to the broker.",
+    type: BadRequestException,
+  })
+  @ApiInternalServerErrorResponse({
+    status: 500,
+    description: "Unknown error",
+  })
+  @ApiOperation({
+    summary: "Send basic message",
+    description:
+      "Method will send basic message to a connection. The id of the response will be matched when you receive event from the websocket",
+    tags: ["Credentials Offers"],
+  })
+  sendMeesage(@Body() message: MakeBasicMessageRequestDto) {
+    return this.amClient.sendPayload({
+      pattern: "messages",
+      payload: {
+        source: "messages",
+        data: message,
+        type: MESSAGE_MAKE,
+      },
+    });
+  }
 }

compose/docker-compose.simple.yml

+version: '3.8'
+
+services:
+  pg_db:
+    image: 'postgres:latest'
+    ports:
+      - '5432:5432'
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+    volumes:
+      - ./data/db-simple/:/var/lib/postgresql/data/
+
+  agent-issuer-simple:
+    container_name: agent-issuer-simple
+    build:
+      context: "../"
+      dockerfile: "./apps/agent/deployment/Dockerfile"
+    env_file:
+      - ./env/issuer.simple.env
+    ports:
+      - "8080:8080"
+      - "8001:8001"
+    depends_on:
+      pg_db:
+        condition: service_started
+
+  agent-holder-simple:
+    container_name: agent-holder-simple
+    build:
+      context: "../"
+      dockerfile: "./apps/agent/deployment/Dockerfile"
+    env_file:
+      - ./env/holder.simple.env
+    ports:
+      - "8081:8080"
+      - "6001:6001"
+    depends_on:
+      pg_db:
+        condition: service_started

compose/env/holder.simple.env

+LEDGERS="BCOVRIN_TEST"
+IDUNION_KEY=
+
+AGENT_PEER_URL="http://agent-holder:6001"
+AGENT_NAME=DEV_SIMPLE_AGENT_HOLDER_OCM # this should be changed to company name
+AGENT_KEY=DEV_SIMPLE_AGENT_HOLDER_OCM #example random string
+AGENT_DID_SEED=200000000000000000000000TCuste21xh #did private key seed min lenght 32
+AGENT_DB_HOST=pg_db:5432
+AGENT_DB_USER=postgres
+AGENT_DB_PASS=postgres
+AGENT_PORT=8081
+AGENT_CONSUMER_NAME=agent_1
+AGENT_IS_REST=true
+AGENT_MAX_MESSAGES=10
+AGENT_RETE_LIMIT=5
+
+NATS_SERVERS=broker-holder:4222
+NATS_STREAM_NAME=ssi_holder_stream
+NATS_SUBJECTS="connections.*,proofs.*,credentials.*,schemas.*"
+
+GATEWAY_HTTP_PORT=8081
+GATEWAY_TCP_PORT=8881
+GATEWAY_SOCKET_EVENT_NAME=message
+GATEWAY_MESSAGE_PATTERN=webhook
+GATEWAY_HOST=gateway-holder
+
+CONNECTION_SERVICE_TCP_PORT=8882
+CONNECTION_SERVICE_HOST=cm-holder
+
+ATTESTATION_SERVICE_TCP_PORT=8883
+ATTESTATION_SERVICE_HOST=ap-holder
+
+PROOF_SERVICE_TCP_PORT=8884
+PROOF_SERVICE_HOST=pm-holder

compose/env/issuer.simple.env

+LEDGERS="BCOVRIN_TEST"
+IDUNION_KEY=
+
+AGENT_PEER_URL="http://agent-issuer:8001"
+AGENT_NAME=DEV_SIMPLE_AGENT_ISSUER_OCM # this should be changed to company name
+AGENT_KEY=DEV_SIMPLE_AGENT_ISSUER_OCM #example random string
+AGENT_DID_SEED=200000000000000000000000TCuste21js #did private key seed min lenght 32
+AGENT_DB_HOST=pg_db:5432
+AGENT_DB_USER=postgres
+AGENT_DB_PASS=postgres
+AGENT_PORT=8081
+AGENT_CONSUMER_NAME=agent_1
+AGENT_IS_REST=true
+AGENT_MAX_MESSAGES=10
+AGENT_RETE_LIMIT=5
+
+NATS_SERVERS=broker-issuer:4222
+NATS_STREAM_NAME=ssi_issuer_stream
+NATS_SUBJECTS="connections.*,proofs.*,credentials.*,schemas.*"
+
+GATEWAY_HTTP_PORT=8081
+GATEWAY_TCP_PORT=8881
+GATEWAY_SOCKET_EVENT_NAME=message
+GATEWAY_MESSAGE_PATTERN=webhook
+GATEWAY_HOST=gateway-issuer
+
+CONNECTION_SERVICE_TCP_PORT=8882
+CONNECTION_SERVICE_HOST=cm-issuer
+
+ATTESTATION_SERVICE_TCP_PORT=8883
+ATTESTATION_SERVICE_HOST=ap-issuer
+
+PROOF_SERVICE_TCP_PORT=8884
+PROOF_SERVICE_HOST=pm-issuer

libs/asker/src/asker-nats/agent.consumer.service.ts

@@ -28,6 +28,8 @@ import {
   GetSchemaRequestDto,
   IssueCredentialRequestDto,
   IssueProofRequestDto,
+  MakeBasicMessageRequestDto,
+  MESSAGE_MAKE,
   PROOF_ACCEPT,
   PROOF_ISSUE,
   PROOF_LIST,
@@ -35,6 +37,7 @@ import {
   SCHEMA_GET,
   SCHEMA_LIST,
 } from "@ocm-engine/dtos";
+import asyncRetry from "async-retry";
 
 @Injectable()
 export class AgentConsumerService implements OnModuleInit, OnModuleDestroy {
@@ -61,77 +64,98 @@ export class AgentConsumerService implements OnModuleInit, OnModuleDestroy {
       let data;
       let dto;
 
-      switch (event.type) {
-        case CONNECTION_CREATE:
-          data = await this.agentService.createInvitation();
-          break;
-        case CONNECTION_ACCEPT:
-          dto = event.data as CreateInvitationResponseDto;
-          data = await this.agentService.acceptInvitation(dto.invitationUrl);
-          break;
-
-        case CONNECTION_LIST:
-          data = await this.agentService.fetchConnections();
-          break;
-
-        case CONNECTION_GET:
-          dto = event.data as GetConnectionRequestDto;
-          data = await this.agentService.getConnectionById(dto.connectionId);
-          break;
-
-        case SCHEMA_CREATE:
-          dto = event.data as CreateSchemaRequestDto;
-          data = await this.agentService.createSchema(dto);
-          break;
-        case SCHEMA_LIST:
-          data = await this.agentService.fetchSchemas();
-          break;
-
-        case SCHEMA_GET:
-          dto = event.data as GetSchemaRequestDto;
-          data = await this.agentService.getSchemaById(dto.schemaId);
-          break;
-        case CRED_DEF_CREATE:
-          data = await this.agentService.createCredentialDefinition(
-            event.data as CreateCredentialDefinitionRequsetDto,
-          );
-          break;
-
-        case CRED_ISSUE:
-          data = await this.agentService.issueCredential(
-            event.data as IssueCredentialRequestDto,
-          );
-          break;
-
-        case CRED_LIST:
-          data = await this.agentService.credentials();
-          break;
-
-        case CRED_OFFER_LIST:
-          data = await this.agentService.credentialByStatedOfferReceived();
-          break;
-
-        case CRED_OFFER_ACCEPT:
-          dto = event.data as AcceptCredentialOfferRequestDto;
-          data = await this.agentService.acceptCredential(
-            dto.credentialRecordId,
-          );
-          break;
-
-        case PROOF_ISSUE:
-          dto = event.data as IssueProofRequestDto;
-          data = await this.agentService.issueProof(dto);
-          break;
-
-        case PROOF_LIST:
-          data = await this.agentService.proofs();
-          break;
-
-        case PROOF_ACCEPT:
-          dto = event.data as AcceptProofRequestDto;
-          data = await this.agentService.acceptProof(dto.proofRecordId);
-          break;
-      }
+      await asyncRetry(
+        async () => {
+          switch (event.type) {
+            case CONNECTION_CREATE:
+              data = await this.agentService.createInvitation();
+              break;
+            case CONNECTION_ACCEPT:
+              dto = event.data as CreateInvitationResponseDto;
+              data = await this.agentService.acceptInvitation(
+                dto.invitationUrl,
+              );
+              break;
+
+            case CONNECTION_LIST:
+              data = await this.agentService.fetchConnections();
+              break;
+
+            case CONNECTION_GET:
+              dto = event.data as GetConnectionRequestDto;
+              data = await this.agentService.getConnectionById(
+                dto.connectionId,
+              );
+              break;
+
+            case SCHEMA_CREATE:
+              dto = event.data as CreateSchemaRequestDto;
+              data = await this.agentService.createSchema(dto);
+              break;
+            case SCHEMA_LIST:
+              data = await this.agentService.fetchSchemas();
+              break;
+
+            case SCHEMA_GET:
+              dto = event.data as GetSchemaRequestDto;
+              data = await this.agentService.getSchemaById(dto.schemaId);
+              break;
+            case CRED_DEF_CREATE:
+              data = await this.agentService.createCredentialDefinition(
+                event.data as CreateCredentialDefinitionRequsetDto,
+              );
+              break;
+
+            case CRED_ISSUE:
+              data = await this.agentService.issueCredential(
+                event.data as IssueCredentialRequestDto,
+              );
+              break;
+
+            case CRED_LIST:
+              data = await this.agentService.credentials();
+              break;
+
+            case CRED_OFFER_LIST:
+              data = await this.agentService.credentialByStatedOfferReceived();
+              break;
+
+            case CRED_OFFER_ACCEPT:
+              dto = event.data as AcceptCredentialOfferRequestDto;
+              data = await this.agentService.acceptCredential(
+                dto.credentialRecordId,
+              );
+              break;
+
+            case PROOF_ISSUE:
+              dto = event.data as IssueProofRequestDto;
+              data = await this.agentService.issueProof(dto);
+              break;
+
+            case PROOF_LIST:
+              data = await this.agentService.proofs();
+              break;
+
+            case PROOF_ACCEPT:
+              dto = event.data as AcceptProofRequestDto;
+              data = await this.agentService.acceptProof(dto.proofRecordId);
+              break;
+
+            case MESSAGE_MAKE:
+              dto = event.data as MakeBasicMessageRequestDto;
+              data = await this.agentService.sendMessage(dto);
+              break;
+          }
+        },
+        {
+          retries: 5,
+          maxTimeout: 5000,
+          onRetry: (error) => {
+            this.logger.log(JSON.stringify(error, null, 2));
+            this.logger.error(`Fail...${error.message}`);
+          },
+        },
+      );
 
       event.data = data;
       this.logger.debug(`before sending ${JSON.stringify(event, null, 2)}`);

libs/asker/src/asker-rest/rest.controller.ts

@@ -6,11 +6,12 @@ import {
   IssueCredentialRequestDto,
   IssueProofRequestDto,
   CreateSchemaRequestDto,
+  MakeBasicMessageRequestDto,
 } from "@ocm-engine/dtos";
 import { AllExceptionsHandler } from "./exception.handler";
 
 @UseFilters(AllExceptionsHandler)
-@Controller()
+@Controller("v1")
 export class RestController {
   constructor(private readonly agentService: AgentService) {}
 
@@ -84,4 +85,9 @@ export class RestController {
   resolve(@Body("did") did: string) {
     return this.agentService.resolve(did);
   }
+
+  @Post("/messages")
+  sendMeesage(@Body() message: MakeBasicMessageRequestDto) {
+    return this.agentService.sendMessage(message);
+  }
 }

libs/asker/src/asker/agent.event.listener.servce.ts

+import { Injectable, Logger, OnModuleInit } from "@nestjs/common";
+import { GatewayClient } from "@ocm-engine/clients";
+import { AskerService } from "./asker.service";
+import {
+  BasicMessageEventTypes,
+  BasicMessageRole,
+  BasicMessageStateChangedEvent,
+} from "@aries-framework/core";
+import {
+  MakeBasicMessageResponseDto,
+  makeEvent,
+  MESSAGE_MAKE,
+} from "@ocm-engine/dtos";
+import { IConfAgent } from "@ocm-engine/config";
+import { ConfigService } from "@nestjs/config";
+
+@Injectable()
+export class AgentEventListenerServce implements OnModuleInit {
+  private agentConfig: IConfAgent;
+  private readonly logger: Logger = new Logger(AgentEventListenerServce.name);
+
+  constructor(
+    private readonly gatewayClient: GatewayClient,
+    private readonly asker: AskerService,
+    private readonly configService: ConfigService,
+  ) {}
+
+  onModuleInit(): void {
+    this.logger.debug("Agent is listening for AFJ events");
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    this.agentConfig = this.configService.get<IConfAgent>("agent")!;
+
+    this.asker.agent.events.on(
+      BasicMessageEventTypes.BasicMessageStateChanged,
+      async (ev: BasicMessageStateChangedEvent) => {
+        if (ev.payload.basicMessageRecord.role === BasicMessageRole.Receiver) {
+          this.logger.debug(JSON.stringify(ev, null, 2));
+
+          const dto = new MakeBasicMessageResponseDto();
+          dto.message = ev.payload.basicMessageRecord.content;
+          dto.id = ev.payload.basicMessageRecord.id;
+
+          if (this.agentConfig.agentIsRest) {
+            this.logger.debug(
+              "agent is configured as rest, webhook still not implemented",
+            );
+
+            return;
+          }
+
+          const event = makeEvent({
+            data: dto,
+            type: MESSAGE_MAKE,
+            source: "agent-basic-message-afj",
+          });
+
+          this.logger.debug("Sending message event to gateway");
+          this.gatewayClient.sendPayload(event);
+        }
+      },
+    );
+  }
+}

libs/asker/src/asker/agent.service.ts

@@ -13,6 +13,8 @@ import {
   IssueCredentialResponseDto,
   IssueProofRequestDto,
   IssueProofResponseDto,
+  MakeBasicMessageRequestDto,
+  MakeBasicMessageResponseDto,
   SchemaNotCreatedError,
 } from "@ocm-engine/dtos";
 import {
@@ -309,4 +311,18 @@ export class AgentService {
   resolve = async (did: string) => {
     return this.asker.agent.dids.resolve(did);
   };
+
+  sendMessage = async (message: MakeBasicMessageRequestDto) => {
+    const response = new MakeBasicMessageResponseDto();
+
+    const m = await this.asker.agent.basicMessages.sendMessage(
+      message.connectionId,
+      message.message,
+    );
+
+    response.id = m.id;
+    response.message = m.content;
+
+    return response;
+  };
 }

libs/asker/src/asker/asker.module.ts

@@ -3,11 +3,12 @@ import { AskerService } from "./asker.service";
 import { AgentService } from "./agent.service";
 import { ConfigModule } from "@nestjs/config";
 import { LedgersModule } from "@ocm-engine/ledgers";
+import { AgentEventListenerServce } from "./agent.event.listener.servce";
 
 @Global()
 @Module({
   imports: [ConfigModule, LedgersModule],
-  providers: [AgentService, AskerService],
+  providers: [AgentService, AskerService, AgentEventListenerServce],
   exports: [AgentService, AskerService],
 })
 export class AskerModule {}

libs/asker/src/asker/asker.service.ts

@@ -10,7 +10,6 @@ import {
   ConsoleLogger,
   HttpOutboundTransport,
   InitConfig,
-  KeyDerivationMethod,
   LogLevel,
   TypedArrayEncoder,
   WsOutboundTransport,

libs/clients/src/lib/gateway.client.ts

@@ -16,7 +16,6 @@ export class GatewayClient {
 
   constructor(configService: ConfigService) {
     this.gatewayConf = configService.get<IGateway>("gateway")!;
-    console.log(this.gatewayConf);
 
     this.client = ClientProxyFactory.create({
       transport: Transport.TCP,

libs/dtos/src/dtos/requests/make.basic.message.request.dto.ts

+import { IsNotEmpty, IsString } from "class-validator";
+
+export class MakeBasicMessageRequestDto {
+  @IsNotEmpty()
+  @IsString()
+  connectionId: string;
+
+  @IsNotEmpty()
+  @IsString()
+  message: string;
+}

libs/dtos/src/dtos/responses/make.basic.message.response.dto.ts

+import { IsNotEmpty, IsString } from "class-validator";
+
+export class MakeBasicMessageResponseDto {
+  @IsNotEmpty()
+  @IsString()
+  id: string;
+
+  @IsNotEmpty()
+  @IsString()
+  message: string;
+}

libs/dtos/src/events/dtoToEventTransformer.ts

 import {
   ALL_EVENTS,
+  BasicMessageEvent,
   ConnectionEvent,
   CredentialEvent,
   ProofEvent,
@@ -14,6 +15,8 @@ import { GetConnectionRequestDto } from "../dtos/requests/get.connection.request
 import { IssueCredentialRequestDto } from "../dtos/requests/issue.credential.request.dto";
 import { IssueProofRequestDto } from "../dtos/requests/issue.proof.request.dto";
 import { AcceptCredentialOfferRequestDto } from "../dtos/requests/accept.credential.offer.request.dto";
+import { MakeBasicMessageResponseDto } from "../dtos/responses/make.basic.message.response.dto";
+import { MakeBasicMessageRequestDto } from "../dtos/requests/make.basic.message.request.dto";
 
 export const makeEvent = (payload: {
   data:
@@ -25,8 +28,15 @@ export const makeEvent = (payload: {
     | CreateSchemaRequestDto
     | CreateSchemaRequestDto
     | CreateCredentialDefinitionRequsetDto
-    | IssueCredentialRequestDto;
-  type: SchemaEvent | CredentialEvent | ProofEvent | ConnectionEvent;
+    | IssueCredentialRequestDto
+    | MakeBasicMessageResponseDto
+    | MakeBasicMessageRequestDto;
+  type:
+    | SchemaEvent
+    | CredentialEvent
+    | ProofEvent
+    | ConnectionEvent
+    | BasicMessageEvent;
   source: string;
 }) => {
   if (ALL_EVENTS.includes(payload.type)) {

libs/dtos/src/events/types.ts

@@ -60,9 +60,15 @@ export const PROOF_EVENTS: ProofEvent[] = [
   PROOF_ISSUE,
 ];
 
+export type BasicMessageEvent = "messages.make";
+export const MESSAGE_MAKE = "messages.make";
+
+export const DIDCOMM_EVENTS: BasicMessageEvent[] = [MESSAGE_MAKE];
+
 export const ALL_EVENTS = [
   ...SCHEMA_EVENTS,
   ...CRED_EVENTS,
   ...PROOF_EVENTS,
   ...CONNECTION_EVENTS,
+  ...DIDCOMM_EVENTS,
 ];

libs/dtos/src/index.ts

@@ -7,6 +7,7 @@ export * from "./dtos/requests/get.connection.request.dto";
 export * from "./dtos/requests/get.schema.request.dto";
 export * from "./dtos/requests/accept.credential.offer.request.dto";
 export * from "./dtos/requests/accept.proof.request.dto";
+export * from "./dtos/requests/make.basic.message.request.dto";
 
 export * from "./dtos/responses/create.invitation.response.dto";
 export * from "./dtos/responses/accept.invitation.response.dto";
@@ -15,6 +16,7 @@ export * from "./dtos/responses/create.credential.definition.response.dto";
 export * from "./dtos/responses/issue.credential.response.dto";
 export * from "./dtos/responses/issue.proof.response.dto";
 export * from "./dtos/responses/gateway.accepted.response.dto";
+export * from "./dtos/responses/make.basic.message.response.dto";
 
 export * from "./errors/connection.not.found.error";
 export * from "./errors/schema.not.created.error";

package.json

@@ -28,7 +28,9 @@
     "infra:issuer:stop": "cd compose && docker-compose --profile issuer stop",
     "infra:holder:stop": "cd compose && docker-compose --profile holder stop",
     "infra:local": "cd compose && docker-compose -f=docker-compose.infra.yml up -d",
-    "infra:local:stop": "cd compose && docker-compose -f=docker-compose.infra.yml stop"
+    "infra:local:stop": "cd compose && docker-compose -f=docker-compose.infra.yml stop",
+
+    "infra:simple": "cd compose && docker-compose -f docker-compose.simple.yml up -d"
   },
   "private": true,
   "dependencies": {