index.js 8.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266
  1. const CoreClass = require("../../core.js");
  2. const redis = require("redis");
  3. const config = require("config");
  4. const mongoose = require("mongoose");
  5. // Lightweight / convenience wrapper around redis module for our needs
  6. const pubs = {},
  7. subs = {};
  8. class CacheModule extends CoreClass {
  9. constructor() {
  10. super("cache");
  11. }
  12. initialize() {
  13. return new Promise((resolve, reject) => {
  14. this.schemas = {
  15. session: require("./schemas/session"),
  16. station: require("./schemas/station"),
  17. playlist: require("./schemas/playlist"),
  18. officialPlaylist: require("./schemas/officialPlaylist"),
  19. song: require("./schemas/song"),
  20. punishment: require("./schemas/punishment"),
  21. };
  22. this.url = config.get("redis").url;
  23. this.password = config.get("redis").password;
  24. this.log("INFO", "Connecting...");
  25. this.client = redis.createClient({
  26. url: this.url,
  27. password: this.password,
  28. retry_strategy: (options) => {
  29. if (this.getStatus() === "LOCKDOWN") return;
  30. if (this.getStatus() !== "RECONNECTING")
  31. this.setStatus("RECONNECTING");
  32. this.log("INFO", `Attempting to reconnect.`);
  33. if (options.attempt >= 10) {
  34. this.log("ERROR", `Stopped trying to reconnect.`);
  35. this.setStatus("FAILED");
  36. // this.failed = true;
  37. // this._lockdown();
  38. return undefined;
  39. }
  40. return 3000;
  41. },
  42. });
  43. this.client.on("error", (err) => {
  44. if (this.getStatus() === "INITIALIZING") reject(err);
  45. if (this.getStatus() === "LOCKDOWN") return;
  46. this.log("ERROR", `Error ${err.message}.`);
  47. });
  48. this.client.on("connect", () => {
  49. this.log("INFO", "Connected succesfully.");
  50. if (this.getStatus() === "INITIALIZING") resolve();
  51. else if (
  52. this.getStatus() === "FAILED" ||
  53. this.getStatus() === "RECONNECTING"
  54. )
  55. this.setStatus("READY");
  56. });
  57. });
  58. }
  59. /**
  60. * Gracefully closes all the Redis client connections
  61. */
  62. QUIT(payload) {
  63. return new Promise((resolve, reject) => {
  64. if (this.client.connected) {
  65. this.client.quit();
  66. Object.keys(pubs).forEach((channel) => pubs[channel].quit());
  67. Object.keys(subs).forEach((channel) =>
  68. subs[channel].client.quit()
  69. );
  70. }
  71. resolve();
  72. });
  73. }
  74. /**
  75. * Sets a single value in a table
  76. *
  77. * @param {String} table - name of the table we want to set a key of (table === redis hash)
  78. * @param {String} key - name of the key to set
  79. * @param {*} value - the value we want to set
  80. * @param {Function} cb - gets called when the value has been set in Redis
  81. * @param {Boolean} [stringifyJson=true] - stringify 'value' if it's an Object or Array
  82. */
  83. HSET(payload) {
  84. //table, key, value, cb, stringifyJson = true
  85. return new Promise((resolve, reject) => {
  86. let key = payload.key;
  87. let value = payload.value;
  88. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  89. // automatically stringify objects and arrays into JSON
  90. if (["object", "array"].includes(typeof value))
  91. value = JSON.stringify(value);
  92. this.client.hset(payload.table, key, value, (err) => {
  93. if (err) return reject(new Error(err));
  94. else resolve(JSON.parse(value));
  95. });
  96. });
  97. }
  98. /**
  99. * Gets a single value from a table
  100. *
  101. * @param {String} table - name of the table to get the value from (table === redis hash)
  102. * @param {String} key - name of the key to fetch
  103. * @param {Function} cb - gets called when the value is returned from Redis
  104. * @param {Boolean} [parseJson=true] - attempt to parse returned data as JSON
  105. */
  106. HGET(payload) {
  107. //table, key, cb, parseJson = true
  108. return new Promise((resolve, reject) => {
  109. // if (!key || !table)
  110. // return typeof cb === "function" ? cb(null, null) : null;
  111. let key = payload.key;
  112. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  113. this.client.hget(payload.table, key, (err, value) => {
  114. if (err) return reject(new Error(err));
  115. try {
  116. value = JSON.parse(value);
  117. } catch (e) {}
  118. resolve(value);
  119. });
  120. });
  121. }
  122. /**
  123. * Deletes a single value from a table
  124. *
  125. * @param {String} table - name of the table to delete the value from (table === redis hash)
  126. * @param {String} key - name of the key to delete
  127. * @param {Function} cb - gets called when the value has been deleted from Redis or when it returned an error
  128. */
  129. HDEL(payload) {
  130. //table, key, cb
  131. return new Promise((resolve, reject) => {
  132. // if (!payload.key || !table || typeof key !== "string")
  133. // return cb(null, null);
  134. let key = payload.key;
  135. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  136. this.client.hdel(payload.table, key, (err) => {
  137. if (err) return reject(new Error(err));
  138. else return resolve();
  139. });
  140. });
  141. }
  142. /**
  143. * Returns all the keys for a table
  144. *
  145. * @param {String} table - name of the table to get the values from (table === redis hash)
  146. * @param {Function} cb - gets called when the values are returned from Redis
  147. * @param {Boolean} [parseJson=true] - attempts to parse all values as JSON by default
  148. */
  149. HGETALL(payload) {
  150. //table, cb, parseJson = true
  151. return new Promise((resolve, reject) => {
  152. this.client.hgetall(payload.table, (err, obj) => {
  153. if (err) return reject(new Error(err));
  154. if (obj)
  155. Object.keys(obj).forEach((key) => {
  156. try {
  157. obj[key] = JSON.parse(obj[key]);
  158. } catch (e) {}
  159. });
  160. else if (!obj) obj = [];
  161. resolve(obj);
  162. });
  163. });
  164. }
  165. /**
  166. * Publish a message to a channel, caches the redis client connection
  167. *
  168. * @param {String} channel - the name of the channel we want to publish a message to
  169. * @param {*} value - the value we want to send
  170. * @param {Boolean} [stringifyJson=true] - stringify 'value' if it's an Object or Array
  171. */
  172. PUB(payload) {
  173. //channel, value, stringifyJson = true
  174. return new Promise((resolve, reject) => {
  175. /*if (pubs[channel] === undefined) {
  176. pubs[channel] = redis.createClient({ url: this.url });
  177. pubs[channel].on('error', (err) => console.error);
  178. }*/
  179. let value = payload.value;
  180. if (["object", "array"].includes(typeof value))
  181. value = JSON.stringify(value);
  182. //pubs[channel].publish(channel, value);
  183. this.client.publish(payload.channel, value);
  184. resolve();
  185. });
  186. }
  187. /**
  188. * Subscribe to a channel, caches the redis client connection
  189. *
  190. * @param {String} channel - name of the channel to subscribe to
  191. * @param {Function} cb - gets called when a message is received
  192. * @param {Boolean} [parseJson=true] - parse the message as JSON
  193. */
  194. SUB(payload) {
  195. //channel, cb, parseJson = true
  196. return new Promise((resolve, reject) => {
  197. if (subs[payload.channel] === undefined) {
  198. subs[payload.channel] = {
  199. client: redis.createClient({
  200. url: this.url,
  201. password: this.password,
  202. }),
  203. cbs: [],
  204. };
  205. subs[payload.channel].client.on(
  206. "message",
  207. (channel, message) => {
  208. try {
  209. message = JSON.parse(message);
  210. } catch (e) {}
  211. subs[channel].cbs.forEach((cb) => cb(message));
  212. }
  213. );
  214. subs[payload.channel].client.subscribe(payload.channel);
  215. }
  216. subs[payload.channel].cbs.push(payload.cb);
  217. resolve();
  218. });
  219. }
  220. GET_SCHEMA(payload) {
  221. return new Promise((resolve, reject) => {
  222. resolve(this.schemas[payload.schemaName]);
  223. });
  224. }
  225. }
  226. module.exports = new CacheModule();