Implemented Socket.io

- added socket for chat messages
- added socket for new requests
- added socket for post creation
- added socket for created post

src/lib/InternalEvents.ts

@@ -0,0 +1,8 @@
+export enum InternalEvents {
+    CHATCREATE = "chatCreate",
+    CHATMESSAGE = "chatMessage",
+    GQLCHATMESSAGE = "graphqlChatMessage",
+    REQUESTCREATE = "requestCreate",
+    POSTCREATE = "postCreate",
+    GQLPOSTCREATE = "graphqlPostCreate",
+}

src/lib/dataaccess/ChatMessage.ts

@@ -15,4 +15,17 @@ export class ChatMessage {
     public htmlContent(): string {
         return markdown.renderInline(this.content);
     }
+
+    /**
+     * Returns resolved and rendered content of the chat message.
+     */
+    public resolvedContent() {
+        return {
+            author: this.author.id,
+            chat: this.chat.id,
+            content: this.content,
+            createdAt: this.createdAt,
+            htmlContent: this.htmlContent(),
+        };
+    }
 }

src/lib/dataaccess/Chatroom.ts

@@ -5,8 +5,10 @@ import {User} from "./User";
 
 export class Chatroom {
 
-    constructor(private readonly id: number) {
+    public namespace: string;
+    constructor(public readonly id: number) {
         this.id = Number(id);
+        this.namespace = `/chat/${id}`;
     }
 
     /**

src/lib/dataaccess/Post.ts

@@ -11,6 +11,19 @@ export class Post extends DataObject {
     private $author: number;
     private $type: string;
 
+    /**
+     * Returns the resolved data of the post.
+     */
+    public async resolvedData() {
+        await this.loadDataIfNotExists();
+        return {
+            authorId: this.$author,
+            content: this.$content,
+            createdAt: this.$createdAt,
+            id: this.id,
+            type: this.$type,
+        };
+    }
     /**
      * Returns the upvotes of a post.
      */

src/lib/dataaccess/Request.ts

@@ -8,5 +8,17 @@ export class Request {
     constructor(
         public readonly sender: User,
         public readonly receiver: User,
-        public readonly type: dataaccess.RequestType) {}
+        public readonly type: dataaccess.RequestType) {
+    }
+
+    /**
+     * Returns the resolved request data.
+     */
+    public resolvedData() {
+        return {
+            receiverId: this.receiver.id,
+            senderId: this.sender.id,
+            type: this.type,
+        };
+    }
 }

src/lib/dataaccess/index.ts

@@ -2,6 +2,7 @@ import {Pool} from "pg";
 import {ChatNotFoundError} from "../errors/ChatNotFoundError";
 import {UserNotFoundError} from "../errors/UserNotFoundError";
 import globals from "../globals";
+import {InternalEvents} from "../InternalEvents";
 import {QueryHelper} from "../QueryHelper";
 import {ChatMessage} from "./ChatMessage";
 import {Chatroom} from "./Chatroom";
@@ -125,7 +126,9 @@ namespace dataaccess {
             text: "INSERT INTO posts (content, author, type) VALUES ($1, $2, $3) RETURNING *",
             values: [content, authorId, type],
         });
-        return new Post(result.id, result);
+        const post = new Post(result.id, result);
+        globals.internalEmitter.emit(InternalEvents.POSTCREATE, post);
+        return post;
     }
 
     /**
@@ -167,7 +170,9 @@ namespace dataaccess {
         } finally {
             transaction.release();
         }
-        return new Chatroom(id);
+        const chat = new Chatroom(id);
+        globals.internalEmitter.emit(InternalEvents.CHATCREATE, chat);
+        return chat;
     }
 
     /**
@@ -183,12 +188,28 @@ namespace dataaccess {
                 text: "INSERT INTO chat_messages (chat, author, content) values ($1, $2, $3) RETURNING *",
                 values: [chatId, authorId, content],
             });
-            return new ChatMessage(new User(result.author), chat, result.created_at, result.content);
+            const message = new ChatMessage(new User(result.author), chat, result.created_at, result.content);
+            globals.internalEmitter.emit(InternalEvents.CHATMESSAGE, message);
+            return message;
         } else {
             throw new ChatNotFoundError(chatId);
         }
     }
 
+    /**
+     * Returns all chats.
+     */
+    export async function getAllChats(): Promise<Chatroom[]> {
+        const result = await queryHelper.all({
+            text: "SELECT id FROM chats;",
+        });
+        const chats = [];
+        for (const row of result) {
+            chats.push(new Chatroom(row.id));
+        }
+        return chats;
+    }
+
     /**
      * Sends a request to a user.
      * @param sender
@@ -202,7 +223,9 @@ namespace dataaccess {
             text: "INSERT INTO requests (sender, receiver, type) VALUES ($1, $2, $3) RETURNING *",
             values: [sender, receiver, type],
         });
-        return new Request(new User(result.sender), new User(result.receiver), result.type);
+        const request = new Request(new User(result.sender), new User(result.receiver), result.type);
+        globals.internalEmitter.emit(InternalEvents.REQUESTCREATE, Request);
+        return request;
     }
 
     /**

src/lib/globals.ts

@@ -5,6 +5,7 @@
  * Partly taken from {@link https://github.com/Trivernis/whooshy}
  */
 
+import {EventEmitter} from "events";
 import * as fsx from "fs-extra";
 import * as yaml from "js-yaml";
 import * as winston from "winston";
