-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
4 changed files
with
257 additions
and
254 deletions.
There are no files selected for viewing
252 changes: 252 additions & 0 deletions
252
apps/api/src/workflow-triggers/controllers/chatbot.controller.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,252 @@ | ||
| import { XmtpMessageOutput } from '@app/definitions/integration-definitions/xmtp/xmtp.common' | ||
| import { BadRequestException, Body, Controller, Logger, Post, Req, UnauthorizedException } from '@nestjs/common' | ||
| import { Request } from 'express' | ||
| import { uniq } from 'lodash' | ||
| import { ObjectId } from 'mongodb' | ||
| import { Types } from 'mongoose' | ||
| import { RunnerService } from '../../../../runner/src/services/runner.service' | ||
| import { ContactService } from '../../contacts/services/contact.service' | ||
| import { IntegrationTrigger } from '../../integration-triggers/entities/integration-trigger' | ||
| import { IntegrationTriggerService } from '../../integration-triggers/services/integration-trigger.service' | ||
| import { Integration } from '../../integrations/entities/integration' | ||
| import { IntegrationService } from '../../integrations/services/integration.service' | ||
| import { WorkflowActionService } from '../../workflow-actions/services/workflow-action.service' | ||
| import { WorkflowRunStatus } from '../../workflow-runs/entities/workflow-run-status' | ||
| import { WorkflowSleep } from '../../workflow-runs/entities/workflow-sleep' | ||
| import { WorkflowRunService } from '../../workflow-runs/services/workflow-run.service' | ||
| import { WorkflowSleepService } from '../../workflow-runs/services/workflow-sleep.service' | ||
| import { Workflow } from '../../workflows/entities/workflow' | ||
| import { WorkflowService } from '../../workflows/services/workflow.service' | ||
| import { WorkflowTrigger } from '../entities/workflow-trigger' | ||
| import { WorkflowTriggerService } from '../services/workflow-trigger.service' | ||
| import { WorkflowUsedIdService } from '../services/workflow-used-id.service' | ||
|
|
||
| @Controller('/chatbots') | ||
| export class ChatbotController { | ||
| private readonly logger = new Logger(ChatbotController.name) | ||
|
|
||
| private chatbotIntegration: Integration | ||
| private chatbotIntegrationTrigger: IntegrationTrigger | ||
| private xmtpIntegration: Integration | ||
| private xmtpIntegrationTrigger: IntegrationTrigger | ||
|
|
||
| constructor( | ||
| private readonly integrationService: IntegrationService, | ||
| private readonly integrationTriggerService: IntegrationTriggerService, | ||
| private readonly workflowService: WorkflowService, | ||
| private readonly workflowTriggerService: WorkflowTriggerService, | ||
| private readonly workflowActionService: WorkflowActionService, | ||
| private readonly workflowRunService: WorkflowRunService, | ||
| private readonly runnerService: RunnerService, | ||
| private workflowUsedIdService: WorkflowUsedIdService, | ||
| private workflowSleepService: WorkflowSleepService, | ||
| private contactService: ContactService, | ||
| ) {} | ||
|
|
||
| async onModuleInit() { | ||
| this.chatbotIntegration = (await this.integrationService.findOne({ key: 'chatbot' })) as Integration | ||
| this.chatbotIntegrationTrigger = (await this.integrationTriggerService.findOne({ | ||
| integration: this.chatbotIntegration._id, | ||
| key: 'newChatbotMessage', | ||
| })) as IntegrationTrigger | ||
|
|
||
| this.xmtpIntegration = (await this.integrationService.findOne({ key: 'xmtp' })) as Integration | ||
| this.xmtpIntegrationTrigger = (await this.integrationTriggerService.findOne({ | ||
| integration: this.xmtpIntegration._id, | ||
| key: 'newMessage', | ||
| })) as IntegrationTrigger | ||
| } | ||
|
|
||
| @Post('/') | ||
| async received(@Body() body: Record<string, any>, @Req() req: Request) { | ||
| if (req.headers?.authorization !== process.env.CHATBOT_SECRET) { | ||
| throw new UnauthorizedException() | ||
| } | ||
| if (!body.user || !body.message) { | ||
| throw new BadRequestException() | ||
| } | ||
|
|
||
| const chatbotWorkflowTriggers = await this.workflowTriggerService.find({ | ||
| owner: new ObjectId(body.user), | ||
| integrationTrigger: this.chatbotIntegrationTrigger._id, | ||
| enabled: true, | ||
| planLimited: { $ne: true }, | ||
| }) | ||
| const chatbotPromises = chatbotWorkflowTriggers.map(async (workflowTrigger) => | ||
| this.processChatbotMessage(body.message, workflowTrigger), | ||
| ) | ||
| await Promise.all(chatbotPromises) | ||
|
|
||
| const xmtpWorkflowTriggers = await this.workflowTriggerService.find({ | ||
| owner: new ObjectId(body.user), | ||
| integrationTrigger: this.xmtpIntegrationTrigger._id, | ||
| enabled: true, | ||
| planLimited: { $ne: true }, | ||
| }) | ||
| const xmtpPromises = xmtpWorkflowTriggers.map(async (workflowTrigger) => | ||
| this.processXmtpMessage(body.message, workflowTrigger), | ||
| ) | ||
| await Promise.all(xmtpPromises) | ||
|
|
||
| return { ok: true } | ||
| } | ||
|
|
||
| async processChatbotMessage(message: XmtpMessageOutput, workflowTrigger: WorkflowTrigger) { | ||
| try { | ||
| await this.workflowUsedIdService.createOne({ | ||
| workflow: workflowTrigger.workflow, | ||
| triggerId: message.id, | ||
| }) | ||
| } catch (e) { | ||
| return | ||
| } | ||
|
|
||
| const workflow = await this.workflowService.findOne({ _id: workflowTrigger.workflow }) | ||
| if (!workflow) { | ||
| return | ||
| } | ||
|
|
||
| this.logger.log(`Processing chatbot message: ${message.id} for workflow: ${workflow._id}`) | ||
|
|
||
| const workflowSleeps = await this.workflowSleepService.find({ | ||
| workflow: workflowTrigger.workflow, | ||
| uniqueGroup: message.conversation.id, | ||
| }) | ||
|
|
||
| // continue previous conversation | ||
| if (workflowSleeps.length > 0) { | ||
| void this.continueConversation(workflow, workflowTrigger, workflowSleeps, message) | ||
| return | ||
| } | ||
|
|
||
| const tags = workflowTrigger.inputs?.tags?.split(',').map((tag) => tag.trim()) ?? [] | ||
| const contact = await this.contactService.findOne({ | ||
| owner: workflow.owner, | ||
| address: message.senderAddress, | ||
| }) | ||
| if (!contact) { | ||
| await this.contactService.createOne({ | ||
| owner: workflow.owner, | ||
| address: message.senderAddress, | ||
| tags, | ||
| }) | ||
| } else if (workflowTrigger.inputs?.tags) { | ||
| const newTags = uniq([...contact.tags, ...tags]) | ||
| if (newTags.length !== contact.tags.length) { | ||
| await this.contactService.updateById(contact._id, { | ||
| tags: contact.tags, | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| const hookTriggerOutputs = { | ||
| id: message.id, | ||
| outputs: { | ||
| [workflowTrigger.id]: message as Record<string, any>, | ||
| trigger: message as Record<string, any>, | ||
| contact: { | ||
| address: message.senderAddress, | ||
| }, | ||
| }, | ||
| } | ||
| const rootActions = await this.workflowActionService.find({ workflow: workflow._id, isRootAction: true }) | ||
| const workflowRun = await this.workflowRunService.createOneByInstantTrigger( | ||
| this.chatbotIntegration, | ||
| this.chatbotIntegrationTrigger, | ||
| workflow, | ||
| workflowTrigger, | ||
| rootActions.length > 0, | ||
| ) | ||
| await this.workflowTriggerService.updateById(workflowTrigger._id, { | ||
| lastId: message.id, | ||
| lastItem: message, | ||
| }) | ||
| void this.runnerService.runWorkflowActions(rootActions, [hookTriggerOutputs], workflowRun) | ||
| } | ||
|
|
||
| async continueConversation( | ||
| workflow: Workflow, | ||
| workflowTrigger: WorkflowTrigger, | ||
| workflowSleeps: WorkflowSleep[], | ||
| outputs: XmtpMessageOutput, | ||
| ) { | ||
| const workflowSleep = workflowSleeps[0] | ||
|
|
||
| // clean up | ||
| await this.workflowSleepService.deleteManyNative({ | ||
| _id: { | ||
| $in: workflowSleeps.map((workflowSleep) => workflowSleep._id), | ||
| }, | ||
| }) | ||
|
|
||
| this.logger.log(`Continuing chatbot conversation ${workflowSleep.id} for workflow ${workflowTrigger.workflow}`) | ||
|
|
||
| const workflowAction = await this.workflowActionService.findById(workflowSleep.workflowAction.toString()) | ||
| const workflowRun = await this.workflowRunService.findById(workflowSleep.workflowRun.toString()) | ||
|
|
||
| if (!workflowAction || !workflowRun) { | ||
| this.logger.error(`Missing workflow action or workflow run for workflow sleep ${workflowSleep.id}`) | ||
| await this.workflowRunService.updateById(workflowSleep._id, { status: WorkflowRunStatus.failed }) | ||
| return | ||
| } | ||
|
|
||
| await this.workflowRunService.wakeUpWorkflowRun(workflowRun) | ||
| const nextActionInputs = { | ||
| ...(workflowSleep.nextActionInputs ?? {}), | ||
| [workflowAction.id]: { | ||
| ...((workflowSleep.nextActionInputs?.[workflowAction.id] as any) ?? {}), | ||
| responseId: outputs.id, | ||
| responseContent: outputs.content, | ||
| }, | ||
| } as Record<string, Record<string, unknown>> | ||
| const actions = await this.workflowActionService.findByIds( | ||
| workflowAction.nextActions.map((next) => next.action) as Types.ObjectId[], | ||
| ) | ||
| const promises = actions.map((action) => | ||
| this.runnerService.runWorkflowActionsTree(workflow, action, nextActionInputs, workflowRun, workflowSleep.itemId), | ||
| ) | ||
| void Promise.all(promises).then(() => { | ||
| return this.workflowRunService.markWorkflowRunAsCompleted(workflowRun._id) | ||
| }) | ||
| } | ||
|
|
||
| async processXmtpMessage(message: XmtpMessageOutput, workflowTrigger: WorkflowTrigger) { | ||
| try { | ||
| await this.workflowUsedIdService.createOne({ | ||
| workflow: workflowTrigger.workflow, | ||
| triggerId: message.id, | ||
| }) | ||
| } catch (e) { | ||
| return | ||
| } | ||
|
|
||
| const workflow = await this.workflowService.findOne({ _id: workflowTrigger.workflow }) | ||
| if (!workflow) { | ||
| return | ||
| } | ||
|
|
||
| this.logger.log(`Processing xmtp message: ${message.id} for workflow: ${workflow._id}`) | ||
|
|
||
| const hookTriggerOutputs = { | ||
| id: message.id, | ||
| outputs: { | ||
| [workflowTrigger.id]: message as Record<string, any>, | ||
| trigger: message as Record<string, any>, | ||
| }, | ||
| } | ||
|
|
||
| const rootActions = await this.workflowActionService.find({ workflow: workflow._id, isRootAction: true }) | ||
| const workflowRun = await this.workflowRunService.createOneByInstantTrigger( | ||
| this.xmtpIntegration, | ||
| this.xmtpIntegrationTrigger, | ||
| workflow, | ||
| workflowTrigger, | ||
| rootActions.length > 0, | ||
| ) | ||
| await this.workflowTriggerService.updateById(workflowTrigger._id, { | ||
| lastId: message.id, | ||
| lastItem: message, | ||
| }) | ||
| void this.runnerService.runWorkflowActions(rootActions, [hookTriggerOutputs], workflowRun) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.