Enhancement/websockets (#413)

Author: Uros Marolt

backend/package-lock.json

@@ -95,6 +95,7 @@
     "sanitize-html": "^2.7.1",
     "sequelize": "6.21.2",
     "sequelize-cli-typescript": "^3.2.0-c",
+    "socket.io": "^4.5.4",
     "stripe": "10.0.0",
     "superagent": "^8.0.0",
     "swagger-ui-dist": "4.1.3",

backend/package.json

@@ -1,5 +1,5 @@
 import Error403 from '../../errors/Error403'
-import { RedisCache } from '../../utils/redis'
+import { RedisCache } from '../../utils/redis/redisCache'
 
 export default async (req, res) => {
   if (!req.currentUser || !req.currentUser.id) {

backend/src/api/auth/authMe.ts

@@ -3,6 +3,7 @@ import cors from 'cors'
 import helmet from 'helmet'
 import bunyanMiddleware from 'bunyan-middleware'
 import { PostHog } from 'posthog-node'
+import * as http from 'http'
 import { authMiddleware } from '../middlewares/authMiddleware'
 import { tenantMiddleware } from '../middlewares/tenantMiddleware'
 import { databaseMiddleware } from '../middlewares/databaseMiddleware'
@@ -16,16 +17,41 @@ import { responseHandlerMiddleware } from '../middlewares/responseHandlerMiddlew
 import { errorMiddleware } from '../middlewares/errorMiddleware'
 import { passportStrategyMiddleware } from '../middlewares/passportStrategyMiddleware'
 import { redisMiddleware } from '../middlewares/redisMiddleware'
-import { createRedisClient } from '../utils/redis'
 import { POSTHOG_CONFIG } from '../config'
+import { createRedisClient, createRedisPubSubPair } from '../utils/redis'
+import WebSockets from './websockets'
+import RedisPubSubReceiver from '../utils/redis/pubSubReceiver'
+import { ApiWebsocketMessage } from '../types/mq/apiWebsocketMessage'
 
 const serviceLogger = createServiceLogger()
 
 const app = express()
 
+const server = http.createServer(app)
+
 setImmediate(async () => {
   const redis = await createRedisClient(true)
 
+  const redisPubSubPair = await createRedisPubSubPair()
+  const userNamespace = await WebSockets.initialize(server)
+
+  const pubSubReceiver = new RedisPubSubReceiver('api-pubsub', redisPubSubPair.subClient, (err) => {
+    serviceLogger.error(err, 'Error while listening to Redis Pub/Sub api-ws channel!')
+    process.exit(1)
+  })
+
+  pubSubReceiver.subscribe('user', async (message) => {
+    const data = message as ApiWebsocketMessage
+
+    if (data.tenantId) {
+      await userNamespace.emitForTenant(data.tenantId, data.event, data.data)
+    } else if (data.userId) {
+      userNamespace.emitToUserRoom(data.userId, data.event, data.data)
+    } else {
+      serviceLogger.error({ type: data.type }, 'Received invalid websocket message!')
+    }
+  })
+
   let posthog = null
 
   if (POSTHOG_CONFIG.apiKey) {
@@ -151,4 +177,4 @@ setImmediate(async () => {
   app.use(io.expressErrorHandler())
 })
 
-export default app
+export default server

backend/src/api/index.ts

@@ -4,7 +4,7 @@ import identifyTenant from '../../segment/identifyTenant'
 import track from '../../segment/track'
 import MemberService from '../../services/memberService'
 import PermissionChecker from '../../services/user/permissionChecker'
-import { RedisCache } from '../../utils/redis'
+import { RedisCache } from '../../utils/redis/redisCache'
 
 /**
  * POST /tenant/{tenantId}/member/export