feat(api-service): Conversational Agents — full inbound/outbound pipeline with bridge executor fixes NV-7346

apps/api/package.json

@@ -44,6 +44,10 @@
   },
   "dependencies": {
     "@aws-sdk/client-secrets-manager": "^3.971.0",
+    "@chat-adapter/slack": "^4.25.0",
+    "@chat-adapter/state-redis": "^4.25.0",
+    "@chat-adapter/teams": "^4.25.0",
+    "@chat-adapter/whatsapp": "^4.25.0",
     "@godaddy/terminus": "^4.12.1",
     "@google-cloud/storage": "^6.2.3",
     "@nestjs/axios": "3.0.3",
@@ -82,6 +86,7 @@
     "bcrypt": "^5.0.0",
     "body-parser": "^2.2.1",
     "bull": "^4.2.1",
+    "chat": "^4.25.0",
     "class-transformer": "0.5.1",
     "class-validator": "0.14.1",
     "clickhouse-migrations": "^1.2.0",

apps/api/src/app.module.ts

@@ -83,6 +83,10 @@ const enterpriseImports = (): Array<Type | DynamicModule | Promise<DynamicModule
       modules.push(require('@novu/ee-ai')?.AiModule);
     }
 
+    if (require('@novu/ee-api')?.ConversationsModule) {
+      modules.push(require('@novu/ee-api')?.ConversationsModule);
+    }
+
     modules.push(SupportModule);
     modules.push(OutboundWebhooksModule.forRoot());
   }

apps/api/src/app/agents/agents-webhook.controller.ts

@@ -0,0 +1,80 @@
+import {
+  Body,
+  Controller,
+  HttpCode,
+  HttpException,
+  HttpStatus,
+  Param,
+  Post,
+  Req,
+  Res,
+  UseGuards,
+} from '@nestjs/common';
+import { ApiExcludeController } from '@nestjs/swagger';
+import { UserSessionData } from '@novu/shared';
+import { Request, Response } from 'express';
+import { RequireAuthentication } from '../auth/framework/auth.decorator';
+import { ExternalApiAccessible } from '../auth/framework/external-api.decorator';
+import { UserSession } from '../shared/framework/user.decorator';
+import { AgentReplyPayloadDto } from './dtos/agent-reply-payload.dto';
+import { AgentConversationEnabledGuard } from './guards/agent-conversation-enabled.guard';
+import { ChatSdkService } from './services/chat-sdk.service';
+import { HandleAgentReplyCommand, Signal } from './usecases/handle-agent-reply/handle-agent-reply.command';
+import { HandleAgentReply } from './usecases/handle-agent-reply/handle-agent-reply.usecase';
+
+@Controller('/agents')
+@UseGuards(AgentConversationEnabledGuard)
+@ApiExcludeController()
+export class AgentsWebhookController {
+  constructor(
+    private chatSdkService: ChatSdkService,
+    private handleAgentReplyUsecase: HandleAgentReply
+  ) {}
+
+  @Post('/:agentId/reply')
+  @HttpCode(HttpStatus.OK)
+  @RequireAuthentication()
+  @ExternalApiAccessible()
+  async handleAgentReply(
+    @UserSession() user: UserSessionData,
+    @Param('agentId') agentId: string,
+    @Body() body: AgentReplyPayloadDto
+  ) {
+    return this.handleAgentReplyUsecase.execute(
+      HandleAgentReplyCommand.create({
+        userId: user._id,
+        environmentId: user.environmentId,
+        organizationId: user.organizationId,
+        conversationId: body.conversationId,
+        agentIdentifier: agentId,
+        integrationIdentifier: body.integrationIdentifier,
+        reply: body.reply,
+        update: body.update,
+        resolve: body.resolve,
+        signals: body.signals as Signal[],
+      })
+    );
+  }
+
+  @Post('/:agentId/webhook/:integrationIdentifier')
+  @HttpCode(HttpStatus.OK)
+  async handleInboundWebhook(
+    @Param('agentId') agentId: string,
+    @Param('integrationIdentifier') integrationIdentifier: string,
+    @Req() req: Request,
+    @Res() res: Response
+  ) {
+    try {
+      console.log('handleInboundWebhook', agentId, integrationIdentifier);
+      await this.chatSdkService.handleWebhook(agentId, integrationIdentifier, req, res);
+      console.log('handleInboundWebhook success');
+    } catch (err) {
+      console.log(err);
+      if (err instanceof HttpException) {
+        res.status(err.getStatus()).json(err.getResponse());
+      } else {
+        res.status(HttpStatus.INTERNAL_SERVER_ERROR).json({ error: 'Internal server error' });
+      }
+    }
+  }
+}

