import config from "config"; import crypto from "crypto"; import redis from "redis"; import CoreClass from "../core"; import utils from "./utils"; class NotificationsModule extends CoreClass { constructor() { super("notifications"); this.subscriptions = []; } initialize() { return new Promise((resolve, reject) => { const url = (this.url = config.get("redis").url); const password = (this.password = config.get("redis").password); this.pub = redis.createClient({ url, password, retry_strategy: options => { if (this.getStatus() === "LOCKDOWN") return; if (this.getStatus() !== "RECONNECTING") this.setStatus("RECONNECTING"); this.log("INFO", `Attempting to reconnect.`); if (options.attempt >= 10) { this.log("ERROR", `Stopped trying to reconnect.`); this.setStatus("FAILED"); // this.failed = true; // this._lockdown(); } } }); this.sub = redis.createClient({ url, password, retry_strategy: options => { if (this.getStatus() === "LOCKDOWN") return; if (this.getStatus() !== "RECONNECTING") this.setStatus("RECONNECTING"); this.log("INFO", `Attempting to reconnect.`); if (options.attempt >= 10) { this.log("ERROR", `Stopped trying to reconnect.`); this.setStatus("FAILED"); // this.failed = true; // this._lockdown(); } } }); this.sub.on("error", err => { if (this.getStatus() === "INITIALIZING") reject(err); if (this.getStatus() === "LOCKDOWN") return; this.log("ERROR", `Error ${err.message}.`); }); this.pub.on("error", err => { if (this.getStatus() === "INITIALIZING") reject(err); if (this.getStatus() === "LOCKDOWN") return; this.log("ERROR", `Error ${err.message}.`); }); this.sub.on("connect", () => { this.log("INFO", "Sub connected succesfully."); if (this.getStatus() === "INITIALIZING") resolve(); else if (this.getStatus() === "LOCKDOWN" || this.getStatus() === "RECONNECTING") this.setStatus("READY"); }); this.pub.on("connect", () => { this.log("INFO", "Pub connected succesfully."); this.pub.config("GET", "notify-keyspace-events", async (err, response) => { if (err) { const formattedErr = await utils.runJob("GET_ERROR", { error: err }); this.log( "ERROR", "NOTIFICATIONS_INITIALIZE", `Getting notify-keyspace-events gave an error. ${formattedErr}` ); this.log( "STATION_ISSUE", `Getting notify-keyspace-events gave an error. ${formattedErr}. ${response}` ); return; } if (response[1] === "xE") { this.log("INFO", "NOTIFICATIONS_INITIALIZE", `notify-keyspace-events is set correctly`); this.log("STATION_ISSUE", `notify-keyspace-events is set correctly`); } else { this.log( "ERROR", "NOTIFICATIONS_INITIALIZE", `notify-keyspace-events is NOT correctly! It is set to: ${response[1]}` ); this.log( "STATION_ISSUE", `notify-keyspace-events is NOT correctly! It is set to: ${response[1]}` ); } }); if (this.getStatus() === "INITIALIZING") resolve(); else if (this.getStatus() === "LOCKDOWN" || this.getStatus() === "RECONNECTING") this.setStatus("INITIALIZED"); }); this.sub.on("pmessage", (pattern, channel, expiredKey) => { this.log( "STATION_ISSUE", `PMESSAGE1 - Pattern: ${pattern}; Channel: ${channel}; ExpiredKey: ${expiredKey}` ); this.subscriptions.forEach(sub => { this.log( "STATION_ISSUE", `PMESSAGE2 - Sub name: ${sub.name}; Calls cb: ${!(sub.name !== expiredKey)}` ); if (sub.name !== expiredKey) return; sub.cb(); }); }); this.sub.psubscribe(`__keyevent@${this.pub.options.db}__:expired`); }); } /** * Schedules a notification to be dispatched in a specific amount of milliseconds, * notifications are unique by name, and the first one is always kept, as in * attempting to schedule a notification that already exists won't do anything * * @param {object} payload - object containing the payload * @param {string} payload.name - the name of the notification we want to schedule * @param {number} payload.time - how long in milliseconds until the notification should be fired * @returns {Promise} - returns a promise (resolve, reject) */ SCHEDULE(payload) { // name, time, cb, station return new Promise((resolve, reject) => { const time = Math.round(payload.time); this.log( "STATION_ISSUE", `SCHEDULE - Time: ${time}; Name: ${payload.name}; Key: ${crypto .createHash("md5") .update(`_notification:${payload.name}_`) .digest("hex")}; StationId: ${payload.station._id}; StationName: ${payload.station.name}` ); this.pub.set( crypto.createHash("md5").update(`_notification:${payload.name}_`).digest("hex"), "", "PX", time, "NX", err => { if (err) reject(err); else resolve(); } ); }); } /** * Subscribes a callback function to be called when a notification gets called * * @param {object} payload - object containing the payload * @param {string} payload.name - the name of the notification we want to subscribe to * @param {boolean} payload.unique - only subscribe if another subscription with the same name doesn't already exist * @returns {Promise} - returns a promise (resolve, reject) */ SUBSCRIBE(payload) { // name, cb, unique = false, station return new Promise(resolve => { this.log( "STATION_ISSUE", `SUBSCRIBE - Name: ${payload.name}; Key: ${crypto .createHash("md5") .update(`_notification:${payload.name}_`) .digest("hex")}, StationId: ${payload.station._id}; StationName: ${payload.station.name}; Unique: ${ payload.unique }; SubscriptionExists: ${!!this.subscriptions.find( subscription => subscription.originalName === payload.name )};` ); if (payload.unique && !!this.subscriptions.find(subscription => subscription.originalName === payload.name)) return resolve({ subscription: this.subscriptions.find(subscription => subscription.originalName === payload.name) }); const subscription = { originalName: payload.name, name: crypto.createHash("md5").update(`_notification:${payload.name}_`).digest("hex"), cb: payload.cb }; this.subscriptions.push(subscription); return resolve({ subscription }); }); } /** * Remove a notification subscription * * @param {object} payload - object containing the payload * @param {object} payload.subscription - the subscription object returned by {@link subscribe} * @returns {Promise} - returns a promise (resolve, reject) */ REMOVE(payload) { // subscription return new Promise(resolve => { const index = this.subscriptions.indexOf(payload.subscription); if (index) this.subscriptions.splice(index, 1); resolve(); }); } UNSCHEDULE(payload) { // name return new Promise((resolve, reject) => { this.log( "STATION_ISSUE", `UNSCHEDULE - Name: ${payload.name}; Key: ${crypto .createHash("md5") .update(`_notification:${payload.name}_`) .digest("hex")}` ); this.pub.del(crypto.createHash("md5").update(`_notification:${payload.name}_`).digest("hex"), err => { if (err) reject(err); else resolve(); }); }); } } export default new NotificationsModule();