index.js 9.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348
  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. ratings: await importSchema("ratings")
  34. };
  35. return new Promise((resolve, reject) => {
  36. this.url = config.get("redis").url;
  37. this.password = config.get("redis").password;
  38. this.log("INFO", "Connecting...");
  39. this.client = redis.createClient({
  40. url: this.url,
  41. password: this.password,
  42. retry_strategy: options => {
  43. if (this.getStatus() === "LOCKDOWN") return;
  44. if (this.getStatus() !== "RECONNECTING") this.setStatus("RECONNECTING");
  45. this.log("INFO", `Attempting to reconnect.`);
  46. if (options.attempt >= 10) {
  47. this.log("ERROR", `Stopped trying to reconnect.`);
  48. this.setStatus("FAILED");
  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. return new Promise((resolve, reject) => {
  91. let { key } = payload;
  92. let { value } = payload;
  93. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  94. // automatically stringify objects and arrays into JSON
  95. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  96. CacheModule.client.hset(payload.table, key, value, err => {
  97. if (err) return reject(new Error(err));
  98. return resolve(JSON.parse(value));
  99. });
  100. });
  101. }
  102. /**
  103. * Gets a single value from a table
  104. *
  105. * @param {object} payload - object containing payload
  106. * @param {string} payload.table - name of the table to get the value from (table === redis hash)
  107. * @param {string} payload.key - name of the key to fetch
  108. * @param {boolean} [payload.parseJson=true] - attempt to parse returned data as JSON
  109. * @returns {Promise} - returns a promise (resolve, reject)
  110. */
  111. HGET(payload) {
  112. return new Promise((resolve, reject) => {
  113. let { key } = payload;
  114. if (!key) {
  115. reject(new Error("Invalid key!"));
  116. return;
  117. }
  118. if (!payload.table) {
  119. reject(new Error("Invalid table!"));
  120. return;
  121. }
  122. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  123. CacheModule.client.hget(payload.table, key, (err, value) => {
  124. if (err) {
  125. reject(new Error(err));
  126. return;
  127. }
  128. try {
  129. value = JSON.parse(value);
  130. } catch (e) {
  131. reject(err);
  132. return;
  133. }
  134. resolve(value);
  135. });
  136. });
  137. }
  138. /**
  139. * Deletes a single value from a table
  140. *
  141. * @param {object} payload - object containing payload
  142. * @param {string} payload.table - name of the table to delete the value from (table === redis hash)
  143. * @param {string} payload.key - name of the key to delete
  144. * @returns {Promise} - returns a promise (resolve, reject)
  145. */
  146. HDEL(payload) {
  147. return new Promise((resolve, reject) => {
  148. let { key } = payload;
  149. if (!payload.table) {
  150. reject(new Error("Invalid table!"));
  151. return;
  152. }
  153. if (!key) {
  154. reject(new Error("Invalid key!"));
  155. return;
  156. }
  157. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  158. CacheModule.client.hdel(payload.table, key, err => {
  159. if (err) {
  160. reject(new Error(err));
  161. return;
  162. }
  163. resolve();
  164. });
  165. });
  166. }
  167. /**
  168. * Returns all the keys for a table
  169. *
  170. * @param {object} payload - object containing payload
  171. * @param {string} payload.table - name of the table to get the values from (table === redis hash)
  172. * @param {boolean} [payload.parseJson=true] - attempts to parse all values as JSON by default
  173. * @returns {Promise} - returns a promise (resolve, reject)
  174. */
  175. HGETALL(payload) {
  176. return new Promise((resolve, reject) => {
  177. if (!payload.table) {
  178. reject(new Error("Invalid table!"));
  179. return;
  180. }
  181. CacheModule.client.hgetall(payload.table, (err, obj) => {
  182. if (err) {
  183. reject(new Error(err));
  184. return;
  185. }
  186. if (obj)
  187. Object.keys(obj).forEach(key => {
  188. obj[key] = JSON.parse(obj[key]);
  189. });
  190. else if (!obj) obj = [];
  191. resolve(obj);
  192. });
  193. });
  194. }
  195. /**
  196. * Deletes a single value
  197. *
  198. * @param {object} payload - object containing payload
  199. * @param {string} payload.key - name of the key to delete
  200. * @returns {Promise} - returns a promise (resolve, reject)
  201. */
  202. DEL(payload) {
  203. return new Promise((resolve, reject) => {
  204. let { key } = payload;
  205. if (!key) {
  206. reject(new Error("Invalid key!"));
  207. return;
  208. }
  209. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  210. CacheModule.client.del(key, err => {
  211. if (err) {
  212. reject(new Error(err));
  213. return;
  214. }
  215. resolve();
  216. });
  217. });
  218. }
  219. /**
  220. * Publish a message to a channel, caches the redis client connection
  221. *
  222. * @param {object} payload - object containing payload
  223. * @param {string} payload.channel - the name of the channel we want to publish a message to
  224. * @param {*} payload.value - the value we want to send
  225. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  226. * @returns {Promise} - returns a promise (resolve, reject)
  227. */
  228. PUB(payload) {
  229. return new Promise((resolve, reject) => {
  230. let { value } = payload;
  231. if (!payload.channel) {
  232. reject(new Error("Invalid channel!"));
  233. return;
  234. }
  235. if (!value) {
  236. reject(new Error("Invalid value!"));
  237. return;
  238. }
  239. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  240. CacheModule.client.publish(payload.channel, value, err => {
  241. if (err) reject(err);
  242. else resolve();
  243. });
  244. });
  245. }
  246. /**
  247. * Subscribe to a channel, caches the redis client connection
  248. *
  249. * @param {object} payload - object containing payload
  250. * @param {string} payload.channel - name of the channel to subscribe to
  251. * @param {boolean} [payload.parseJson=true] - parse the message as JSON
  252. * @returns {Promise} - returns a promise (resolve, reject)
  253. */
  254. SUB(payload) {
  255. return new Promise((resolve, reject) => {
  256. if (!payload.channel) {
  257. reject(new Error("Invalid channel!"));
  258. return;
  259. }
  260. if (subs[payload.channel] === undefined) {
  261. subs[payload.channel] = {
  262. client: redis.createClient({
  263. url: CacheModule.url,
  264. password: CacheModule.password
  265. }),
  266. cbs: []
  267. };
  268. subs[payload.channel].client.on("message", (channel, message) => {
  269. if (message.startsWith("[") || message.startsWith("{"))
  270. try {
  271. message = JSON.parse(message);
  272. } catch (err) {
  273. console.error(err);
  274. }
  275. else if (message.startsWith('"') && message.endsWith('"'))
  276. message = message.substring(1).substring(0, message.length - 2);
  277. return subs[channel].cbs.forEach(cb => cb(message));
  278. });
  279. subs[payload.channel].client.subscribe(payload.channel);
  280. }
  281. subs[payload.channel].cbs.push(payload.cb);
  282. resolve();
  283. });
  284. }
  285. /**
  286. * Returns a redis schema
  287. *
  288. * @param {object} payload - object containing the payload
  289. * @param {string} payload.schemaName - the name of the schema to get
  290. * @returns {Promise} - returns promise (reject, resolve)
  291. */
  292. GET_SCHEMA(payload) {
  293. return new Promise(resolve => {
  294. resolve(CacheModule.schemas[payload.schemaName]);
  295. });
  296. }
  297. }
  298. export default new _CacheModule();