apps/api/src/app/agents/agents.controller.ts

@@ -10,9 +10,10 @@ import {
   Patch,
   Post,
   Query,
+  UseGuards,
   UseInterceptors,
 } from '@nestjs/common';
-import { ApiOperation, ApiTags } from '@nestjs/swagger';
+import { ApiExcludeController, ApiOperation } from '@nestjs/swagger';
 import { RequirePermissions } from '@novu/application-generic';
 import { ApiRateLimitCategoryEnum, DirectionEnum, PermissionsEnum, UserSessionData } from '@novu/shared';
 import { RequireAuthentication } from '../auth/framework/auth.decorator';
@@ -36,6 +37,7 @@ import {
   UpdateAgentIntegrationRequestDto,
   UpdateAgentRequestDto,
 } from './dtos';
+import { AgentConversationEnabledGuard } from './guards/agent-conversation-enabled.guard';
 import { AddAgentIntegrationCommand } from './usecases/add-agent-integration/add-agent-integration.command';
 import { AddAgentIntegration } from './usecases/add-agent-integration/add-agent-integration.usecase';
 import { CreateAgentCommand } from './usecases/create-agent/create-agent.command';
@@ -59,7 +61,8 @@ import { UpdateAgent } from './usecases/update-agent/update-agent.usecase';
 @ApiCommonResponses()
 @Controller('/agents')
 @UseInterceptors(ClassSerializerInterceptor)