@@ -42,6 +43,7 @@ namespace globals {
             }),
         ],
     });
+    export const internalEmitter = new EventEmitter();
     cache.on("set", (key) => logger.debug(`Caching '${key}'.`));
     cache.on("miss", (key) => logger.debug(`Cache miss for '${key}'`));
     cache.on("hit", (key) => logger.debug(`Cache hit for '${key}'`));

src/public/graphql/schema.graphql

@@ -189,6 +189,9 @@ type Request {
 
 "represents a chatroom"
 type ChatRoom {
+    "the socket.io namespace for the chatroom"
+    namespace: String
+
     "the members of the chatroom"
     members: [User!]
 

src/routes/home.ts

@@ -1,17 +1,22 @@
 import {Router} from "express";
 import {GraphQLError} from "graphql";
 import * as status from "http-status";
-import {Server} from "socket.io";
+import {Namespace, Server} from "socket.io";
 import dataaccess from "../lib/dataaccess";
+import {ChatMessage} from "../lib/dataaccess/ChatMessage";
 import {Chatroom} from "../lib/dataaccess/Chatroom";
 import {Post} from "../lib/dataaccess/Post";
 import {Profile} from "../lib/dataaccess/Profile";
+import {Request} from "../lib/dataaccess/Request";
 import {User} from "../lib/dataaccess/User";
 import {NotLoggedInGqlError} from "../lib/errors/graphqlErrors";
 import globals from "../lib/globals";
+import {InternalEvents} from "../lib/InternalEvents";
 import {is} from "../lib/regex";
 import Route from "../lib/Route";
 
+const chatRooms: Namespace[] = [];
+
 /**
  * Class for the home route.
  */
@@ -30,6 +35,33 @@ class HomeRoute extends Route {
      */
     public async init(io: Server) {
         this.io = io;
+
+        io.on("connection", (socket) => {
+            socket.on("postCreate", async (content) => {
+                if (socket.handshake.session.userId) {
+                    const post = await dataaccess.createPost(content, socket.handshake.session.userId);
+                    io.emit("post", await post.resolvedData());
+                } else {
+                    socket.emit("error", "Not logged in!");
+                }
+            });
+            globals.internalEmitter.on(InternalEvents.REQUESTCREATE, (request: Request) => {
+                if (request.receiver.id === socket.handshake.session.userId) {
+                    socket.emit("request", request.resolvedData());
+                }
+            });
+            globals.internalEmitter.on(InternalEvents.GQLPOSTCREATE, async (post: Post) => {
+                socket.emit("post", await post.resolvedData());
+            });
+        });
+
+        const chats = await dataaccess.getAllChats();
+        for (const chat of chats) {
+            chatRooms[chat.id] = this.getChatSocketNamespace(chat.id);
+        }
+        globals.internalEmitter.on(InternalEvents.CHATCREATE, (chat: Chatroom) => {
+            chatRooms[chat.id] = this.getChatSocketNamespace(chat.id);
+        });
     }
 
     /**
@@ -145,7 +177,9 @@ class HomeRoute extends Route {
             async createPost({content}: { content: string }) {
                 if (content) {
                     if (req.session.userId) {
-                        return await dataaccess.createPost(content, req.session.userId);
+                        const post = await dataaccess.createPost(content, req.session.userId);
+                        globals.internalEmitter.emit(InternalEvents.GQLPOSTCREATE, post);
+                        return post;
                     } else {
                         res.status(status.UNAUTHORIZED);
                         return new NotLoggedInGqlError();
@@ -175,7 +209,6 @@ class HomeRoute extends Route {
                         chatMembers.push(...members);
                     }
                     return await dataaccess.createChat(...chatMembers);
-
                 } else {
                     res.status(status.UNAUTHORIZED);
                     return new NotLoggedInGqlError();
@@ -188,7 +221,9 @@ class HomeRoute extends Route {
                 }
                 if (chatId && content) {
                     try {
-                        return await dataaccess.sendChatMessage(req.session.userId, chatId, content);
+                        const message = await dataaccess.sendChatMessage(req.session.userId, chatId, content);
+                        globals.internalEmitter.emit(InternalEvents.GQLCHATMESSAGE, message);
+                        return message;
                     } catch (err) {
                         res.status(status.BAD_REQUEST);
                         return err.graphqlError;
@@ -245,6 +280,35 @@ class HomeRoute extends Route {
             },
         };
     }
+
+    /**
+     * Returns the namespace socket for a chat socket.
+     * @param chatId
+     */
+    private getChatSocketNamespace(chatId: number) {
+        if (chatRooms[chatId]) {
+            return chatRooms[chatId];
+        }
+        const chatNs = this.io.of(`/chat/${chatId}`);
+        chatNs.on("connection", (socket) => {
+            socket.on("chatMessage", async (content) => {
+                if (socket.handshake.session.userId) {
+                    const userId = socket.handshake.session.userId;
+                    const message = await dataaccess.sendChatMessage(userId, chatId, content);
+                    socket.broadcast.emit("chatMessage", message.resolvedContent());
+                    socket.emit("chatMessageSent", message.resolvedContent());
+                } else {
+                    socket.emit("error", "Not logged in!");
+                }
+            });
+            globals.internalEmitter.on(InternalEvents.GQLCHATMESSAGE, (message: ChatMessage) => {
+                if (message.chat.id === chatId) {
+                    socket.emit("chatMessage", message.resolvedContent());
+                }
+            });
+        });
+        return chatNs;
+    }
 }
 
 export default HomeRoute;