index.js 9.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347
  1. import config from "config";
  2. import redis from "redis";
  3. import mongoose from "mongoose";
  4. import CoreClass from "../../core";
  5. // Lightweight / convenience wrapper around redis module for our needs
  6. const pubs = {};
  7. const subs = {};
  8. let CacheModule;
  9. class _CacheModule extends CoreClass {
  10. // eslint-disable-next-line require-jsdoc
  11. constructor() {
  12. super("cache");
  13. CacheModule = this;
  14. }
  15. /**
  16. * Initialises the cache/redis module
  17. *
  18. * @returns {Promise} - returns promise (reject, resolve)
  19. */
  20. async initialize() {
  21. const importSchema = schemaName =>
  22. new Promise(resolve => {
  23. import(`./schemas/${schemaName}`).then(schema => resolve(schema.default));
  24. });
  25. this.schemas = {
  26. session: await importSchema("session"),
  27. station: await importSchema("station"),
  28. playlist: await importSchema("playlist"),
  29. officialPlaylist: await importSchema("officialPlaylist"),
  30. song: await importSchema("song"),
  31. punishment: await importSchema("punishment"),
  32. recentActivity: await importSchema("recentActivity")
  33. };
  34. return new Promise((resolve, reject) => {
  35. this.url = config.get("redis").url;
  36. this.password = config.get("redis").password;
  37. this.log("INFO", "Connecting...");
  38. this.client = redis.createClient({
  39. url: this.url,
  40. password: this.password,
  41. retry_strategy: options => {
  42. if (this.getStatus() === "LOCKDOWN") return;
  43. if (this.getStatus() !== "RECONNECTING") this.setStatus("RECONNECTING");
  44. this.log("INFO", `Attempting to reconnect.`);
  45. if (options.attempt >= 10) {
  46. this.log("ERROR", `Stopped trying to reconnect.`);
  47. this.setStatus("FAILED");
  48. }
  49. }
  50. });
  51. this.client.on("error", err => {
  52. if (this.getStatus() === "INITIALIZING") reject(err);
  53. if (this.getStatus() === "LOCKDOWN") return;
  54. this.log("ERROR", `Error ${err.message}.`);
  55. });
  56. this.client.on("connect", () => {
  57. this.log("INFO", "Connected succesfully.");
  58. if (this.getStatus() === "INITIALIZING") resolve();
  59. else if (this.getStatus() === "FAILED" || this.getStatus() === "RECONNECTING") this.setStatus("READY");
  60. });
  61. });
  62. }
  63. /**
  64. * Quits redis client
  65. *
  66. * @returns {Promise} - returns promise (reject, resolve)
  67. */
  68. QUIT() {
  69. return new Promise(resolve => {
  70. if (CacheModule.client.connected) {
  71. CacheModule.client.quit();
  72. Object.keys(pubs).forEach(channel => pubs[channel].quit());
  73. Object.keys(subs).forEach(channel => subs[channel].client.quit());
  74. }
  75. resolve();
  76. });
  77. }
  78. /**
  79. * Sets a single value in a table
  80. *
  81. * @param {object} payload - object containing payload
  82. * @param {string} payload.table - name of the table we want to set a key of (table === redis hash)
  83. * @param {string} payload.key - name of the key to set
  84. * @param {*} payload.value - the value we want to set
  85. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  86. * @returns {Promise} - returns a promise (resolve, reject)
  87. */
  88. HSET(payload) {
  89. return new Promise((resolve, reject) => {
  90. let { key } = payload;
  91. let { value } = payload;
  92. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  93. // automatically stringify objects and arrays into JSON
  94. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  95. CacheModule.client.hset(payload.table, key, value, err => {
  96. if (err) return reject(new Error(err));
  97. return resolve(JSON.parse(value));
  98. });
  99. });
  100. }
  101. /**
  102. * Gets a single value from a table
  103. *
  104. * @param {object} payload - object containing payload
  105. * @param {string} payload.table - name of the table to get the value from (table === redis hash)
  106. * @param {string} payload.key - name of the key to fetch
  107. * @param {boolean} [payload.parseJson=true] - attempt to parse returned data as JSON
  108. * @returns {Promise} - returns a promise (resolve, reject)
  109. */
  110. HGET(payload) {
  111. return new Promise((resolve, reject) => {
  112. let { key } = payload;
  113. if (!key) {
  114. reject(new Error("Invalid key!"));
  115. return;
  116. }
  117. if (!payload.table) {
  118. reject(new Error("Invalid table!"));
  119. return;
  120. }
  121. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  122. CacheModule.client.hget(payload.table, key, (err, value) => {
  123. if (err) {
  124. reject(new Error(err));
  125. return;
  126. }
  127. try {
  128. value = JSON.parse(value);
  129. } catch (e) {
  130. reject(err);
  131. return;
  132. }
  133. resolve(value);
  134. });
  135. });
  136. }
  137. /**
  138. * Deletes a single value from a table
  139. *
  140. * @param {object} payload - object containing payload
  141. * @param {string} payload.table - name of the table to delete the value from (table === redis hash)
  142. * @param {string} payload.key - name of the key to delete
  143. * @returns {Promise} - returns a promise (resolve, reject)
  144. */
  145. HDEL(payload) {
  146. return new Promise((resolve, reject) => {
  147. let { key } = payload;
  148. if (!payload.table) {
  149. reject(new Error("Invalid table!"));
  150. return;
  151. }
  152. if (!key) {
  153. reject(new Error("Invalid key!"));
  154. return;
  155. }
  156. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  157. CacheModule.client.hdel(payload.table, key, err => {
  158. if (err) {
  159. reject(new Error(err));
  160. return;
  161. }
  162. resolve();
  163. });
  164. });
  165. }
  166. /**
  167. * Returns all the keys for a table
  168. *
  169. * @param {object} payload - object containing payload
  170. * @param {string} payload.table - name of the table to get the values from (table === redis hash)
  171. * @param {boolean} [payload.parseJson=true] - attempts to parse all values as JSON by default
  172. * @returns {Promise} - returns a promise (resolve, reject)
  173. */
  174. HGETALL(payload) {
  175. return new Promise((resolve, reject) => {
  176. if (!payload.table) {
  177. reject(new Error("Invalid table!"));
  178. return;
  179. }
  180. CacheModule.client.hgetall(payload.table, (err, obj) => {
  181. if (err) {
  182. reject(new Error(err));
  183. return;
  184. }
  185. if (obj)
  186. Object.keys(obj).forEach(key => {
  187. obj[key] = JSON.parse(obj[key]);
  188. });
  189. else if (!obj) obj = [];
  190. resolve(obj);
  191. });
  192. });
  193. }
  194. /**
  195. * Deletes a single value
  196. *
  197. * @param {object} payload - object containing payload
  198. * @param {string} payload.key - name of the key to delete
  199. * @returns {Promise} - returns a promise (resolve, reject)
  200. */
  201. DEL(payload) {
  202. return new Promise((resolve, reject) => {
  203. let { key } = payload;
  204. if (!key) {
  205. reject(new Error("Invalid key!"));
  206. return;
  207. }
  208. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  209. CacheModule.client.del(key, err => {
  210. if (err) {
  211. reject(new Error(err));
  212. return;
  213. }
  214. resolve();
  215. });
  216. });
  217. }
  218. /**
  219. * Publish a message to a channel, caches the redis client connection
  220. *
  221. * @param {object} payload - object containing payload
  222. * @param {string} payload.channel - the name of the channel we want to publish a message to
  223. * @param {*} payload.value - the value we want to send
  224. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  225. * @returns {Promise} - returns a promise (resolve, reject)
  226. */
  227. PUB(payload) {
  228. return new Promise((resolve, reject) => {
  229. let { value } = payload;
  230. if (!payload.channel) {
  231. reject(new Error("Invalid channel!"));
  232. return;
  233. }
  234. if (!value) {
  235. reject(new Error("Invalid value!"));
  236. return;
  237. }
  238. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  239. CacheModule.client.publish(payload.channel, value, err => {
  240. if (err) reject(err);
  241. else resolve();
  242. });
  243. });
  244. }
  245. /**
  246. * Subscribe to a channel, caches the redis client connection
  247. *
  248. * @param {object} payload - object containing payload
  249. * @param {string} payload.channel - name of the channel to subscribe to
  250. * @param {boolean} [payload.parseJson=true] - parse the message as JSON
  251. * @returns {Promise} - returns a promise (resolve, reject)
  252. */
  253. SUB(payload) {
  254. return new Promise((resolve, reject) => {
  255. if (!payload.channel) {
  256. reject(new Error("Invalid channel!"));
  257. return;
  258. }
  259. if (subs[payload.channel] === undefined) {
  260. subs[payload.channel] = {
  261. client: redis.createClient({
  262. url: CacheModule.url,
  263. password: CacheModule.password
  264. }),
  265. cbs: []
  266. };
  267. subs[payload.channel].client.on("message", (channel, message) => {
  268. if (message.startsWith("[") || message.startsWith("{"))
  269. try {
  270. message = JSON.parse(message);
  271. } catch (err) {
  272. console.error(err);
  273. }
  274. else if (message.startsWith('"') && message.endsWith('"'))
  275. message = message.substring(1).substring(0, message.length - 2);
  276. return subs[channel].cbs.forEach(cb => cb(message));
  277. });
  278. subs[payload.channel].client.subscribe(payload.channel);
  279. }
  280. subs[payload.channel].cbs.push(payload.cb);
  281. resolve();
  282. });
  283. }
  284. /**
  285. * Returns a redis schema
  286. *
  287. * @param {object} payload - object containing the payload
  288. * @param {string} payload.schemaName - the name of the schema to get
  289. * @returns {Promise} - returns promise (reject, resolve)
  290. */
  291. GET_SCHEMA(payload) {
  292. return new Promise(resolve => {
  293. resolve(CacheModule.schemas[payload.schemaName]);
  294. });
  295. }
  296. }
  297. export default new _CacheModule();