-@ApiTags('Agents')
+@UseGuards(AgentConversationEnabledGuard)
+@ApiExcludeController()
 @RequireAuthentication()
 export class AgentsController {
   constructor(

apps/api/src/app/agents/agents.module.ts

@@ -1,14 +1,39 @@
 import { Module } from '@nestjs/common';
+import {
+  ChannelConnectionRepository,
+  ChannelEndpointRepository,
+  ConversationActivityRepository,
+  ConversationRepository,
+} from '@novu/dal';
 
 import { AuthModule } from '../auth/auth.module';
 import { SharedModule } from '../shared/shared.module';
 import { AgentsController } from './agents.controller';
+import { AgentsWebhookController } from './agents-webhook.controller';
+import { AgentConversationService } from './services/agent-conversation.service';
+import { AgentCredentialService } from './services/agent-credential.service';
+import { AgentInboundHandler } from './services/agent-inbound-handler.service';
+import { AgentSubscriberResolver } from './services/agent-subscriber-resolver.service';
+import { BridgeExecutorService } from './services/bridge-executor.service';
+import { ChatSdkService } from './services/chat-sdk.service';
 import { USE_CASES } from './usecases';
 
 @Module({
   imports: [SharedModule, AuthModule],
-  controllers: [AgentsController],
-  providers: [...USE_CASES],
-  exports: [...USE_CASES],
+  controllers: [AgentsController, AgentsWebhookController],
+  providers: [
+    ...USE_CASES,
+    ChannelConnectionRepository,
+    ChannelEndpointRepository,
+    ConversationRepository,
+    ConversationActivityRepository,
+    AgentCredentialService,
+    AgentSubscriberResolver,
+    AgentConversationService,
+    AgentInboundHandler,
+    BridgeExecutorService,
+    ChatSdkService,
+  ],
+  exports: [...USE_CASES, ChatSdkService],
 })
 export class AgentsModule {}

apps/api/src/app/agents/dtos/agent-event.enum.ts

@@ -0,0 +1,6 @@
+export enum AgentEventEnum {
+  ON_START = 'onStart',
+  ON_MESSAGE = 'onMessage',
+  ON_ACTION = 'onAction',
+  ON_RESOLVE = 'onResolve',
+}

apps/api/src/app/agents/dtos/agent-platform.enum.ts

@@ -0,0 +1,5 @@
+export enum AgentPlatformEnum {
+  SLACK = 'slack',
+  WHATSAPP = 'whatsapp',
+  TEAMS = 'teams',
+}

apps/api/src/app/agents/dtos/agent-reply-payload.dto.ts

@@ -0,0 +1,67 @@
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+import { Type } from 'class-transformer';
+import { IsArray, IsIn, IsNotEmpty, IsObject, IsOptional, IsString, MaxLength, ValidateNested } from 'class-validator';
+
+const SIGNAL_TYPES = ['metadata', 'trigger'] as const;
+
+export class TextContentDto {
+  @ApiProperty()
+  @IsString()
+  @IsNotEmpty()
+  @MaxLength(40_000)
+  text: string;
+}
+
+export class ResolveDto {
+  @ApiPropertyOptional()
+  @IsOptional()
+  @IsString()
+  summary?: string;
+}
+
+export class SignalDto {
+  @ApiProperty({ enum: SIGNAL_TYPES })
+  @IsString()
+  @IsIn(SIGNAL_TYPES)
+  type: (typeof SIGNAL_TYPES)[number];
+}
+
+export class AgentReplyPayloadDto {
+  @ApiProperty()
+  @IsString()
+  @IsNotEmpty()
+  conversationId: string;
+
+  @ApiProperty()
+  @IsString()
+  @IsNotEmpty()
+  integrationIdentifier: string;
+
+  @ApiPropertyOptional({ type: TextContentDto })
+  @IsOptional()
+  @IsObject()
+  @ValidateNested()
+  @Type(() => TextContentDto)
+  reply?: TextContentDto;
+
+  @ApiPropertyOptional({ type: TextContentDto })
+  @IsOptional()
+  @IsObject()
+  @ValidateNested()
+  @Type(() => TextContentDto)
+  update?: TextContentDto;
+
+  @ApiPropertyOptional({ type: ResolveDto })
+  @IsOptional()
+  @IsObject()
+  @ValidateNested()
+  @Type(() => ResolveDto)
+  resolve?: ResolveDto;
+
+  @ApiPropertyOptional({ type: [SignalDto] })
+  @IsOptional()
+  @IsArray()
+  @ValidateNested({ each: true })
+  @Type(() => SignalDto)
+  signals?: SignalDto[];
+}

apps/api/src/app/agents/e2e/agents.e2e.ts

@@ -8,6 +8,10 @@ describe('Agents API - /agents #novu-v2', () => {
   const agentRepository = new AgentRepository();
   const agentIntegrationRepository = new AgentIntegrationRepository();
 
+  before(() => {
+    process.env.IS_CONVERSATIONAL_AGENTS_ENABLED = 'true';
+  });
+
   beforeEach(async () => {
     session = new UserSession();
     await session.initialize();

apps/api/src/app/agents/guards/agent-conversation-enabled.guard.ts

@@ -0,0 +1,26 @@
+import { CanActivate, ExecutionContext, Injectable, NotFoundException } from '@nestjs/common';
+import { FeatureFlagsService } from '@novu/application-generic';
+import { FeatureFlagsKeysEnum, UserSessionData } from '@novu/shared';
+
+@Injectable()
+export class AgentConversationEnabledGuard implements CanActivate {
+  constructor(private readonly featureFlagsService: FeatureFlagsService) {}
+
+  async canActivate(context: ExecutionContext): Promise<boolean> {
+    const request = context.switchToHttp().getRequest();
+    const user: UserSessionData | undefined = request.user;
+
+    const isEnabled = await this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_CONVERSATIONAL_AGENTS_ENABLED,
+      defaultValue: false,
+      organization: { _id: user?.organizationId ?? '' },
+      environment: { _id: user?.environmentId ?? '' },
+    });
+
+    if (!isEnabled) {
+      throw new NotFoundException();
+    }
+
+    return true;
+  }
+}