MusareFork/backend/logic/notifications.js

'use strict';

const coreClass = require("../core");

const crypto = require('crypto');
const redis = require('redis');
const config = require('config');

const subscriptions = [];

module.exports = class extends coreClass {
	initialize() {
		return new Promise((resolve, reject) => {
			this.setStage(1);

			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.state === "LOCKDOWN") return;
					if (this.state !== "RECONNECTING") this.setState("RECONNECTING");

					this.logger.info("NOTIFICATIONS_MODULE", `Attempting to reconnect pub.`);

					if (options.attempt >= 10) {
						this.logger.error("NOTIFICATIONS_MODULE", `Stopped trying to reconnect pub.`);

						this.failed = true;
						this._lockdown();

						return undefined;
					}

					return 3000;
				}
			});
			this.sub = redis.createClient({
				url,
				password,
				retry_strategy: (options) => {
					if (this.state === "LOCKDOWN") return;
					if (this.state !== "RECONNECTING") this.setState("RECONNECTING");

					this.logger.info("NOTIFICATIONS_MODULE", `Attempting to reconnect sub.`);

					if (options.attempt >= 10) {
						this.logger.error("NOTIFICATIONS_MODULE", `Stopped trying to reconnect sub.`);

						this.failed = true;
						this._lockdown();

						return undefined;
					}

					return 3000;
				}
			});

			this.sub.on('error', (err) => {
				if (this.state === "INITIALIZING") reject(err);
				if(this.state === "LOCKDOWN") return;

				this.logger.error("NOTIFICATIONS_MODULE", `Sub error ${err.message}.`);
			});

			this.pub.on('error', (err) => {
				if (this.state === "INITIALIZING") reject(err);
				if(this.state === "LOCKDOWN") return; 

				this.logger.error("NOTIFICATIONS_MODULE", `Pub error ${err.message}.`);
			});

			this.sub.on("connect", () => {
				this.logger.info("NOTIFICATIONS_MODULE", "Sub connected succesfully.");

				if (this.state === "INITIALIZING") resolve();
				else if (this.state === "LOCKDOWN" || this.state === "RECONNECTING") this.setState("INITIALIZED");
				
			});

			this.pub.on("connect", () => {
				this.logger.info("NOTIFICATIONS_MODULE", "Pub connected succesfully.");

				if (this.state === "INITIALIZING") resolve();
				else if (this.state === "LOCKDOWN" || this.state === "RECONNECTING") this.setState("INITIALIZED");
			});

			this.sub.on('pmessage', (pattern, channel, expiredKey) => {
				this.logger.stationIssue(`PMESSAGE1 - Pattern: ${pattern}; Channel: ${channel}; ExpiredKey: ${expiredKey}`);
				subscriptions.forEach((sub) => {
					this.logger.stationIssue(`PMESSAGE2 - Sub name: ${sub.name}; Calls cb: ${!(sub.name !== expiredKey)}`);
					if (sub.name !== expiredKey) return;
					sub.cb();
				});
			});

			this.sub.psubscribe('__keyevent@0__: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 {String} name - the name of the notification we want to schedule
	 * @param {Integer} time - how long in milliseconds until the notification should be fired
	 * @param {Function} cb - gets called when the notification has been scheduled
	 */
	async schedule(name, time, cb, station) {
		try { await this._validateHook(); } catch { return; }

		if (!cb) cb = ()=>{};

		time = Math.round(time);
		this.logger.stationIssue(`SCHEDULE - Time: ${time}; Name: ${name}; Key: ${crypto.createHash('md5').update(`_notification:${name}_`).digest('hex')}; StationId: ${station._id}; StationName: ${station.name}`);
		this.pub.set(crypto.createHash('md5').update(`_notification:${name}_`).digest('hex'), '', 'PX', time, 'NX', cb);
	}

	/**
	 * Subscribes a callback function to be called when a notification gets called
	 *
	 * @param {String} name - the name of the notification we want to subscribe to
	 * @param {Function} cb - gets called when the subscribed notification gets called
	 * @param {Boolean} unique - only subscribe if another subscription with the same name doesn't already exist
	 * @return {Object} - the subscription object
	 */
	async subscribe(name, cb, unique = false, station) {
		try { await this._validateHook(); } catch { return; }

		this.logger.stationIssue(`SUBSCRIBE - Name: ${name}; Key: ${crypto.createHash('md5').update(`_notification:${name}_`).digest('hex')}, StationId: ${station._id}; StationName: ${station.name}; Unique: ${unique}; SubscriptionExists: ${!!subscriptions.find((subscription) => subscription.originalName == name)};`);
		if (unique && !!subscriptions.find((subscription) => subscription.originalName == name)) return;
		let subscription = { originalName: name, name: crypto.createHash('md5').update(`_notification:${name}_`).digest('hex'), cb };
		subscriptions.push(subscription);
		return subscription;
	}

	/**
	 * Remove a notification subscription
	 *
	 * @param {Object} subscription - the subscription object returned by {@link subscribe}
	 */
	async remove(subscription) {
		try { await this._validateHook(); } catch { return; }

		let index = subscriptions.indexOf(subscription);
		if (index) subscriptions.splice(index, 1);
	}

	async unschedule(name) {
		try { await this._validateHook(); } catch { return; }

		this.logger.stationIssue(`UNSCHEDULE - Name: ${name}; Key: ${crypto.createHash('md5').update(`_notification:${name}_`).digest('hex')}`);
		this.pub.del(crypto.createHash('md5').update(`_notification:${name}_`).digest('hex'));
	}
}