feat(converse): converse api (#194)

* update api.rest * basic collector * refact * fix * comment * example collect * create user and convo api.rest * Update packages/server/src/converse/service.ts Co-authored-by: Laurent Leclerc Poulin <laurentlp@users.noreply.github.com> * message queue * refact * runMessageQueue * change timeout * fix * queue by threadId * fix failing message failed the entire queue * improve queue * fix merge Co-authored-by: Laurent Leclerc Poulin <laurentlp@users.noreply.github.com>
botpress · Oct 18, 2021 · f4ecbca · f4ecbca
1 parent 335cf3c
commit f4ecbca
Show file tree

Hide file tree

Showing 7 changed files with 133 additions and 16 deletions.
diff --git a/api.rest b/api.rest
@@ -40,15 +40,32 @@ Authorization: Bearer {{authToken}}
 x-bp-messaging-client-id: {{clientId}}
 x-bp-messaging-client-token: {{clientToken}}
 
+### Create User
+POST {{baseUrl}}/api/users
+Authorization: Bearer {{authToken}}
+x-bp-messaging-client-id: {{clientId}}
+x-bp-messaging-client-token: {{clientToken}}
+
+### Create Conversation
+POST {{baseUrl}}/api/conversations
+Authorization: Bearer {{authToken}}
+x-bp-messaging-client-id: {{clientId}}
+x-bp-messaging-client-token: {{clientToken}}
+Content-Type: application/json
+
+{
+  "userId": "my-user-id"
+}
+
 ### Send Text
-POST {{baseUrl}}/api/chat/reply
+POST {{baseUrl}}/api/messages
 Authorization: Bearer {{authToken}}
 x-bp-messaging-client-id: {{clientId}}
 x-bp-messaging-client-token: {{clientToken}}
 Content-Type: application/json
 
 {
-  "channel": "{{channel}}",
+  "collect": false,
   "conversationId": "{{convoId}}",
   "payload": {
     "type": "text",
@@ -57,14 +74,13 @@ Content-Type: application/json
 }
 
 ### Send Image
-POST {{baseUrl}}/api/chat/reply
+POST {{baseUrl}}/api/messages
 Authorization: Bearer {{authToken}}
 x-bp-messaging-client-id: {{clientId}}
 x-bp-messaging-client-token: {{clientToken}}
 Content-Type: application/json
 
 {
-  "channel": "{{channel}}",
   "conversationId": "{{convoId}}",
   "payload": {
     "type": "image",
@@ -74,14 +90,13 @@ Content-Type: application/json
 }
 
 ### Send Choices
-POST {{baseUrl}}/api/chat/reply
+POST {{baseUrl}}/api/messages
 Authorization: Bearer {{authToken}}
 x-bp-messaging-client-id: {{clientId}}
 x-bp-messaging-client-token: {{clientToken}}
 Content-Type: application/json
 
 {
-  "channel": "{{channel}}",
   "conversationId": "{{convoId}}",
   "payload": {
     "type": "single-choice",
@@ -94,14 +109,13 @@ Content-Type: application/json
 }
 
 ### Send Card
-POST {{baseUrl}}/api/chat/reply
+POST {{baseUrl}}/api/messages
 Authorization: Bearer {{authToken}}
 x-bp-messaging-client-id: {{clientId}}
 x-bp-messaging-client-token: {{clientToken}}
 Content-Type: application/json
 
 {
-  "channel": "{{channel}}",
   "conversationId": "{{convoId}}",
   "payload": {
     "type": "card",
@@ -129,14 +143,13 @@ Content-Type: application/json
 }
 
 ### Send Carousel
-POST {{baseUrl}}/api/chat/reply
+POST {{baseUrl}}/api/messages
 Authorization: Bearer {{authToken}}
 x-bp-messaging-client-id: {{clientId}}
 x-bp-messaging-client-token: {{clientToken}}
 Content-Type: application/json
 
 {
-  "channel": "{{channel}}",
   "conversationId": "{{convoId}}",
   "payload": {
     "type": "carousel",

diff --git a/packages/server/src/api.ts b/packages/server/src/api.ts
@@ -46,6 +46,7 @@ export class Api {
       this.sockets,
       this.app.conversations,
       this.app.messages,
+      this.app.converse,
       this.app.sockets
     )
     this.channels = new ChannelApi(this.root, this.app)

diff --git a/packages/server/src/app.ts b/packages/server/src/app.ts
@@ -4,6 +4,7 @@ import { ChannelService } from './channels/service'
 import { ClientService } from './clients/service'
 import { ConduitService } from './conduits/service'
 import { ConversationService } from './conversations/service'
+import { ConverseService } from './converse/service'
 import { CryptoService } from './crypto/service'
 import { DatabaseService } from './database/service'
 import { DistributedService } from './distributed/service'
@@ -43,6 +44,7 @@ export class App {
   users: UserService
   conversations: ConversationService
   messages: MessageService
+  converse: ConverseService
   mapping: MappingService
   status: StatusService
   instances: InstanceService
@@ -70,6 +72,7 @@ export class App {
     this.users = new UserService(this.database, this.caching, this.batching)
     this.conversations = new ConversationService(this.database, this.caching, this.batching, this.users)
     this.messages = new MessageService(this.database, this.caching, this.batching, this.conversations)
+    this.converse = new ConverseService(this.messages)
     this.mapping = new MappingService(this.database, this.caching, this.batching, this.users, this.conversations)
     this.status = new StatusService(this.database, this.distributed, this.caching, this.conduits)
     this.instances = new InstanceService(
@@ -138,6 +141,7 @@ export class App {
     await this.users.setup()
     await this.conversations.setup()
     await this.messages.setup()
+    await this.converse.setup()
     await this.mapping.setup()
     await this.status.setup()
     await this.instances.setup()

diff --git a/packages/server/src/converse/service.ts b/packages/server/src/converse/service.ts
@@ -0,0 +1,75 @@
+import { Message, uuid } from '@botpress/messaging-base'
+import { Service } from '../base/service'
+import { MessageCreatedEvent, MessageEvents } from '../messages/events'
+import { MessageService } from '../messages/service'
+import { Collector } from './types'
+
+export class ConverseService extends Service {
+  private collectors: { [conversationId: string]: Collector[] } = {}
+
+  constructor(private messages: MessageService) {
+    super()
+  }
+
+  async setup() {
+    this.messages.events.on(MessageEvents.Created, this.handleMessageCreated.bind(this))
+  }
+
+  private async handleMessageCreated({ message }: MessageCreatedEvent) {
+    const collectors = this.collectors[message.conversationId] || []
+
+    for (const collector of collectors) {
+      collector.messages.push(message)
+
+      if (collector.messages.length > 0) {
+        this.resetCollectorTimeout(collector, 250)
+      }
+    }
+  }
+
+  async collect(conversationId: uuid): Promise<Message[]> {
+    const collector = this.addCollector(conversationId)
+    this.resetCollectorTimeout(collector, 5000)
+
+    return new Promise<Message[]>((resolve) => {
+      collector.resolve = resolve
+    })
+  }
+
+  private addCollector(conversationId: uuid): Collector {
+    if (!this.collectors[conversationId]) {
+      this.collectors[conversationId] = []
+    }
+
+    const collector: Collector = {
+      conversationId,
+      messages: []
+    }
+
+    this.collectors[conversationId].push(collector)
+
+    return collector
+  }
+
+  private removeCollector(collector: Collector) {
+    const { conversationId } = collector
+
+    const index = this.collectors[conversationId].indexOf(collector)
+    this.collectors[conversationId].splice(index, 1)
+
+    if (this.collectors[conversationId].length === 0) {
+      delete this.collectors[conversationId]
+    }
+  }
+
+  private resetCollectorTimeout(collector: Collector, time: number) {
+    if (collector.timeout) {
+      clearTimeout(collector.timeout)
+    }
+
+    collector.timeout = setTimeout(() => {
+      this.removeCollector(collector)
+      collector.resolve!(collector.messages)
+    }, time)
+  }
+}
diff --git a/packages/server/src/converse/types.ts b/packages/server/src/converse/types.ts
@@ -0,0 +1,8 @@
+import { uuid, Message } from '@botpress/messaging-base'
+
+export interface Collector {
+  conversationId: uuid
+  messages: Message[]
+  resolve?: (x: Message[]) => void
+  timeout?: NodeJS.Timeout
+}
diff --git a/packages/server/src/messages/api.ts b/packages/server/src/messages/api.ts
@@ -2,6 +2,7 @@ import { uuid } from '@botpress/messaging-base'
 import { Router } from 'express'
 import { Auth } from '../base/auth/auth'
 import { ConversationService } from '../conversations/service'
+import { ConverseService } from '../converse/service'
 import { SocketManager } from '../socket/manager'
 import { SocketService } from '../socket/service'
 import {
@@ -21,6 +22,7 @@ export class MessageApi {
     private sockets: SocketManager,
     private conversations: ConversationService,
     private messages: MessageService,
+    private converse: ConverseService,
     private socketService: SocketService
   ) {}
 
@@ -33,7 +35,7 @@ export class MessageApi {
           return res.status(400).send(error.message)
         }
 
-        const { conversationId, authorId, payload } = req.body
+        const { conversationId, authorId, payload, collect } = req.body
         const conversation = await this.conversations.get(conversationId)
 
         if (!conversation) {
@@ -42,11 +44,24 @@ export class MessageApi {
           return res.sendStatus(403)
         }
 
-        const message = await this.messages.create(conversationId, authorId, payload, {
-          client: { id: req.client!.id }
-        })
-
-        res.send(message)
+        const collector = collect ? this.converse.collect(conversationId) : undefined
+
+        const message = await this.messages.create(
+          conversationId,
+          authorId,
+          payload,
+          authorId
+            ? undefined
+            : {
+                client: { id: req.client!.id }
+              }
+        )
+
+        if (collect) {
+          res.send(await collector)
+        } else {
+          res.send(message)
+        }
       })
     )
 

diff --git a/packages/server/src/messages/schema.ts b/packages/server/src/messages/schema.ts
@@ -1,6 +1,7 @@
 import Joi from 'joi'
 
 export const CreateMsgSchema = Joi.object({
+  collect: Joi.boolean().optional(),
   conversationId: Joi.string().guid().required(),
   authorId: Joi.string().guid().optional(),
   payload: Joi.object().required()