index.js 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557
  1. import async from "async";
  2. import config from "config";
  3. import redis from "redis";
  4. import mongoose from "mongoose";
  5. import CoreClass from "../../core";
  6. // Lightweight / convenience wrapper around redis module for our needs
  7. const pubs = {};
  8. const subs = {};
  9. let CacheModule;
  10. class _CacheModule extends CoreClass {
  11. // eslint-disable-next-line require-jsdoc
  12. constructor() {
  13. super("cache");
  14. CacheModule = this;
  15. }
  16. /**
  17. * Initialises the cache/redis module
  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.log("INFO", "Connecting...");
  37. this.client = redis.createClient({
  38. ...config.get("redis"),
  39. reconnectStrategy: retries => {
  40. if (this.getStatus() !== "LOCKDOWN") {
  41. if (this.getStatus() !== "RECONNECTING") this.setStatus("RECONNECTING");
  42. this.log("INFO", `Attempting to reconnect.`);
  43. if (retries >= 10) {
  44. this.log("ERROR", `Stopped trying to reconnect.`);
  45. this.setStatus("FAILED");
  46. new Error("Stopped trying to reconnect.");
  47. } else {
  48. Math.min(retries * 50, 500);
  49. }
  50. }
  51. }
  52. });
  53. this.client.on("error", err => {
  54. if (this.getStatus() === "INITIALIZING") reject(err);
  55. if (this.getStatus() === "LOCKDOWN") return;
  56. this.log("ERROR", `Error ${err.message}.`);
  57. });
  58. this.client.on("ready", () => {
  59. this.log("INFO", "Redis is ready.");
  60. if (this.getStatus() === "INITIALIZING") resolve();
  61. else if (this.getStatus() === "FAILED" || this.getStatus() === "RECONNECTING") this.setStatus("READY");
  62. });
  63. this.client.connect().then(async () => {
  64. this.log("INFO", "Connected succesfully.");
  65. });
  66. // TODO move to a better place
  67. CacheModule.runJob("KEYS", { pattern: "longJobs.*" }).then(keys => {
  68. async.eachLimit(keys, 1, (key, next) => {
  69. CacheModule.runJob("DEL", { key }).finally(() => {
  70. next();
  71. });
  72. });
  73. });
  74. });
  75. }
  76. /**
  77. * Quits redis client
  78. * @returns {Promise} - returns promise (reject, resolve)
  79. */
  80. QUIT() {
  81. return new Promise(resolve => {
  82. if (CacheModule.client.connected) {
  83. CacheModule.client.quit();
  84. Object.keys(pubs).forEach(channel => pubs[channel].quit());
  85. Object.keys(subs).forEach(channel => subs[channel].client.quit());
  86. }
  87. resolve();
  88. });
  89. }
  90. /**
  91. * Sets a single value
  92. * @param {object} payload - object containing payload
  93. * @param {string} payload.key - name of the key to set
  94. * @param {*} payload.value - the value we want to set
  95. * @param {number} payload.ttl - ttl of the key in seconds
  96. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  97. * @returns {Promise} - returns a promise (resolve, reject)
  98. */
  99. SET(payload) {
  100. return new Promise((resolve, reject) => {
  101. let { key, value } = payload;
  102. const { ttl } = payload;
  103. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  104. // automatically stringify objects and arrays into JSON
  105. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  106. let options = null;
  107. if (ttl) {
  108. options = {
  109. EX: ttl
  110. };
  111. }
  112. CacheModule.client
  113. .SET(key, value, options)
  114. .then(() => {
  115. let parsed = value;
  116. try {
  117. parsed = JSON.parse(value);
  118. } catch {
  119. // Do nothing
  120. }
  121. resolve(parsed);
  122. })
  123. .catch(err => reject(new Error(err)));
  124. });
  125. }
  126. /**
  127. * Sets a single value in a table
  128. * @param {object} payload - object containing payload
  129. * @param {string} payload.table - name of the table we want to set a key of (table === redis hash)
  130. * @param {string} payload.key - name of the key to set
  131. * @param {*} payload.value - the value we want to set
  132. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  133. * @returns {Promise} - returns a promise (resolve, reject)
  134. */
  135. HSET(payload) {
  136. return new Promise((resolve, reject) => {
  137. let { key } = payload;
  138. let { value } = payload;
  139. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  140. // automatically stringify objects and arrays into JSON
  141. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  142. CacheModule.client
  143. .HSET(payload.table, key, value)
  144. .then(() => resolve(JSON.parse(value)))
  145. .catch(err => reject(new Error(err)));
  146. });
  147. }
  148. /**
  149. * Gets a single value
  150. * @param {object} payload - object containing payload
  151. * @param {string} payload.key - name of the key to fetch
  152. * @param {boolean} [payload.parseJson=true] - attempt to parse returned data as JSON
  153. * @returns {Promise} - returns a promise (resolve, reject)
  154. */
  155. GET(payload) {
  156. return new Promise((resolve, reject) => {
  157. let { key } = payload;
  158. if (!key) {
  159. reject(new Error("Invalid key!"));
  160. return;
  161. }
  162. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  163. CacheModule.client
  164. .GET(key, payload.value)
  165. .then(value => {
  166. if (value && !value.startsWith("{") && !value.startsWith("[")) return resolve(value);
  167. let parsedValue;
  168. try {
  169. parsedValue = JSON.parse(value);
  170. } catch (err) {
  171. return reject(err);
  172. }
  173. return resolve(parsedValue);
  174. })
  175. .catch(err => reject(new Error(err)));
  176. });
  177. }
  178. /**
  179. * Gets a single value from a table
  180. * @param {object} payload - object containing payload
  181. * @param {string} payload.table - name of the table to get the value from (table === redis hash)
  182. * @param {string} payload.key - name of the key to fetch
  183. * @param {boolean} [payload.parseJson=true] - attempt to parse returned data as JSON
  184. * @returns {Promise} - returns a promise (resolve, reject)
  185. */
  186. HGET(payload) {
  187. return new Promise((resolve, reject) => {
  188. let { key } = payload;
  189. if (!key) {
  190. reject(new Error("Invalid key!"));
  191. return;
  192. }
  193. if (!payload.table) {
  194. reject(new Error("Invalid table!"));
  195. return;
  196. }
  197. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  198. CacheModule.client
  199. .HGET(payload.table, key, payload.value)
  200. .then(value => {
  201. let parsedValue;
  202. try {
  203. parsedValue = JSON.parse(value);
  204. } catch (err) {
  205. return reject(err);
  206. }
  207. return resolve(parsedValue);
  208. })
  209. .catch(err => reject(new Error(err)));
  210. });
  211. }
  212. /**
  213. * Deletes a single value from a table
  214. * @param {object} payload - object containing payload
  215. * @param {string} payload.table - name of the table to delete the value from (table === redis hash)
  216. * @param {string} payload.key - name of the key to delete
  217. * @returns {Promise} - returns a promise (resolve, reject)
  218. */
  219. HDEL(payload) {
  220. return new Promise((resolve, reject) => {
  221. let { key } = payload;
  222. if (!payload.table) {
  223. reject(new Error("Invalid table!"));
  224. return;
  225. }
  226. if (!key) {
  227. reject(new Error("Invalid key!"));
  228. return;
  229. }
  230. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  231. CacheModule.client
  232. .HDEL(payload.table, key)
  233. .then(() => resolve())
  234. .catch(err => reject(new Error(err)));
  235. });
  236. }
  237. /**
  238. * Returns all the keys for a table
  239. * @param {object} payload - object containing payload
  240. * @param {string} payload.table - name of the table to get the values from (table === redis hash)
  241. * @param {boolean} [payload.parseJson=true] - attempts to parse all values as JSON by default
  242. * @returns {Promise} - returns a promise (resolve, reject)
  243. */
  244. HGETALL(payload) {
  245. return new Promise((resolve, reject) => {
  246. if (!payload.table) {
  247. reject(new Error("Invalid table!"));
  248. return;
  249. }
  250. CacheModule.client
  251. .HGETALL(payload.table)
  252. .then(obj => {
  253. if (obj)
  254. Object.keys(obj).forEach(key => {
  255. obj[key] = JSON.parse(obj[key]);
  256. });
  257. else if (!obj) obj = [];
  258. resolve(obj);
  259. })
  260. .catch(err => reject(new Error(err)));
  261. });
  262. }
  263. /**
  264. * Deletes a single value
  265. * @param {object} payload - object containing payload
  266. * @param {string} payload.key - name of the key to delete
  267. * @returns {Promise} - returns a promise (resolve, reject)
  268. */
  269. DEL(payload) {
  270. return new Promise((resolve, reject) => {
  271. let { key } = payload;
  272. if (!key) {
  273. reject(new Error("Invalid key!"));
  274. return;
  275. }
  276. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  277. CacheModule.client
  278. .DEL(key)
  279. .then(() => resolve())
  280. .catch(err => reject(new Error(err)));
  281. });
  282. }
  283. /**
  284. * Publish a message to a channel, caches the redis client connection
  285. * @param {object} payload - object containing payload
  286. * @param {string} payload.channel - the name of the channel we want to publish a message to
  287. * @param {*} payload.value - the value we want to send
  288. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  289. * @returns {Promise} - returns a promise (resolve, reject)
  290. */
  291. PUB(payload) {
  292. return new Promise((resolve, reject) => {
  293. let { value } = payload;
  294. if (!payload.channel) {
  295. reject(new Error("Invalid channel!"));
  296. return;
  297. }
  298. if (!value) {
  299. reject(new Error("Invalid value!"));
  300. return;
  301. }
  302. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  303. CacheModule.client
  304. .publish(payload.channel, value)
  305. .then(() => resolve())
  306. .catch(err => reject(new Error(err)));
  307. });
  308. }
  309. /**
  310. * Subscribe to a channel, caches the redis client connection
  311. * @param {object} payload - object containing payload
  312. * @param {string} payload.channel - name of the channel to subscribe to
  313. * @param {boolean} [payload.parseJson=true] - parse the message as JSON
  314. * @returns {Promise} - returns a promise (resolve, reject)
  315. */
  316. SUB(payload) {
  317. return new Promise((resolve, reject) => {
  318. if (!payload.channel) {
  319. reject(new Error("Invalid channel!"));
  320. return;
  321. }
  322. if (subs[payload.channel] === undefined) {
  323. subs[payload.channel] = {
  324. client: redis.createClient(config.get("redis")),
  325. cbs: []
  326. };
  327. subs[payload.channel].client.connect().then(() => {
  328. subs[payload.channel].client.subscribe(payload.channel, (message, channel) => {
  329. if (message.startsWith("[") || message.startsWith("{"))
  330. try {
  331. message = JSON.parse(message);
  332. } catch (err) {
  333. console.error(err);
  334. }
  335. else if (message.startsWith('"') && message.endsWith('"'))
  336. message = message.substring(1).substring(0, message.length - 2);
  337. subs[channel].cbs.forEach(cb => cb(message));
  338. });
  339. });
  340. }
  341. subs[payload.channel].cbs.push(payload.cb);
  342. resolve();
  343. });
  344. }
  345. /**
  346. * Gets a full list from Redis
  347. * @param {object} payload - object containing payload
  348. * @param {string} payload.key - name of the table to get the value from (table === redis hash)
  349. * @returns {Promise} - returns a promise (resolve, reject)
  350. */
  351. LRANGE(payload) {
  352. return new Promise((resolve, reject) => {
  353. let { key } = payload;
  354. if (!key) {
  355. reject(new Error("Invalid key!"));
  356. return;
  357. }
  358. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  359. CacheModule.client
  360. .LRANGE(key, 0, -1)
  361. .then(list => resolve(list))
  362. .catch(err => reject(new Error(err)));
  363. });
  364. }
  365. /**
  366. * Adds a value to a list in Redis
  367. * @param {object} payload - object containing payload
  368. * @param {string} payload.key - name of the list
  369. * @param {*} payload.value - the value we want to set
  370. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  371. * @returns {Promise} - returns a promise (resolve, reject)
  372. */
  373. RPUSH(payload) {
  374. return new Promise((resolve, reject) => {
  375. let { key, value } = payload;
  376. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  377. // automatically stringify objects and arrays into JSON
  378. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  379. CacheModule.client
  380. .RPUSH(key, value)
  381. .then(() => resolve())
  382. .catch(err => reject(new Error(err)));
  383. });
  384. }
  385. /**
  386. * Adds a value to a list in Redis using LPUSH
  387. * @param {object} payload - object containing payload
  388. * @param {string} payload.key - name of the list
  389. * @param {*} payload.value - the value we want to set
  390. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  391. * @returns {Promise} - returns a promise (resolve, reject)
  392. */
  393. LPUSH(payload) {
  394. return new Promise((resolve, reject) => {
  395. let { key, value } = payload;
  396. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  397. // automatically stringify objects and arrays into JSON
  398. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  399. CacheModule.client
  400. .LPUSH(key, value)
  401. .then(() => resolve())
  402. .catch(err => reject(new Error(err)));
  403. });
  404. }
  405. /**
  406. * Gets the length of a Redis list
  407. * @param {object} payload - object containing payload
  408. * @param {string} payload.key - name of the list
  409. * @returns {Promise} - returns a promise (resolve, reject)
  410. */
  411. LLEN(payload) {
  412. return new Promise((resolve, reject) => {
  413. const { key } = payload;
  414. CacheModule.client
  415. .LLEN(key)
  416. .then(len => resolve(len))
  417. .catch(err => reject(new Error(err)));
  418. });
  419. }
  420. /**
  421. * Removes an item from a list using RPOP
  422. * @param {object} payload - object containing payload
  423. * @param {string} payload.key - name of the list
  424. * @returns {Promise} - returns a promise (resolve, reject)
  425. */
  426. RPOP(payload) {
  427. return new Promise((resolve, reject) => {
  428. const { key } = payload;
  429. CacheModule.client
  430. .RPOP(key)
  431. .then(() => resolve())
  432. .catch(err => reject(new Error(err)));
  433. });
  434. }
  435. /**
  436. * Removes a value from a list in Redis
  437. * @param {object} payload - object containing payload
  438. * @param {string} payload.key - name of the list
  439. * @param {*} payload.value - the value we want to remove
  440. * @param {boolean} [payload.stringifyJson=true] - stringify 'value' if it's an Object or Array
  441. * @returns {Promise} - returns a promise (resolve, reject)
  442. */
  443. LREM(payload) {
  444. return new Promise((resolve, reject) => {
  445. let { key, value } = payload;
  446. if (mongoose.Types.ObjectId.isValid(key)) key = key.toString();
  447. // automatically stringify objects and arrays into JSON
  448. if (["object", "array"].includes(typeof value)) value = JSON.stringify(value);
  449. CacheModule.client
  450. .LREM(key, 1, value)
  451. .then(() => resolve())
  452. .catch(err => reject(new Error(err)));
  453. });
  454. }
  455. /**
  456. * Gets a list of keys in Redis with a matching pattern
  457. * @param {object} payload - object containing payload
  458. * @param {string} payload.pattern - pattern to search for
  459. * @returns {Promise} - returns a promise (resolve, reject)
  460. */
  461. KEYS(payload) {
  462. return new Promise((resolve, reject) => {
  463. const { pattern } = payload;
  464. CacheModule.client
  465. .KEYS(pattern)
  466. .then(keys => resolve(keys))
  467. .catch(err => reject(new Error(err)));
  468. });
  469. }
  470. /**
  471. * Returns a redis schema
  472. * @param {object} payload - object containing the payload
  473. * @param {string} payload.schemaName - the name of the schema to get
  474. * @returns {Promise} - returns promise (reject, resolve)
  475. */
  476. GET_SCHEMA(payload) {
  477. return new Promise(resolve => {
  478. resolve(CacheModule.schemas[payload.schemaName]);
  479. });
  480. }
  481. }
  482. export default new _CacheModule();