index.js 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285
  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) return reject(new Error("Invalid key!"));
  114. if (!payload.table) return reject(new Error("Invalid table!"));
  115. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  116. return CacheModule.client.hget(payload.table, key, (err, value) => {
  117. if (err) return reject(new Error(err));
  118. try {
  119. value = JSON.parse(value);
  120. } catch (e) {
  121. return reject(err);
  122. }
  123. return resolve(value);
  124. });
  125. });
  126. }
  127. /**
  128. * Deletes a single value from a table
  129. *
  130. * @param {object} payload - object containing payload
  131. * @param {string} payload.table - name of the table to delete the value from (table === redis hash)
  132. * @param {string} payload.key - name of the key to delete
  133. * @returns {Promise} - returns a promise (resolve, reject)
  134. */
  135. HDEL(payload) {
  136. return new Promise((resolve, reject) => {
  137. let { key } = payload;
  138. if (!payload.table) return reject(new Error("Invalid table!"));
  139. if (!key) return reject(new Error("Invalid key!"));
  140. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  141. return CacheModule.client.hdel(payload.table, key, err => {
  142. if (err) return reject(new Error(err));
  143. return resolve();
  144. });
  145. });
  146. }
  147. /**
  148. * Returns all the keys for a table
  149. *
  150. * @param {object} payload - object containing payload
  151. * @param {string} payload.table - name of the table to get the values from (table === redis hash)
  152. * @param {boolean} [payload.parseJson=true] - attempts to parse all values as JSON by default
  153. * @returns {Promise} - returns a promise (resolve, reject)
  154. */
  155. HGETALL(payload) {
  156. return new Promise((resolve, reject) => {
  157. if (!payload.table) return reject(new Error("Invalid table!"));
  158. return CacheModule.client.hgetall(payload.table, (err, obj) => {
  159. if (err) return reject(new Error(err));
  160. if (obj)
  161. Object.keys(obj).forEach(key => {
  162. obj[key] = JSON.parse(obj[key]);
  163. });
  164. else if (!obj) obj = [];
  165. return resolve(obj);
  166. });
  167. });
  168. }
  169. /**
  170. * Publish a message to a channel, caches the redis client connection
  171. *
  172. * @param {object} payload - object containing payload
  173. * @param {string} payload.channel - the name of the channel we want to publish a message to
  174. * @param {*} payload.value - the value we want to send
  175. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  176. * @returns {Promise} - returns a promise (resolve, reject)
  177. */
  178. PUB(payload) {
  179. return new Promise((resolve, reject) => {
  180. let { value } = payload;
  181. if (!payload.channel) return reject(new Error("Invalid channel!"));
  182. if (!value) return reject(new Error("Invalid value!"));
  183. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  184. return CacheModule.client.publish(payload.channel, value, err => {
  185. if (err) reject(err);
  186. else resolve();
  187. });
  188. });
  189. }
  190. /**
  191. * Subscribe to a channel, caches the redis client connection
  192. *
  193. * @param {object} payload - object containing payload
  194. * @param {string} payload.channel - name of the channel to subscribe to
  195. * @param {boolean} [payload.parseJson=true] - parse the message as JSON
  196. * @returns {Promise} - returns a promise (resolve, reject)
  197. */
  198. SUB(payload) {
  199. return new Promise((resolve, reject) => {
  200. if (!payload.channel) return reject(new Error("Invalid channel!"));
  201. if (subs[payload.channel] === undefined) {
  202. subs[payload.channel] = {
  203. client: redis.createClient({
  204. url: CacheModule.url,
  205. password: CacheModule.password
  206. }),
  207. cbs: []
  208. };
  209. subs[payload.channel].client.on("message", (channel, message) => {
  210. if (message.startsWith("[") || message.startsWith("{"))
  211. try {
  212. message = JSON.parse(message);
  213. } catch (err) {
  214. console.error(err);
  215. }
  216. else if (message.startsWith('"') && message.endsWith('"'))
  217. message = message.substring(1).substring(0, message.length - 2);
  218. return subs[channel].cbs.forEach(cb => cb(message));
  219. });
  220. subs[payload.channel].client.subscribe(payload.channel);
  221. }
  222. subs[payload.channel].cbs.push(payload.cb);
  223. return resolve();
  224. });
  225. }
  226. /**
  227. * Returns a redis schema
  228. *
  229. * @param {object} payload - object containing the payload
  230. * @param {string} payload.schemaName - the name of the schema to get
  231. * @returns {Promise} - returns promise (reject, resolve)
  232. */
  233. GET_SCHEMA(payload) {
  234. return new Promise(resolve => {
  235. resolve(CacheModule.schemas[payload.schemaName]);
  236. });
  237. }
  238. }
  239. export default new _CacheModule();