feat(socket): socket package (#222)

* user token start * tests * refactor auth * caching * api * fix test * comment * expiry * batching * remove user token implementation * Revert "remove user token implementation" This reverts commit 1e1dd78. * refact auth * remove file * fix * refact client auth function * refact function * refact * fix * remove file * end turn route * responses * return message also * use cache for collectors * hintRespondingTo * incomingId * bump messaging client * turn id * collect flag * fix ui * backend validate user token is correct * schema * socket auth * update testing ui * bring changes * remove user token api * remove user token api * revert * remove correct user token api * bring changes * extract constant * verifyToken * fix tests * users.get * socket prevent lost requests * conversations socket * some refact and changes * refact schema validation * refact user rights check * remove nonsense * socket refact * move socket code at of api start * socket root class * remove private * refact schema * fix * fix * dont server client * remove ui from changes * custom timeout * bring changes * fix * tests start * fix * more tests * tests * pr comments * fix
botpress · Oct 29, 2021 · 4c1141e · 4c1141e
1 parent b16535d
commit 4c1141e
Show file tree

Hide file tree

Showing 27 changed files with 693 additions and 366 deletions.
diff --git a/packages/server/src/api.ts b/packages/server/src/api.ts
@@ -8,13 +8,10 @@ import { ChannelApi } from './channels/api'
 import { ConversationApi } from './conversations/api'
 import { HealthApi } from './health/api'
 import { MessageApi } from './messages/api'
-import { SocketManager } from './socket/manager'
 import { SyncApi } from './sync/api'
 import { UserApi } from './users/api'
 
 export class Api {
-  public readonly sockets: SocketManager
-
   private router: Router
   private auth: Auth
 
@@ -28,35 +25,11 @@ export class Api {
   constructor(private app: App, private root: Router) {
     this.router = Router()
     this.auth = new Auth(app.clients)
-    this.sockets = new SocketManager(this.app.sockets)
     this.syncs = new SyncApi(this.router, this.auth, this.app.syncs, this.app.clients, this.app.channels)
     this.health = new HealthApi(this.router, this.auth, this.app.health)
-    this.users = new UserApi(
-      this.router,
-      this.auth,
-      this.app.clients,
-      this.sockets,
-      this.app.users,
-      this.app.userTokens,
-      this.app.sockets
-    )
-    this.conversations = new ConversationApi(
-      this.router,
-      this.auth,
-      this.sockets,
-      this.app.users,
-      this.app.conversations,
-      this.app.sockets
-    )
-    this.messages = new MessageApi(
-      this.router,
-      this.auth,
-      this.sockets,
-      this.app.conversations,
-      this.app.messages,
-      this.app.converse,
-      this.app.sockets
-    )
+    this.users = new UserApi(this.router, this.auth, this.app.users)
+    this.conversations = new ConversationApi(this.router, this.auth, this.app.conversations)
+    this.messages = new MessageApi(this.router, this.auth, this.app.conversations, this.app.messages, this.app.converse)
     this.channels = new ChannelApi(this.root, this.app)
   }
 

diff --git a/packages/server/src/conversations/api.ts b/packages/server/src/conversations/api.ts
@@ -1,26 +1,16 @@
 import { Router } from 'express'
 import { Auth } from '../base/auth/auth'
-import { SocketManager } from '../socket/manager'
-import { SocketService } from '../socket/service'
-import { UserService } from '../users/service'
-import { CreateConvoSchema, GetConvoSchema, ListConvosSchema, RecentConvoSchema, UseConvoSocketSchema } from './schema'
+import { Schema } from './schema'
 import { ConversationService } from './service'
 
 export class ConversationApi {
-  constructor(
-    private router: Router,
-    private auth: Auth,
-    private sockets: SocketManager,
-    private users: UserService,
-    private conversations: ConversationService,
-    private socketService: SocketService
-  ) {}
+  constructor(private router: Router, private auth: Auth, private conversations: ConversationService) {}
 
   async setup() {
     this.router.post(
       '/conversations',
       this.auth.client.auth(async (req, res) => {
-        const { error } = CreateConvoSchema.validate(req.body)
+        const { error } = Schema.Api.Create.validate(req.body)
         if (error) {
           return res.status(400).send(error.message)
         }
@@ -35,7 +25,7 @@ export class ConversationApi {
     this.router.get(
       '/conversations/:id',
       this.auth.client.auth(async (req, res) => {
-        const { error } = GetConvoSchema.validate(req.params)
+        const { error } = Schema.Api.Get.validate(req.params)
         if (error) {
           return res.status(400).send(error.message)
         }
@@ -54,9 +44,9 @@ export class ConversationApi {
     )
 
     this.router.get(
-      '/conversations/',
+      '/conversations',
       this.auth.client.auth(async (req, res) => {
-        const { error } = ListConvosSchema.validate(req.query)
+        const { error } = Schema.Api.List.validate(req.query)
         if (error) {
           return res.status(400).send(error.message)
         }
@@ -75,7 +65,7 @@ export class ConversationApi {
     this.router.get(
       '/conversations/:userId/recent',
       this.auth.client.auth(async (req, res) => {
-        const { error } = RecentConvoSchema.validate(req.params)
+        const { error } = Schema.Api.Recent.validate(req.params)
         if (error) {
           return res.status(400).send(error.message)
         }
@@ -89,30 +79,5 @@ export class ConversationApi {
         res.send(conversation)
       })
     )
-
-    this.sockets.handle('conversations.use', async (socket, message) => {
-      const { error } = UseConvoSocketSchema.validate(message.data)
-      if (error) {
-        return this.sockets.reply(socket, message, { error: true, message: error.message })
-      }
-
-      const userId = this.socketService.getUserId(socket)
-      if (!userId) {
-        return this.sockets.reply(socket, message, {
-          error: true,
-          message: 'socket does not have user rights'
-        })
-      }
-      const { conversationId }: { conversationId?: string } = message.data
-
-      const user = await this.users.get(userId)
-      let conversation = conversationId && (await this.conversations.get(conversationId))
-
-      if (!conversation || conversation.userId !== userId) {
-        conversation = await this.conversations.create(user!.clientId, userId)
-      }
-
-      this.sockets.reply(socket, message, conversation)
-    })
   }
 }
diff --git a/packages/server/src/conversations/schema.ts b/packages/server/src/conversations/schema.ts
@@ -1,22 +1,38 @@
 import Joi from 'joi'
 
-export const CreateConvoSchema = Joi.object({
-  userId: Joi.string().guid().required()
-})
-
-export const GetConvoSchema = Joi.object({
-  id: Joi.string().guid().required()
-})
-
-export const ListConvosSchema = Joi.object({
-  userId: Joi.string().guid().required(),
-  limit: Joi.number().required()
-})
-
-export const RecentConvoSchema = Joi.object({
-  userId: Joi.string().guid().required()
-})
-
-export const UseConvoSocketSchema = Joi.object({
-  conversationId: Joi.string().guid().optional()
-})
+const Api = {
+  Create: Joi.object({
+    userId: Joi.string().guid().required()
+  }),
+
+  Get: Joi.object({
+    id: Joi.string().guid().required()
+  }),
+
+  List: Joi.object({
+    userId: Joi.string().guid().required(),
+    limit: Joi.number().required()
+  }),
+
+  Recent: Joi.object({
+    userId: Joi.string().guid().required()
+  })
+}
+
+const Socket = {
+  Create: Joi.object({}),
+
+  Get: Joi.object({
+    id: Joi.string().guid().required()
+  }),
+
+  List: Joi.object({
+    limit: Joi.number().required()
+  }),
+
+  Delete: Joi.object({
+    id: Joi.string().guid().required()
+  })
+}
+
+export const Schema = { Api, Socket }
diff --git a/packages/server/src/conversations/socket.ts b/packages/server/src/conversations/socket.ts
@@ -0,0 +1,56 @@
+import { SocketManager, SocketRequest } from '../socket/manager'
+import { UserService } from '../users/service'
+import { Schema } from './schema'
+import { ConversationService } from './service'
+
+export class ConversationSocket {
+  constructor(private sockets: SocketManager, private users: UserService, private conversations: ConversationService) {}
+
+  setup() {
+    this.sockets.handle('conversations.create', Schema.Socket.Create, this.create.bind(this))
+    this.sockets.handle('conversations.get', Schema.Socket.Get, this.get.bind(this))
+    this.sockets.handle('conversations.list', Schema.Socket.List, this.list.bind(this))
+    this.sockets.handle('conversations.delete', Schema.Socket.Delete, this.delete.bind(this))
+  }
+
+  async create(socket: SocketRequest) {
+    const user = await this.users.get(socket.userId)
+    const conversation = await this.conversations.create(user!.clientId, user!.id)
+
+    socket.reply(conversation)
+  }
+
+  async get(socket: SocketRequest) {
+    const { id } = socket.data
+    const conversation = await this.conversations.get(id)
+
+    if (!conversation || conversation.userId !== socket.userId) {
+      return socket.reply(undefined)
+    }
+
+    socket.reply(conversation)
+  }
+
+  async list(socket: SocketRequest) {
+    const { limit } = socket.data
+
+    const user = await this.users.get(socket.userId)
+    const conversations = await this.conversations.listByUserId(user!.clientId, socket.userId, +limit)
+
+    socket.reply(conversations)
+  }
+
+  async delete(socket: SocketRequest) {
+    const { id } = socket.data
+    const conversation = await this.conversations.get(id)
+
+    if (!conversation) {
+      return socket.reply(false)
+    } else if (conversation.userId !== socket.userId) {
+      return socket.forbid('Conversation does not belong to user')
+    }
+
+    await this.conversations.delete(id)
+    socket.reply(true)
+  }
+}
diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts
@@ -7,6 +7,7 @@ import yn from 'yn'
 import { Api } from './api'
 import { App } from './app'
 import { Launcher } from './launcher'
+import { Socket } from './socket'
 
 // Set NODE_ENV to production when starting messaging using the binary version
 process.env.NODE_ENV = process.pkg ? 'production' : process.env.NODE_ENV
@@ -18,8 +19,9 @@ const launch = async () => {
 
   const app = new App()
   const api = new Api(app, router)
+  const socket = new Socket(app)
 
-  const launcher = new Launcher(router, app, api)
+  const launcher = new Launcher(router, app, api, socket)
   await launcher.launch()
 }
 

diff --git a/packages/server/src/launcher.ts b/packages/server/src/launcher.ts
@@ -10,6 +10,7 @@ import { Api } from './api'
 import { App } from './app'
 import { ShutDownSignal } from './base/errors'
 import { Logger } from './logger/types'
+import { Socket } from './socket'
 
 const pkg = require('../package.json')
 
@@ -19,7 +20,7 @@ export class Launcher {
   private httpTerminator: HttpTerminator | undefined
   private readonly shutdownTimeout: number = ms('5s')
 
-  constructor(private express: Express, private app: App, private api: Api) {
+  constructor(private express: Express, private app: App, private api: Api, private socket: Socket) {
     this.logger = new Logger('Launcher')
 
     process.on('uncaughtException', async (e) => {
@@ -56,6 +57,7 @@ export class Launcher {
       this.printChannels()
 
       await this.api.setup()
+      await this.socket.setup()
 
       let port = process.env.PORT
       if (!port) {
@@ -64,7 +66,7 @@ export class Launcher {
       }
 
       const server = this.express.listen(port)
-      await this.api.sockets.setup(server)
+      await this.socket.manager.setup(server)
       this.httpTerminator = createHttpTerminator({ server, gracefulTerminationTimeout: this.shutdownTimeout })
 
       if (!yn(process.env.SPINNED)) {
@@ -97,7 +99,7 @@ export class Launcher {
 
       this.logger.info('Server gracefully closing down...')
 
-      await this.api.sockets.destroy()
+      await this.socket.manager.destroy()
       await this.httpTerminator?.terminate()
       await this.app.destroy()