Fire channel subscriptions on gateway server to Redis

package-lock.json

@@ -45,6 +45,15 @@
             "integrity": "sha512-L8ogeT6vDzT1vxlW3KITTCt+BVXXVkLXfZ/XNm6UqbcJgxf+KPO7yjWx7dQQE8RW07KopL10x2gNMs41+IkMGQ==",
             "dev": true
         },
+        "@types/redis": {
+            "version": "2.8.14",
+            "resolved": "https://registry.npmjs.org/@types/redis/-/redis-2.8.14.tgz",
+            "integrity": "sha512-255dzsOLJdXFHBio9/aMHGozNkoiBUgc+g2nlNjbTSp5qcAlmpm4Z6Xs3pKOBLNIKdZbA2BkUxWvYSIwKra0Yw==",
+            "dev": true,
+            "requires": {
+                "@types/node": "*"
+            }
+        },
         "@types/request": {
             "version": "2.48.4",
             "resolved": "https://registry.npmjs.org/@types/request/-/request-2.48.4.tgz",
@@ -286,6 +295,11 @@
                 "kuler": "1.0.x"
             }
         },
+        "double-ended-queue": {
+            "version": "2.1.0-0",
+            "resolved": "https://registry.npmjs.org/double-ended-queue/-/double-ended-queue-2.1.0-0.tgz",
+            "integrity": "sha1-ED01J/0xUo9AGIEwyEHv3XgmTlw="
+        },
         "ecc-jsbn": {
             "version": "0.1.2",
             "resolved": "https://registry.npmjs.org/ecc-jsbn/-/ecc-jsbn-0.1.2.tgz",
@@ -649,6 +663,26 @@
                 "util-deprecate": "^1.0.1"
             }
         },
+        "redis": {
+            "version": "2.8.0",
+            "resolved": "https://registry.npmjs.org/redis/-/redis-2.8.0.tgz",
+            "integrity": "sha512-M1OkonEQwtRmZv4tEWF2VgpG0JWJ8Fv1PhlgT5+B+uNq2cA3Rt1Yt/ryoR+vQNOQcIEgdCdfH0jr3bDpihAw1A==",
+            "requires": {
+                "double-ended-queue": "^2.1.0-0",
+                "redis-commands": "^1.2.0",
+                "redis-parser": "^2.6.0"
+            }
+        },
+        "redis-commands": {
+            "version": "1.5.0",
+            "resolved": "https://registry.npmjs.org/redis-commands/-/redis-commands-1.5.0.tgz",
+            "integrity": "sha512-6KxamqpZ468MeQC3bkWmCB1fp56XL64D4Kf0zJSwDZbVLLm7KFkoIcHrgRvQ+sk8dnhySs7+yBg94yIkAK7aJg=="
+        },
+        "redis-parser": {
+            "version": "2.6.0",
+            "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-2.6.0.tgz",
+            "integrity": "sha1-Uu0J2srBCPGmMcB+m2mUHnoZUEs="
+        },
         "request": {
             "version": "2.88.0",
             "resolved": "https://registry.npmjs.org/request/-/request-2.88.0.tgz",

package.json

@@ -5,17 +5,19 @@
         "@types/node": "^13.1.2",
         "@types/pg": "^7.14.1",
         "@types/pg-pool": "^2.0.1",
-        "@types/winston": "^2.4.4",
-        "@types/ws": "^6.0.4",
+        "@types/redis": "^2.8.14",
         "@types/request": "^2.48.4",
         "@types/request-promise": "^4.1.45",
         "@types/request-promise-native": "^1.0.17",
+        "@types/winston": "^2.4.4",
+        "@types/ws": "^6.0.4",
         "cross-env": "^5.1"
     },
     "dependencies": {
         "pg": "^7.17.0",
         "pg-cursor": "^2.1.2",
         "pg-pool": "^2.0.9",
+        "redis": "^2.8.0",
         "request": "^2.88.0",
         "request-promise-native": "^1.0.8",
         "winston": "^3.2.1",

socket_server/classes/DiscordRedis.ts

@@ -0,0 +1,41 @@
+
+import DiscordWebSocketData from "./DiscordWebSocketData";
+import logger from "../modules/logger";
+
+import {RedisClient} from "redis";
+
+export default class DiscordRedis {
+    client: RedisClient;
+    websocket: DiscordWebSocketData;
+
+    constructor(discord_ws: DiscordWebSocketData) {
+        this.websocket = discord_ws;
+    }
+
+    connect(){
+        logger.info('Connecting to Redis');
+        this.client = new RedisClient({});
+
+        // Setup events
+        this.client.on('message', (channel, msg) => {
+            this.handle_subscribed_events(channel, msg);
+        })
+    }
+
+    close(): void {
+        this.client.quit();
+    }
+
+    subscribe(channel: string[]): Promise<string> {
+        return new Promise(((resolve, reject) => {
+            this.client.subscribe(channel, (err, msg) => {
+                if(err){  return reject(err) }
+                return resolve(msg);
+            })
+        }));
+    }
+
+    handle_subscribed_events(channel: string, msg: string){
+        logger.info('Received event from Redis on channel %s: %s', channel, msg)
+    }
+}

socket_server/classes/DiscordWebSocketData.ts

@@ -9,6 +9,7 @@ import Payload from "./Payload";
 import {IdentityPayloadType, ReadyEventType, RequestMemberPayloadType} from "../interfaces/payload_types";
 
 import * as Cursor from 'pg-cursor';
+import DiscordRedis from "./DiscordRedis";
 
 export default class DiscordWebSocketData {
     static sql_get_guild_as_unavailable = `
@@ -56,11 +57,13 @@ export default class DiscordWebSocketData {
     userid: number = null;
     ident_data?: IdentityPayloadType;
     participates_in_guild: number[] = [];
+    redis_connection: DiscordRedis;
 
     heartbeat_timer: Timer;
 
     constructor(ws: WebSocket) {
         this.ws = ws;
+        this.redis_connection = new DiscordRedis(this);
     }
 
     send(payload: Payload){
@@ -107,6 +110,8 @@ export default class DiscordWebSocketData {
         logger.info('Closing websocket: %d - %s', code, reason);
         logger.info('Clearing timeout');
         clearTimeout(this.heartbeat_timer);
+        logger.info('Closing redis connection');
+        this.redis_connection.close();
     }
 
     /*
@@ -185,6 +190,7 @@ export default class DiscordWebSocketData {
         this.userid = user_id['id'];
         this.authenticated = true;
         this.ident_data = data;
+        await this.redis_connection.connect();
 
         logger.info(`User ID ${this.userid} has logged in`);
     }
@@ -297,6 +303,10 @@ export default class DiscordWebSocketData {
 
         logger.info(`Sending GUILD_CREATE for guild ${guild.id}`);
         this.send(payload);
+
+        this.redis_connection.subscribe([`dapi.private-guild.${guild.id}`]);
+        let subscribe_channel_names = channels.rows.filter(row => row.type == 0).map(row => `dapi.private-channel.${row.id}`);
+        await this.redis_connection.subscribe(subscribe_channel_names);
     }
 
     /*