index.js 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290
  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. };
  33. return new Promise((resolve, reject) => {
  34. this.url = config.get("redis").url;
  35. this.password = config.get("redis").password;
  36. this.log("INFO", "Connecting...");
  37. this.client = redis.createClient({
  38. url: this.url,
  39. password: this.password,
  40. retry_strategy: options => {
  41. if (this.getStatus() === "LOCKDOWN") return;
  42. if (this.getStatus() !== "RECONNECTING") this.setStatus("RECONNECTING");
  43. this.log("INFO", `Attempting to reconnect.`);
  44. if (options.attempt >= 10) {
  45. this.log("ERROR", `Stopped trying to reconnect.`);
  46. this.setStatus("FAILED");
  47. // this.failed = true;
  48. // this._lockdown();
  49. }
  50. }
  51. });
  52. this.client.on("error", err => {
  53. if (this.getStatus() === "INITIALIZING") reject(err);
  54. if (this.getStatus() === "LOCKDOWN") return;
  55. this.log("ERROR", `Error ${err.message}.`);
  56. });
  57. this.client.on("connect", () => {
  58. this.log("INFO", "Connected succesfully.");
  59. if (this.getStatus() === "INITIALIZING") resolve();
  60. else if (this.getStatus() === "FAILED" || this.getStatus() === "RECONNECTING") this.setStatus("READY");
  61. });
  62. });
  63. }
  64. /**
  65. * Quits redis client
  66. *
  67. * @returns {Promise} - returns promise (reject, resolve)
  68. */
  69. QUIT() {
  70. return new Promise(resolve => {
  71. if (CacheModule.client.connected) {
  72. CacheModule.client.quit();
  73. Object.keys(pubs).forEach(channel => pubs[channel].quit());
  74. Object.keys(subs).forEach(channel => subs[channel].client.quit());
  75. }
  76. resolve();
  77. });
  78. }
  79. /**
  80. * Sets a single value in a table
  81. *
  82. * @param {object} payload - object containing payload
  83. * @param {string} payload.table - name of the table we want to set a key of (table === redis hash)
  84. * @param {string} payload.key - name of the key to set
  85. * @param {*} payload.value - the value we want to set
  86. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  87. * @returns {Promise} - returns a promise (resolve, reject)
  88. */
  89. HSET(payload) {
  90. // table, key, value, cb, stringifyJson = true
  91. return new Promise((resolve, reject) => {
  92. let { key } = payload;
  93. let { value } = payload;
  94. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  95. // automatically stringify objects and arrays into JSON
  96. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  97. CacheModule.client.hset(payload.table, key, value, err => {
  98. if (err) return reject(new Error(err));
  99. return resolve(JSON.parse(value));
  100. });
  101. });
  102. }
  103. /**
  104. * Gets a single value from a table
  105. *
  106. * @param {object} payload - object containing payload
  107. * @param {string} payload.table - name of the table to get the value from (table === redis hash)
  108. * @param {string} payload.key - name of the key to fetch
  109. * @param {boolean} [payload.parseJson=true] - attempt to parse returned data as JSON
  110. * @returns {Promise} - returns a promise (resolve, reject)
  111. */
  112. HGET(payload) {
  113. // table, key, parseJson = true
  114. return new Promise((resolve, reject) => {
  115. let { key } = payload;
  116. if (!key) return reject(new Error("Invalid key!"));
  117. if (!payload.table) return reject(new Error("Invalid table!"));
  118. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  119. return CacheModule.client.hget(payload.table, key, (err, value) => {
  120. if (err) return reject(new Error(err));
  121. try {
  122. value = JSON.parse(value);
  123. } catch (e) {
  124. return reject(err);
  125. }
  126. return resolve(value);
  127. });
  128. });
  129. }
  130. /**
  131. * Deletes a single value from a table
  132. *
  133. * @param {object} payload - object containing payload
  134. * @param {string} payload.table - name of the table to delete the value from (table === redis hash)
  135. * @param {string} payload.key - name of the key to delete
  136. * @returns {Promise} - returns a promise (resolve, reject)
  137. */
  138. HDEL(payload) {
  139. // table, key, cb
  140. return new Promise((resolve, reject) => {
  141. // if (!payload.key || !table || typeof key !== "string")
  142. // return cb(null, null);
  143. let { key } = payload;
  144. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  145. CacheModule.client.hdel(payload.table, key, err => {
  146. if (err) return reject(new Error(err));
  147. return resolve();
  148. });
  149. });
  150. }
  151. /**
  152. * Returns all the keys for a table
  153. *
  154. * @param {object} payload - object containing payload
  155. * @param {string} payload.table - name of the table to get the values from (table === redis hash)
  156. * @param {boolean} [payload.parseJson=true] - attempts to parse all values as JSON by default
  157. * @returns {Promise} - returns a promise (resolve, reject)
  158. */
  159. HGETALL(payload) {
  160. // table, cb, parseJson = true
  161. return new Promise((resolve, reject) => {
  162. CacheModule.client.hgetall(payload.table, (err, obj) => {
  163. if (err) return reject(new Error(err));
  164. if (obj)
  165. Object.keys(obj).forEach(key => {
  166. obj[key] = JSON.parse(obj[key]);
  167. });
  168. else if (!obj) obj = [];
  169. return resolve(obj);
  170. });
  171. });
  172. }
  173. /**
  174. * Publish a message to a channel, caches the redis client connection
  175. *
  176. * @param {object} payload - object containing payload
  177. * @param {string} payload.channel - the name of the channel we want to publish a message to
  178. * @param {*} payload.value - the value we want to send
  179. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  180. * @returns {Promise} - returns a promise (resolve, reject)
  181. */
  182. PUB(payload) {
  183. // channel, value, stringifyJson = true
  184. return new Promise((resolve, reject) => {
  185. /* if (pubs[channel] === undefined) {
  186. pubs[channel] = redis.createClient({ url: CacheModule.url });
  187. pubs[channel].on('error', (err) => console.error);
  188. } */
  189. let { value } = payload;
  190. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  191. // pubs[channel].publish(channel, value);
  192. CacheModule.client.publish(payload.channel, value, err => {
  193. if (err) reject(err);
  194. else resolve();
  195. });
  196. });
  197. }
  198. /**
  199. * Subscribe to a channel, caches the redis client connection
  200. *
  201. * @param {object} payload - object containing payload
  202. * @param {string} payload.channel - name of the channel to subscribe to
  203. * @param {boolean} [payload.parseJson=true] - parse the message as JSON
  204. * @returns {Promise} - returns a promise (resolve, reject)
  205. */
  206. SUB(payload) {
  207. // channel, cb, parseJson = true
  208. return new Promise(resolve => {
  209. if (subs[payload.channel] === undefined) {
  210. subs[payload.channel] = {
  211. client: redis.createClient({
  212. url: CacheModule.url,
  213. password: CacheModule.password
  214. }),
  215. cbs: []
  216. };
  217. subs[payload.channel].client.on("message", (channel, message) => {
  218. try {
  219. CacheModule.log("INFO", `Got message. Channel: ${channel}, message: ${message}`);
  220. message = JSON.parse(message);
  221. } catch (err) {
  222. console.error(err);
  223. }
  224. return subs[channel].cbs.forEach(cb => cb(message));
  225. });
  226. subs[payload.channel].client.subscribe(payload.channel);
  227. }
  228. subs[payload.channel].cbs.push(payload.cb);
  229. resolve();
  230. });
  231. }
  232. /**
  233. * Returns a redis schema
  234. *
  235. * @param {object} payload - object containing the payload
  236. * @param {string} payload.schemaName - the name of the schema to get
  237. * @returns {Promise} - returns promise (reject, resolve)
  238. */
  239. GET_SCHEMA(payload) {
  240. return new Promise(resolve => {
  241. resolve(CacheModule.schemas[payload.schemaName]);
  242. });
  243. }
  244. }
  245. export default new _CacheModule();