reimplement message_server as new chat_server

docker-compose.yml

@@ -15,13 +15,13 @@ services:
       - api.env
     volumes:
       - ./d_data/api/cache:/home/node/app/cache
-  message_server:
+  chat_server:
-    build: packages/message_server
+    build: packages/chat_server
     restart: unless-stopped
     ports:
       - "5001:3020"
     env_file:
-      - messaging.env
+      - chat.env
   marketplace_server:
     build: packages/marketplace_server
     restart: unless-stopped

package.json

@@ -6,8 +6,8 @@
     "scripts": {
         "postinstall": "npm rebuild @tensorflow/tfjs-node --build-from-source",
         "release": "node ./scripts/release.js",
-        "dev": "concurrently -k -n Server,MarketplaceServer,MessageServer,Client -c bgCyan,bgCyan,bgCyan,bgGreen \"yarn dev:server\" \"yarn dev:marketplace_server\" \"yarn dev:message_server\" \"yarn dev:client\"",
+        "dev": "concurrently -k -n Server,MarketplaceServer,ChatServer,Client -c bgCyan,bgCyan,bgCyan,bgGreen \"yarn dev:server\" \"yarn dev:marketplace_server\" \"yarn dev:chat_server\" \"yarn dev:client\"",
-        "dev:message_server": "cd packages/message_server && yarn dev",
+        "dev:chat_server": "cd packages/chat_server && yarn dev",
         "dev:marketplace_server": "cd packages/marketplace_server && yarn dev",
         "dev:server": "cd packages/server && yarn dev",
         "dev:client": "cd packages/app && yarn dev"

packages/app/src/components/LiveChat/index.jsx

@@ -1,10 +1,8 @@
 import React from "react"
 import * as antd from "antd"
 import classnames from "classnames"
-import { io } from "socket.io-client"
 import { TransitionGroup, CSSTransition } from "react-transition-group"
 
-import config from "config"
 import SessionModel from "models/session"
 
 import "./index.less"
@@ -24,12 +22,18 @@ const Line = (props) => {
 export default class LiveChat extends React.Component {
     state = {
         socket: null,
+
         connecting: true,
         connectionEnd: false,
+
         roomInfo: null,
+
         timeline: [],
         temporalTimeline: [],
+
+        lastSentMessage: null,
         writtedMessage: "",
+
         maxTemporalLines: this.props.maxTemporalLines ?? 10,
     }
 
@@ -37,25 +41,36 @@ export default class LiveChat extends React.Component {
 
     timelineRef = React.createRef()
 
-    joinSocketRoom = async () => {
+    socket = app.cores.api.instance().wsInstances.chat
-        await this.setState({ connecting: true })
 
-        const { roomId } = this.props
+    roomEvents = {
+        "room:recive:message": (message) => {
+            if (message.content === this.state.lastSentMessage) {
+                console.timeEnd("[CHATROOM] SUBMIT:MESSAGE")
+            }
 
-        const socketNamespace = `/textRoom/${roomId}`
+            this.pushToTimeline(message)
+        },
+        "room:joined": (info) => {
+            console.log("[CHATROOM] Room joined", info)
 
-        console.log(`Joining socket room [${socketNamespace}]`)
+            this.setState({
+                connecting: false,
+                roomInfo: info,
+            })
+        },
+        "room:leave": (info) => {
+            console.log("[CHATROOM] Room left", info)
 
-        const socket = io(config.remotes.messagingApi, {
+            this.setState({
-            transports: ["websocket"],
+                connecting: false,
-            autoConnect: false,
+                roomInfo: null,
-        })
+            })
-
-        socket.auth = {
-            token: SessionModel.token,
         }
+    }
 
-        socket.on("connect_error", (err) => {
+    socketEvents = {
+        "connect_error": (err) => {
             console.error("Connection error", err)
 
             this.setState({ connectionEnd: true })
@@ -63,40 +78,73 @@ export default class LiveChat extends React.Component {
             if (err.message === "auth:token_invalid") {
                 console.error("Invalid token")
             }
-        })
+        },
+        "disconnect": (reason) => {
+            console.error("Disconnected", reason)
 
-        socket.on("connect", () => {
+            this.setState({ connectionEnd: true })
-            socket.emit("join", { room: socketNamespace }, (error, info) => {
+        },
-                if (error) {
+        "connect": () => {
-                    this.setState({ connectionEnd: true })
+            this.setState({ connectionEnd: false })
-                    return console.error("Error joining room", error)
-                }
 
-                this.setState({
+            this.joinSocketRoom()
-                    connecting: false,
+        }
-                    roomInfo: info,
+    }
-                })
+
+    initializeSocket = async () => {
+        if (!this.socket) {
+            console.error("Socket not initialized/avaliable")
+
+            this.setState({ connectionEnd: true })
+
+            return false
+        }
+
+        for (const [eventName, eventHandler] of Object.entries(this.roomEvents)) {
+            this.socket.on(eventName, eventHandler)
+        }
+
+        for (const [eventName, eventHandler] of Object.entries(this.socketEvents)) {
+            this.socket.on(eventName, eventHandler)
+        }
+    }
+
+    joinSocketRoom = async () => {
+        await this.setState({ connecting: true })
+
+        if (!this.socket.connected) {
+            this.socket.connect()
+        }
+
+        const { roomId } = this.props
+
+        const socketNamespace = `/textRoom/${roomId}`
+
+        console.log(`[CHATROOM] Joining socket room [${socketNamespace}]...`)
+
+        this.socket.emit("join:room", { room: socketNamespace }, (error, info) => {
+            if (error) {
+                this.setState({ connectionEnd: true })
+
+                return console.error("Error joining room", error)
+            }
+
+            this.setState({
+                connecting: true,
             })
         })
-
-        socket.on("message", (message) => {
-            this.pushToTimeline(message)
-        })
-
-        socket.connect()
-
-        this.setState({ socket })
     }
 
     submitMessage = (message) => {
-        const { socket } = this.state
+        console.time("[CHATROOM] SUBMIT:MESSAGE")
 
-        socket.emit("send:message", {
+        this.socket.emit("room:send:message", {
             message
         })
 
         // remove writted message
         this.setState({
+            lastSentMessage: message,
             writtedMessage: ""
         })
     }
@@ -187,8 +235,6 @@ export default class LiveChat extends React.Component {
     scrollTimelineToBottom = () => {
         const scrollingElement = document.getElementById("liveChat_timeline")
 
-        console.log(`Scrolling to bottom`, scrollingElement)
-
         if (scrollingElement) {
             scrollingElement.scrollTo({
                 top: scrollingElement.scrollHeight,
@@ -206,7 +252,15 @@ export default class LiveChat extends React.Component {
             })
         }
 
-        await this.joinSocketRoom()
+        await this.initializeSocket()
+
+        await this.joinSocketRoom().catch((err) => {
+            console.error("Error joining socket room", err)
+
+            this.setState({
+                connectionEnd: true
+            })
+        })
 
         app.ctx = {
             submit: this.submitMessage
@@ -214,7 +268,17 @@ export default class LiveChat extends React.Component {
     }
 
     componentWillUnmount() {
-        this.state.socket.close()
+        if (this.socket) {
+            this.socket.emit("leave:room")
+        }
+
+        for (const [eventName, eventHandler] of Object.entries(this.roomEvents)) {
+            this.socket.off(eventName, eventHandler)
+        }
+
+        for (const [eventName, eventHandler] of Object.entries(this.socketEvents)) {
+            this.socket.off(eventName, eventHandler)
+        }
 
         if (this.debouncedIntervalTimelinePurge) {
             clearInterval(this.debouncedIntervalTimelinePurge)

packages/app/src/cores/api/index.js

@@ -6,6 +6,8 @@ import measurePing from "comty.js/handlers/measurePing"
 import request from "comty.js/handlers/request"
 import useRequest from "comty.js/hooks/useRequest"
 
+import SessionModel from "comty.js/models/session"
+
 export default class APICore extends Core {
     static refName = "api"
     static namespace = "api"
@@ -33,7 +35,16 @@ export default class APICore extends Core {
 
     async onInitialize() {
         this.instance = await createClient({
-            useWs: true,
+            enableWs: true,
+            wsParams: {
+                chat: (opts) => {
+                    opts.auth = {
+                        token: SessionModel.token,
+                    }
+
+                    return opts
+                }
+            }
         })
 
         this.instance.eventBus.on("auth:login_success", () => {

packages/chat_server/src/api.js

@@ -0,0 +1,94 @@
+import fs from "fs"
+import path from "path"
+
+import express from "express"
+import http from "http"
+import EventEmitter from "@foxify/events"
+
+import ComtyClient from "@classes/ComtyClient"
+
+import routes from "./routes"
+
+import ChatServer from "./chatServer"
+
+export default class Server {
+    constructor(options = {}) {
+        this.app = express()
+        this.httpServer = http.createServer(this.app)
+
+        this.websocketServer = new ChatServer(this.httpServer)
+
+        this.options = {
+            listenHost: process.env.LISTEN_HOST || "0.0.0.0",
+            listenPort: process.env.LISTEN_PORT || 3020,
+            ...options
+        }
+    }
+
+    comty = global.comty = ComtyClient()
+
+    eventBus = global.eventBus = new EventEmitter()
+
+    async __registerInternalMiddlewares() {
+        let middlewaresPath = fs.readdirSync(path.resolve(__dirname, "useMiddlewares"))
+
+        for await (const middlewarePath of middlewaresPath) {
+            const middleware = require(path.resolve(__dirname, "useMiddlewares", middlewarePath)).default
+
+            if (!middleware) {
+                console.error(`Middleware ${middlewarePath} not found.`)
+
+                continue
+            }
+
+            this.app.use(middleware)
+        }
+    }
+
+    registerRoutes() {
+        routes.forEach((route) => {
+            const order = []
+
+            if (route.middlewares) {
+                route.middlewares.forEach((middleware) => {
+                    order.push(middleware)
+                })
+            }
+
+            order.push(route.routes)
+
+            this.app.use(route.use, ...order)
+        })
+    }
+
+    async registerBaseRoute() {
+        await this.app.get("/", async (req, res) => {
+            return res.json({
+                uptimeMinutes: Math.floor(process.uptime() / 60),
+            })
+        })
+    }
+
+    initialize = async () => {
+        const startHrTime = process.hrtime()
+
+        await this.websocketServer.initialize()
+
+        await this.__registerInternalMiddlewares()
+
+        this.app.use(express.json({ extended: false }))
+        this.app.use(express.urlencoded({ extended: true }))
+
+        await this.registerBaseRoute()
+        await this.registerRoutes()
+
+        await this.httpServer.listen(this.options.listenPort, this.options.listenHost)
+
+        // calculate elapsed time
+        const elapsedHrTime = process.hrtime(startHrTime)
+        const elapsedTimeInMs = elapsedHrTime[0] * 1000 + elapsedHrTime[1] / 1e6
+
+        // log server started
+        console.log(`🚀 Server started ready on \n\t - http://${this.options.listenHost}:${this.options.listenPort} \n\t - Tooks ${elapsedTimeInMs}ms`)
+    }
+}

packages/chat_server/src/chatServer.js

@@ -0,0 +1,255 @@
+import socketio from "socket.io"
+
+import withWsAuth from "@middlewares/withWsAuth"
+
+function generateFnHandler(fn, socket) {
+    return async (...args) => {
+        if (typeof socket === "undefined") {
+            socket = arguments[0]
+        }
+
+        try {
+            fn(socket, ...args)
+        } catch (error) {
+            console.error(`[HANDLER_ERROR] ${error.message} >`, error.stack)
+
+            if (typeof socket.emit !== "function") {
+                return false
+            }
+
+            return socket.emit("error", {
+                message: error.message,
+            })
+        }
+    }
+}
+
+class Room {
+    constructor(io, roomName) {
+        if (!io) {
+            throw new Error("io is required")
+        }
+
+        this.io = io
+        this.roomName = roomName
+    }
+
+    connections = []
+
+    limitations = {
+        maxMessageLength: 540,
+    }
+
+    events = {
+        "room:send:message": (socket, payload) => {
+            let { message } = payload
+
+            if (!message || typeof message !== "string") {
+                return socket.emit("error", {
+                    message: "Invalid message",
+                })
+            }
+
+            if (message.length > this.limitations.maxMessageLength) {
+                message = message.substring(0, this.limitations.maxMessageLength)
+            }
+
+            this.io.to(this.roomName).emit("room:recive:message", {
+                timestamp: payload.timestamp ?? Date.now(),
+                content: String(message),
+                user: {
+                    user_id: socket.userData._id,
+                    username: socket.userData.username,
+                    fullName: socket.userData.fullName,
+                    avatar: socket.userData.avatar,
+                },
+            })
+        }
+    }
+
+    join = (socket) => {
+        if (socket.connectedRoom) {
+            console.warn(`[${socket.id}][@${socket.userData.username}] already connected to room ${socket.connectedRoom}`)
+
+            this.leave(socket)
+        }
+
+        socket.connectedRoom = this.roomName
+
+        // join room
+        socket.join(this.roomName)
+
+        // emit to self
+        socket.emit("room:joined", {
+            room: this.roomName,
+            limitations: this.limitations,
+            connectedUsers: this.connections.map((conn) => {
+                return conn.user_id
+            }),
+        })
+
+        // emit to others
+        this.io.to(this.roomName).emit("room:user:joined", {
+            user: {
+                user_id: socket.userData._id,
+                username: socket.userData.username,
+                fullName: socket.userData.fullName,
+                avatar: socket.userData.avatar,
+            }
+        })
+
+        for (const [event, fn] of Object.entries(this.events)) {
+            const handler = generateFnHandler(fn, socket)
+
+            if (!Array.isArray(socket.handlers)) {
+                socket.handlers = []
+            }
+
+            socket.handlers.push([event, handler])
+
+            socket.on(event, handler)
+        }
+
+        // add to connections
+        this.connections.push(socket)
+
+        console.log(`[${socket.id}][@${socket.userData.username}] joined room ${this.roomName}`)
+    }
+
+    leave = (socket) => {
+        if (!socket.connectedRoom) {
+            console.warn(`[${socket.id}][@${socket.userData.username}] not connected to any room`)
+            return
+        }
+
+        if (socket.connectedRoom !== this.roomName) {
+            console.warn(`[${socket.id}][@${socket.userData.username}] not connected to room ${this.roomName}, cannot leave`)
+            return false
+        }
+
+        socket.leave(this.roomName)
+
+        socket.emit("room:left", {
+            room: this.roomName,
+        })
+
+        this.io.to(this.roomName).emit("room:user:left", {
+            user: {
+                user_id: socket.userData._id,
+                username: socket.userData.username,
+                fullName: socket.userData.fullName,
+                avatar: socket.userData.avatar,
+            }
+        })
+
+        for (const [event, handler] of socket.handlers) {
+            socket.off(event, handler)
+        }
+
+        this.connections.splice(this.connections.indexOf(socket), 1)
+
+        console.log(`[${socket.id}][@${socket.userData.username}] left room ${this.roomName}`)
+    }
+}
+
+class RoomsController {
+    constructor(io) {
+        if (!io) {
+            throw new Error("io is required")
+        }
+
+        this.io = io
+    }
+
+    rooms = []
+
+    checkRoomExists = (roomName) => {
+        return this.rooms.some((room) => room.roomName === roomName)
+    }
+
+    createRoom = async (roomName) => {
+        if (this.checkRoomExists(roomName)) {
+            throw new Error(`Room ${roomName} already exists`)
+        }
+
+        const room = new Room(this.io, roomName)
+
+        this.rooms.push(room)
+
+        return room
+    }
+
+    connectSocketToRoom = async (socket, roomName) => {
+        if (!this.checkRoomExists(roomName)) {
+            await this.createRoom(roomName)
+        }
+
+        const room = this.rooms.find((room) => room.roomName === roomName)
+
+        return room.join(socket)
+    }
+
+    disconnectSocketFromRoom = async (socket, roomName) => {
+        if (!this.checkRoomExists(roomName)) {
+            return false
+        }
+
+        const room = this.rooms.find((room) => room.roomName === roomName)
+
+        return room.leave(socket)
+    }
+}
+
+export default class ChatServer {
+    constructor(server) {
+        this.io = socketio(server, {
+            cors: {
+                origin: "*",
+                methods: ["GET", "POST"],
+                credentials: true,
+            }
+        })
+
+        this.RoomsController = new RoomsController(this.io)
+    }
+
+    connectionPool = []
+
+    events = {
+        "connection": (socket) => {
+            console.log(`[${socket.id}][${socket.userData.username}] connected to hub.`)
+
+            this.connectionPool.push(socket)
+
+            socket.on("disconnect", () => this.events.disconnect)
+
+            // Rooms
+            socket.on("join:room", (data) => this.RoomsController.connectSocketToRoom(socket, data.room))
+            socket.on("leave:room", (data) => this.RoomsController.disconnectSocketFromRoom(socket, data?.room ?? socket.connectedRoom))
+        },
+        "disconnect": (socket) => {
+            console.log(`[${socket.id}][@${socket.userData.username}] disconnected to hub.`)
+
+            if (socket.connectedRoom) {
+                this.Rooms.leave(socket)
+            }
+
+            // remove from connection pool
+            this.connectionPool = this.connectionPool.filter((client) => client.id !== socket.id)
+        },
+    }
+
+    initialize = async () => {
+        this.io.use(withWsAuth)
+
+        Object.entries(this.events).forEach(([event, handler]) => {
+            this.io.on(event, (socket) => {
+                try {
+                    handler(socket)
+                } catch (error) {
+                    console.error(error)
+                }
+            })
+        })
+    }
+}

packages/chat_server/src/classes/ComtyClient/index.js

@@ -0,0 +1,9 @@
+import createClient from "comty.js"
+
+export default (params = {}) => {
+    return createClient({
+        ...params,
+        accessKey: process.env.COMTY_ACCESS_KEY,
+        privateKey: process.env.COMTY_PRIVATE_KEY,
+    })
+}

packages/chat_server/src/classes/DbManager/index.js

@@ -0,0 +1,58 @@
+import mongoose from "mongoose"
+
+function getConnectionConfig(obj) {
+    const { DB_USER, DB_DRIVER, DB_NAME, DB_PWD, DB_HOSTNAME, DB_PORT } = obj
+
+    let auth = [
+        DB_DRIVER ?? "mongodb",
+        "://",
+    ]
+
+    if (DB_USER && DB_PWD) {
+        auth.push(`${DB_USER}:${DB_PWD}@`)
+    }
+
+    auth.push(DB_HOSTNAME ?? "localhost")
+    auth.push(`:${DB_PORT ?? "27017"}`)
+
+    if (DB_USER) {
+        auth.push("/?authMechanism=DEFAULT")
+    }
+
+    auth = auth.join("")
+
+    return [
+        auth,
+        {
+            dbName: DB_NAME,
+            useNewUrlParser: true,
+            useUnifiedTopology: true,
+        }
+    ]
+}
+
+export default class DBManager {
+    initialize = async (config) => {
+        console.log("🔌 Connecting to DB...")
+
+        const dbConfig = getConnectionConfig(config ?? process.env)
+
+        mongoose.set("strictQuery", false)
+
+        const connection = await mongoose.connect(...dbConfig)
+            .catch((err) => {
+                console.log(`❌ Failed to connect to DB, retrying...\n`)
+                console.log(error)
+
+                // setTimeout(() => {
+                //     this.initialize()
+                // }, 1000)
+
+                return false
+            })
+
+        if (connection) {
+            console.log(`✅ Connected to DB.`)
+        }
+    }
+}

packages/chat_server/src/classes/RedisClient/index.js

@@ -0,0 +1,44 @@
+import { createClient } from "redis"
+
+function composeURL() {
+    // support for auth
+    let url = "redis://"
+
+    if (process.env.REDIS_PASSWORD && process.env.REDIS_USERNAME) {
+        url += process.env.REDIS_USERNAME + ":" + process.env.REDIS_PASSWORD + "@"
+    }
+
+    url += process.env.REDIS_HOST ?? "localhost"
+
+    if (process.env.REDIS_PORT) {
+        url += ":" + process.env.REDIS_PORT
+    }
+
+    return url
+}
+
+export default () => {
+    let client = createClient({
+        url: composeURL(),
+    })
+
+    client.initialize = async () => {
+        console.log("🔌 Connecting to Redis client...")
+
+        await client.connect()
+
+        return client
+    }
+
+    // handle when client disconnects unexpectedly to avoid main crash
+    client.on("error", (error) => {
+        console.error("❌ Redis client error:", error)
+    })
+
+    // handle when client connects
+    client.on("connect", () => {
+        console.log("✅ Redis client connected.")
+    })
+
+    return client
+}

packages/chat_server/src/classes/StorageClient/index.js

@@ -0,0 +1,97 @@
+const Minio = require("minio")
+import path from "path"
+
+export const generateDefaultBucketPolicy = (payload) => {
+    const { bucketName } = payload
+
+    if (!bucketName) {
+        throw new Error("bucketName is required")
+    }
+
+    return {
+        Version: "2012-10-17",
+        Statement: [
+            {
+                Action: [
+                    "s3:GetObject"
+                ],
+                Effect: "Allow",
+                Principal: {
+                    AWS: [
+                        "*"
+                    ]
+                },
+                Resource: [
+                    `arn:aws:s3:::${bucketName}/*`
+                ],
+                Sid: ""
+            }
+        ]
+    }
+}
+
+export class StorageClient extends Minio.Client {
+    constructor(options) {
+        super(options)
+
+        this.defaultBucket = String(options.defaultBucket)
+        this.defaultRegion = String(options.defaultRegion)
+    }
+
+    composeRemoteURL = (key) => {
+        const _path = path.join(this.defaultBucket, key)
+
+        return `${this.protocol}//${this.host}:${this.port}/${_path}`
+    }
+
+    setDefaultBucketPolicy = async (bucketName) => {
+        const policy = generateDefaultBucketPolicy({ bucketName })
+
+        return this.setBucketPolicy(bucketName, JSON.stringify(policy))
+    }
+
+    initialize = async () => {
+        console.log("🔌 Checking if storage client have default bucket...")
+
+        // check connection with s3
+        const bucketExists = await this.bucketExists(this.defaultBucket).catch(() => {
+            return false
+        })
+
+        if (!bucketExists) {
+            console.warn("🪣 Default bucket not exists! Creating new bucket...")
+
+            await this.makeBucket(this.defaultBucket, "s3")
+
+            // set default bucket policy
+            await this.setDefaultBucketPolicy(this.defaultBucket)
+        }
+
+        // check if default bucket policy exists
+        const bucketPolicy = await this.getBucketPolicy(this.defaultBucket).catch(() => {
+            return null
+        })
+
+        if (!bucketPolicy) {
+            // set default bucket policy
+            await this.setDefaultBucketPolicy(this.defaultBucket)
+        }
+
+        console.log("✅ Storage client is ready.")
+    }
+}
+
+export const createStorageClientInstance = (options) => {
+    return new StorageClient({
+        ...options,
+        endPoint: process.env.S3_ENDPOINT,
+        port: Number(process.env.S3_PORT),
+        useSSL: toBoolean(process.env.S3_USE_SSL),
+        accessKey: process.env.S3_ACCESS_KEY,
+        secretKey: process.env.S3_SECRET_KEY,
+        defaultBucket: process.env.S3_BUCKET,
+        defaultRegion: process.env.S3_REGION,
+    })
+}
+
+export default createStorageClientInstance

packages/chat_server/src/index.js

@@ -1,5 +1,24 @@
 require("dotenv").config()
 
+if (typeof process.env.NODE_ENV === "undefined") {
+    process.env.NODE_ENV = "development"
+}
+
+global.isProduction = process.env.NODE_ENV === "production"
+
+import path from "path"
+import { registerBaseAliases } from "linebridge/dist/server"
+
+const customAliases = {
+    "@services": path.resolve(__dirname, "services"),
+}
+
+if (!global.isProduction) {
+    customAliases["comty.js"] = path.resolve(__dirname, "../../comty.js/src")
+}
+
+registerBaseAliases(undefined, customAliases)
+
 // patches
 const { Buffer } = require("buffer")
 

packages/chat_server/src/middlewares/withWsAuth.js

@@ -0,0 +1,55 @@
+export default async (socket, next) => {
+    try {
+        const token = socket.handshake.auth.token
+
+        if (!token) {
+            return next(new Error(`auth:token_missing`))
+        }
+
+        const validation = await global.comty.rest.session.validateToken(token).catch((err) => {
+            console.error(`[${socket.id}] failed to validate session caused by server error`, err)
+
+            return {
+                valid: false,
+                error: err,
+            }
+        })
+
+        if (!validation.valid) {
+            if (validation.error) {
+                return next(new Error(`auth:server_error`))
+            }
+
+            return next(new Error(`auth:token_invalid`))
+        }
+
+        const session = validation.session
+
+        const userData = await global.comty.rest.user.data({
+            user_id: session.user_id,
+        }).catch((err) => {
+            console.error(`[${socket.id}] failed to get user data caused by server error`, err)
+
+            return null
+        })
+
+        if (!userData) {
+            return next(new Error(`auth:user_failed`))
+        }
+
+        try {
+            socket.userData = userData
+            socket.token = token
+            socket.session = session
+        }
+        catch (err) {
+            return next(new Error(`auth:decode_failed`))
+        }
+
+        next()
+    } catch (error) {
+        console.error(`[${socket.id}] failed to connect caused by server error`, error)
+
+        next(new Error(`auth:authentification_failed`))
+    }
+}

packages/comty.js/src/models/session/index.js

@@ -106,9 +106,14 @@ export default class Session {
         return response.data
     }
 
+    // alias for validateToken method
     static validSession = async (token) => {
+        return await Session.validateToken(token)
+    }
+
+    static validateToken = async (token) => {
         const response = await request({
-            method: "POST",
+            method: "post",
             url: "/session/validate",
             data: {
                 token: token

packages/comty.js/src/remotes.js

@@ -13,13 +13,13 @@ function getCurrentHostname() {
 const envOrigins = {
     "development": {
         default: `http://${getCurrentHostname()}:3010`,
-        messaging: `http://${getCurrentHostname()}:3020`,
+        chat: `http://${getCurrentHostname()}:3020`,
         livestreaming: `http://${getCurrentHostname()}:3030`,
         marketplace: `http://${getCurrentHostname()}:3040`,
     },
     "production": {
         default: "https://api.comty.app",
-        messaging: `https://messaging_api.comty.app`,
+        chat: `https://chat_api.comty.app`,
         livestreaming: `https://livestreaming_api.comty.app`,
         marketplace: `https://marketplace_api.comty.app`,
     }
@@ -29,12 +29,12 @@ export default {
     default: {
         origin: composeRemote("default"),
         hasWebsocket: true,
-        needsAuth: true,
+        useClassicAuth: true,
+        autoconnect: true,
     },
-    messaging: {
+    chat: {
-        origin: composeRemote("messaging"),
+        origin: composeRemote("chat"),
         hasWebsocket: true,
-        needsAuth: true,
     },
     livestreaming: {
         origin: composeRemote("livestreaming"),

packages/message_server/src/api.js

@@ -1,266 +0,0 @@
-import fs from "fs"
-import path from "path"
-
-import express from "express"
-import http from "http"
-import socketio from "socket.io"
-import EventEmitter from "@foxify/events"
-import jwt from "jsonwebtoken"
-import axios from "axios"
-
-import routes from "./routes"
-
-const mainAPI = axios.create({
-    baseURL: process.env.MAIN_API_URL ?? "http://localhost:3010",
-    headers: {
-        Authorization: `Server ${process.env.MAIN_SERVER_ID}:${process.env.MAIN_SERVER_TOKEN}`,
-    }
-})
-
-class TextRoomServer {
-    constructor(server, options = {}) {
-        this.io = socketio(server, {
-            cors: {
-                origin: "*",
-                methods: ["GET", "POST"],
-                credentials: true,
-            }
-        })
-
-        this.limitations = {
-            maxMessageLength: 540,
-            ...options.limitations,
-        }
-    }
-
-    connectionPool = []
-
-    roomEventsHandlers = {
-        "send:message": (socket, payload) => {
-            const { connectedRoom } = socket
-            let { message } = payload
-
-            if (message.length > this.limitations.maxMessageLength) {
-                message = message.substring(0, this.limitations.maxMessageLength)
-            }
-
-            this.io.to(connectedRoom).emit("message", {
-                timestamp: payload.timestamp ?? Date.now(),
-                content: String(message),
-                user: {
-                    username: socket.userData.username,
-                    fullName: socket.userData.fullName,
-                    avatar: socket.userData.avatar,
-                },
-            })
-        }
-    }
-
-    initializeSocketIO = () => {
-        this.io.use(async (socket, next) => {
-            try {
-                const token = socket.handshake.auth.token
-
-                if (!token) {
-                    return next(new Error(`auth:token_missing`))
-                }
-
-                const session = await mainAPI.post("/session/validate", {
-                    session: token
-                })
-                    .then((res) => {
-                        return res.data
-                    })
-                    .catch((err) => {
-                        console.error(`[${socket.id}] failed to validate session caused by server error`, err)
-
-                        return false
-                    })
-
-                if (!session) {
-                    return next(new Error(`auth:server_error`))
-                }
-
-                if (!session.valid) {
-                    console.error(`[${socket.id}] failed to validate session caused by invalid token`, session)
-
-                    return next(new Error(`auth:token_invalid`))
-                }
-
-                if (!session.user_id) {
-                    console.error(`[${socket.id}] failed to validate session caused by invalid session. (missing user_id)`, session)
-
-                    return next(new Error(`auth:invalid_session`))
-                }
-
-                const userData = await mainAPI.get(`/user/${session.user_id}/data`)
-                    .then((res) => {
-                        return res.data
-                    })
-                    .catch((err) => {
-                        console.log(err)
-                        return null
-                    })
-
-                if (!userData) {
-                    return next(new Error(`auth:user_failed`))
-                }
-
-                try {
-                    // try to decode the token and get the user's username
-                    const decodedToken = jwt.decode(token)
-
-                    socket.userData = userData
-                    socket.token = token
-                    socket.decodedToken = decodedToken
-                }
-                catch (err) {
-                    return next(new Error(`auth:decode_failed`))
-                }
-
-                console.log(`[${socket.id}] connected`)
-
-                next()
-            } catch (error) {
-                next(new Error(`auth:authentification_failed`))
-            }
-        })
-
-        this.io.on("connection", (socket) => {
-            socket.on("join", (...args) => this.handleClientJoin(socket, ...args))
-
-            socket.on("disconnect", () => {
-                this.handleClientDisconnect(socket)
-            })
-        })
-    }
-
-    async handleClientJoin(socket, payload, cb) {
-        const { room } = payload
-
-        socket.connectedRoom = room
-
-        const pool = await this.attachClientToPool(socket, room).catch((err) => {
-            cb(err)
-            return null
-        })
-
-        if (!pool) return
-
-        console.log(`[${socket.id}] joined room [${room}]`)
-
-        socket.join(room)
-
-        Object.keys(this.roomEventsHandlers).forEach((event) => {
-            socket.on(event, (...args) => this.roomEventsHandlers[event](socket, ...args))
-        })
-
-        const roomConnections = this.connectionPool.filter((client) => client.room === room).length
-
-        cb(null, {
-            roomConnections,
-            limitations: this.limitations,
-        })
-    }
-
-    handleClientDisconnect(socket) {
-        const index = this.connectionPool.findIndex((client) => client.id === socket.id)
-
-        if (index === -1) return
-
-        return this.connectionPool.splice(index, 1)
-    }
-
-    async attachClientToPool(socket, room) {
-        // TODO: check if user can join room or is privated
-
-        if (!room) {
-            throw new Error(`room:invalid`)
-        }
-
-        return this.connectionPool.push({
-            id: socket.id,
-            room,
-            socket
-        })
-    }
-}
-
-export default class Server {
-    constructor(options = {}) {
-        this.app = express()
-        this.httpServer = http.createServer(this.app)
-
-        this.textRoomServer = new TextRoomServer(this.httpServer)
-
-        this.options = {
-            listenHost: process.env.LISTEN_HOST || "0.0.0.0",
-            listenPort: process.env.LISTEN_PORT || 3020,
-            ...options
-        }
-    }
-
-    eventBus = global.eventBus = new EventEmitter()
-
-    async __registerInternalMiddlewares() {
-        let middlewaresPath = fs.readdirSync(path.resolve(__dirname, "useMiddlewares"))
-
-        for await (const middlewarePath of middlewaresPath) {
-            const middleware = require(path.resolve(__dirname, "useMiddlewares", middlewarePath)).default
-
-            if (!middleware) {
-                console.error(`Middleware ${middlewarePath} not found.`)
-
-                continue
-            }
-
-            this.app.use(middleware)
-        }
-    }
-
-    async registerBaseRoute() {
-        await this.app.get("/", async (req, res) => {
-            return res.json({
-                uptimeMinutes: Math.floor(process.uptime() / 60),
-            })
-        })
-    }
-
-    registerRoutes() {
-        routes.forEach((route) => {
-            const order = []
-
-            if (route.middlewares) {
-                route.middlewares.forEach((middleware) => {
-                    order.push(middleware)
-                })
-            }
-
-            order.push(route.routes)
-
-            this.app.use(route.use, ...order)
-        })
-    }
-
-    initialize = async () => {
-        const startHrTime = process.hrtime()
-
-        await this.__registerInternalMiddlewares()
-        this.app.use(express.json({ extended: false }))
-        this.app.use(express.urlencoded({ extended: true }))
-
-        await this.textRoomServer.initializeSocketIO()
-
-        await this.registerBaseRoute()
-        await this.registerRoutes()
-
-        await this.httpServer.listen(this.options.listenPort, this.options.listenHost)
-
-        // calculate elapsed time
-        const elapsedHrTime = process.hrtime(startHrTime)
-        const elapsedTimeInMs = elapsedHrTime[0] * 1000 + elapsedHrTime[1] / 1e6
-
-        // log server started
-        console.log(`🚀 Server started ready on \n\t - http://${this.options.listenHost}:${this.options.listenPort} \n\t - Tooks ${elapsedTimeInMs}ms`)
-    }
-}

packages/message_server/src/middlewares/errorHandler/index.js

@@ -1,5 +0,0 @@
-export const errorHandler = (error, req, res, next) => {
-    res.json({ error: error.message })
-}
-
-export default errorHandler

packages/message_server/src/middlewares/hasPermissions/index.js

@@ -1,30 +0,0 @@
-export const hasPermissions = (req, res, next) => {
-    if (typeof (req.userData) == "undefined") {
-        return res.status(403).json(`User data is not available, please ensure if you are authenticated`)
-    }
-    
-    const { _id, username, roles } = req.userData
-    const { permissions } = req.body
-
-    req.userPermissions = roles
-
-    let check = []
-
-    if (Array.isArray(permissions)) {
-        check = permissions
-    } else {
-        check.push(permissions)
-    }
-
-    if (check.length > 0) {
-        check.forEach((role) => {
-            if (!roles.includes(role)) {
-                return res.status(403).json(`${username} not have permissions ${permissions}`)
-            }
-        })
-    }
-
-    next()
-}
-
-export default hasPermissions

packages/message_server/src/middlewares/index.js

@@ -1,12 +0,0 @@
-// const fileUpload = require("@nanoexpress/middleware-file-upload/cjs")()
-
-export { default as withAuthentication } from "./withAuthentication"
-export { default as withOptionalAuthentication } from "./withOptionalAuthentication"
-
-export { default as errorHandler } from "./errorHandler"
-export { default as hasPermissions } from "./hasPermissions"
-export { default as roles } from "./roles"
-export { default as onlyAdmin } from "./onlyAdmin"
-export { default as permissions } from "./permissions"
-
-// export { fileUpload as fileUpload }

packages/message_server/src/middlewares/onlyAdmin/index.js

@@ -1,7 +0,0 @@
-export default (req, res, next) => {
-    if (!req.user.roles.includes("admin")) {
-        return res.status(403).json({ error: "To make this request it is necessary to have administrator permissions" })
-    }
-
-    next()
-}

packages/message_server/src/middlewares/permissions/index.js

@@ -1,39 +0,0 @@
-import { Config } from "../../models"
-
-export default (req, res, next) => {
-    const requestedPath = `${req.method.toLowerCase()}${req.path.toLowerCase()}`
-
-    Config.findOne({ key: "permissions" }, undefined, {
-        lean: true,
-    }).then(({ value }) => {
-        req.assertedPermissions = []
-
-        const pathRoles = value.pathRoles ?? {}
-
-        if (typeof pathRoles[requestedPath] === "undefined") {
-            console.warn(`[Permissions] No permissions defined for path ${requestedPath}`)
-            return next()
-        }
-
-        const requiredRoles = Array.isArray(pathRoles[requestedPath]) ? pathRoles[requestedPath] : [pathRoles[requestedPath]]
-
-        requiredRoles.forEach((role) => {
-            if (req.user.roles.includes(role)) {
-                req.assertedPermissions.push(role)
-            }
-        })
-
-        if (req.user.roles.includes("admin")) {
-            req.assertedPermissions.push("admin")
-        }
-
-        if (req.assertedPermissions.length === 0 && !req.user.roles.includes("admin")) {
-            return res.status(403).json({
-                error: "forbidden",
-                message: "You don't have permission to access this resource",
-            })
-        }
-
-        next()
-    })
-}

packages/message_server/src/middlewares/roles/index.js

@@ -1,19 +0,0 @@
-export default (req, res, next) => {
-    req.isAdmin = () => {
-        if (req.user.roles.includes("admin")) {
-            return true
-        }
-        
-        return false
-    }
-
-    req.hasRole = (role) => {
-        if (req.user.roles.includes(role)) {
-            return true
-        }
-
-        return false
-    }
-
-    next()
-}

packages/message_server/src/middlewares/withAuthentication/index.js

@@ -1,83 +0,0 @@
-import { Session, User } from "../../models"
-import { Token } from "../../lib"
-import jwt from "jsonwebtoken"
-
-export default (req, res, next) => {
-    function reject(description) {
-        return res.status(403).json({ error: `${description ?? "Invalid session"}` })
-    }
-
-    const authHeader = req.headers?.authorization?.split(" ")
-
-    if (authHeader && authHeader[0] === "Bearer") {
-        const token = authHeader[1]
-        let decoded = null
-
-        try {
-            decoded = jwt.decode(token)
-        } catch (error) {
-            console.error(error)
-        }
-
-        if (!decoded) {
-            return reject("Cannot decode token")
-        }
-
-        jwt.verify(token, global.jwtStrategy.secretOrKey, async (err) => {
-            const sessions = await Session.find({ user_id: decoded.user_id })
-            const currentSession = sessions.find((session) => session.token === token)
-
-            if (!currentSession) {
-                return reject("Cannot find session")
-            }
-
-            const userData = await User.findOne({ _id: currentSession.user_id }).select("+refreshToken")
-
-            if (!userData) {
-                return res.status(404).json({ error: "No user data found" })
-            }
-
-            // if cannot verify token, start regeneration process
-            if (err) {
-                // first check if token is only expired, if is corrupted, reject
-                if (err.name !== "TokenExpiredError") {
-                    return reject("Invalid token, cannot regenerate")
-                }
-
-                let regenerationToken = null
-
-                // check if this expired token has a regeneration token associated
-                const associatedRegenerationToken = await Token.getRegenerationToken(token)
-
-                if (associatedRegenerationToken) {
-                    regenerationToken = associatedRegenerationToken.refreshToken
-                } else {
-                    // create a new regeneration token with the expired token
-                    regenerationToken = await Token.createNewRegenerationToken(token).catch((error) => {
-                        // in case of error, reject
-                        reject(error.message)
-
-                        return null
-                    })
-                }
-
-                if (!regenerationToken) return
-
-                // now send the regeneration token to the client (start Expired Exception Event [EEE])
-                return res.status(401).json({
-                    error: "Token expired",
-                    refreshToken: regenerationToken.refreshToken,
-                })
-            }
-
-            req.user = userData
-            req.jwtToken = token
-            req.decodedToken = decoded
-            req.currentSession = currentSession
-
-            return next()
-        })
-    } else {
-        return reject("Missing token header")
-    }
-}

packages/message_server/src/middlewares/withOptionalAuthentication/index.js

@@ -1,9 +0,0 @@
-import withAuthentication from "../withAuthentication"
-
-export default (req, res, next) => {
-    if (req.headers?.authorization) {
-        withAuthentication(req, res, next)
-    } else {
-        next()
-    }
-}