stations.js 45 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745
  1. import async from "async";
  2. import CoreClass from "../core";
  3. let StationsModule;
  4. let CacheModule;
  5. let DBModule;
  6. let UtilsModule;
  7. let WSModule;
  8. let SongsModule;
  9. let PlaylistsModule;
  10. let NotificationsModule;
  11. class _StationsModule extends CoreClass {
  12. // eslint-disable-next-line require-jsdoc
  13. constructor() {
  14. super("stations");
  15. StationsModule = this;
  16. }
  17. /**
  18. * Initialises the stations module
  19. *
  20. * @returns {Promise} - returns promise (reject, resolve)
  21. */
  22. async initialize() {
  23. CacheModule = this.moduleManager.modules.cache;
  24. DBModule = this.moduleManager.modules.db;
  25. UtilsModule = this.moduleManager.modules.utils;
  26. WSModule = this.moduleManager.modules.ws;
  27. SongsModule = this.moduleManager.modules.songs;
  28. PlaylistsModule = this.moduleManager.modules.playlists;
  29. NotificationsModule = this.moduleManager.modules.notifications;
  30. this.userList = {};
  31. this.usersPerStation = {};
  32. this.usersPerStationCount = {};
  33. // TEMP
  34. CacheModule.runJob("SUB", {
  35. channel: "station.pause",
  36. cb: async stationId => {
  37. NotificationsModule.runJob("REMOVE", {
  38. subscription: `stations.nextSong?id=${stationId}`
  39. }).then();
  40. }
  41. });
  42. CacheModule.runJob("SUB", {
  43. channel: "station.resume",
  44. cb: async stationId => {
  45. StationsModule.runJob("INITIALIZE_STATION", { stationId }).then();
  46. }
  47. });
  48. CacheModule.runJob("SUB", {
  49. channel: "station.queueUpdate",
  50. cb: async stationId => {
  51. StationsModule.runJob("GET_STATION", { stationId }).then(station => {
  52. if (!station.currentSong && station.queue.length > 0) {
  53. StationsModule.runJob("INITIALIZE_STATION", {
  54. stationId
  55. }).then();
  56. }
  57. });
  58. }
  59. });
  60. CacheModule.runJob("SUB", {
  61. channel: "station.newOfficialPlaylist",
  62. cb: async stationId => {
  63. CacheModule.runJob("HGET", {
  64. table: "officialPlaylists",
  65. key: stationId
  66. }).then(playlistObj => {
  67. if (playlistObj) {
  68. WSModule.runJob("EMIT_TO_ROOM", {
  69. room: `station.${stationId}`,
  70. args: ["event:newOfficialPlaylist", { data: { playlist: playlistObj.songs } }]
  71. });
  72. }
  73. });
  74. }
  75. });
  76. const stationModel = (this.stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }));
  77. const stationSchema = (this.stationSchema = await CacheModule.runJob("GET_SCHEMA", { schemaName: "station" }));
  78. return new Promise((resolve, reject) =>
  79. async.waterfall(
  80. [
  81. next => {
  82. this.setStage(2);
  83. CacheModule.runJob("HGETALL", { table: "stations" })
  84. .then(stations => {
  85. next(null, stations);
  86. })
  87. .catch(next);
  88. },
  89. (stations, next) => {
  90. this.setStage(3);
  91. if (!stations) return next();
  92. const stationIds = Object.keys(stations);
  93. return async.each(
  94. stationIds,
  95. (stationId, next) => {
  96. stationModel.findOne({ _id: stationId }, (err, station) => {
  97. if (err) next(err);
  98. else if (!station) {
  99. CacheModule.runJob("HDEL", {
  100. table: "stations",
  101. key: stationId
  102. })
  103. .then(() => {
  104. next();
  105. })
  106. .catch(next);
  107. } else next();
  108. });
  109. },
  110. next
  111. );
  112. },
  113. next => {
  114. this.setStage(4);
  115. stationModel.find({}, next);
  116. },
  117. (stations, next) => {
  118. this.setStage(5);
  119. async.each(
  120. stations,
  121. (station, next2) => {
  122. async.waterfall(
  123. [
  124. next => {
  125. CacheModule.runJob("HSET", {
  126. table: "stations",
  127. key: station._id,
  128. value: stationSchema(station)
  129. })
  130. .then(station => next(null, station))
  131. .catch(next);
  132. },
  133. (station, next) => {
  134. StationsModule.runJob(
  135. "INITIALIZE_STATION",
  136. {
  137. stationId: station._id
  138. },
  139. null,
  140. -1
  141. )
  142. .then(() => {
  143. next();
  144. })
  145. .catch(next);
  146. }
  147. ],
  148. err => {
  149. next2(err);
  150. }
  151. );
  152. },
  153. next
  154. );
  155. }
  156. ],
  157. async err => {
  158. if (err) {
  159. err = await UtilsModule.runJob("GET_ERROR", {
  160. error: err
  161. });
  162. reject(new Error(err));
  163. } else {
  164. resolve();
  165. }
  166. }
  167. )
  168. );
  169. }
  170. /**
  171. * Initialises a station
  172. *
  173. * @param {object} payload - object that contains the payload
  174. * @param {string} payload.stationId - id of the station to initialise
  175. * @returns {Promise} - returns a promise (resolve, reject)
  176. */
  177. INITIALIZE_STATION(payload) {
  178. return new Promise((resolve, reject) => {
  179. async.waterfall(
  180. [
  181. next => {
  182. StationsModule.runJob(
  183. "GET_STATION",
  184. {
  185. stationId: payload.stationId
  186. },
  187. this
  188. )
  189. .then(station => {
  190. next(null, station);
  191. })
  192. .catch(next);
  193. },
  194. (station, next) => {
  195. if (!station) return next("Station not found.");
  196. return NotificationsModule.runJob(
  197. "UNSCHEDULE",
  198. {
  199. name: `stations.nextSong?id=${station._id}`
  200. },
  201. this
  202. )
  203. .then()
  204. .catch()
  205. .finally(() => {
  206. NotificationsModule.runJob("SUBSCRIBE", {
  207. name: `stations.nextSong?id=${station._id}`,
  208. cb: () =>
  209. StationsModule.runJob("SKIP_STATION", {
  210. stationId: station._id,
  211. natural: true
  212. }),
  213. unique: true,
  214. station
  215. })
  216. .then()
  217. .catch();
  218. if (station.paused) return next(true, station);
  219. return next(null, station);
  220. });
  221. },
  222. (station, next) => {
  223. if (!station.currentSong) {
  224. return StationsModule.runJob(
  225. "SKIP_STATION",
  226. {
  227. stationId: station._id,
  228. natural: false
  229. },
  230. this
  231. )
  232. .then(station => {
  233. next(true, station);
  234. })
  235. .catch(next)
  236. .finally(() => {});
  237. }
  238. let timeLeft =
  239. station.currentSong.duration * 1000 - (Date.now() - station.startedAt - station.timePaused);
  240. if (Number.isNaN(timeLeft)) timeLeft = -1;
  241. if (station.currentSong.duration * 1000 < timeLeft || timeLeft < 0) {
  242. return StationsModule.runJob(
  243. "SKIP_STATION",
  244. {
  245. stationId: station._id,
  246. natural: false
  247. },
  248. this
  249. )
  250. .then(station => {
  251. next(null, station);
  252. })
  253. .catch(next);
  254. }
  255. // name, time, cb, station
  256. NotificationsModule.runJob("SCHEDULE", {
  257. name: `stations.nextSong?id=${station._id}`,
  258. time: timeLeft,
  259. station
  260. });
  261. return next(null, station);
  262. }
  263. ],
  264. async (err, station) => {
  265. if (err && err !== true) {
  266. err = await UtilsModule.runJob(
  267. "GET_ERROR",
  268. {
  269. error: err
  270. },
  271. this
  272. );
  273. reject(new Error(err));
  274. } else resolve(station);
  275. }
  276. );
  277. });
  278. }
  279. /**
  280. * Attempts to get the station from Redis. If it's not in Redis, get it from Mongo and add it to Redis.
  281. *
  282. * @param {object} payload - object that contains the payload
  283. * @param {string} payload.stationId - id of the station
  284. * @returns {Promise} - returns a promise (resolve, reject)
  285. */
  286. GET_STATION(payload) {
  287. return new Promise((resolve, reject) => {
  288. async.waterfall(
  289. [
  290. next => {
  291. CacheModule.runJob("HGET", { table: "stations", key: payload.stationId }, this)
  292. .then(station => next(null, station))
  293. .catch(next);
  294. },
  295. (station, next) => {
  296. if (station) return next(true, station);
  297. return StationsModule.stationModel.findOne({ _id: payload.stationId }, next);
  298. },
  299. (station, next) => {
  300. if (station) {
  301. station = StationsModule.stationSchema(station);
  302. CacheModule.runJob("HSET", {
  303. table: "stations",
  304. key: payload.stationId,
  305. value: station
  306. })
  307. .then()
  308. .catch();
  309. next(true, station);
  310. } else next("Station not found");
  311. }
  312. ],
  313. async (err, station) => {
  314. if (err && err !== true) {
  315. err = await UtilsModule.runJob(
  316. "GET_ERROR",
  317. {
  318. error: err
  319. },
  320. this
  321. );
  322. reject(new Error(err));
  323. } else resolve(station);
  324. }
  325. );
  326. });
  327. }
  328. /**
  329. * Attempts to get a station by name, firstly from Redis. If it's not in Redis, get it from Mongo and add it to Redis.
  330. *
  331. * @param {object} payload - object that contains the payload
  332. * @param {string} payload.stationName - the unique name of the station
  333. * @returns {Promise} - returns a promise (resolve, reject)
  334. */
  335. async GET_STATION_BY_NAME(payload) {
  336. return new Promise((resolve, reject) =>
  337. async.waterfall(
  338. [
  339. next => {
  340. StationsModule.stationModel.findOne({ name: payload.stationName }, next);
  341. },
  342. (station, next) => {
  343. if (station) {
  344. station = StationsModule.stationSchema(station);
  345. CacheModule.runJob("HSET", {
  346. table: "stations",
  347. key: station._id,
  348. value: station
  349. });
  350. next(true, station);
  351. } else next("Station not found");
  352. }
  353. ],
  354. (err, station) => {
  355. if (err && err !== true) return reject(new Error(err));
  356. return resolve(station);
  357. }
  358. )
  359. );
  360. }
  361. /**
  362. * Gets stations data
  363. *
  364. * @param {object} payload - object containing the payload
  365. * @param {string} payload.page - the page
  366. * @param {string} payload.pageSize - the page size
  367. * @param {string} payload.properties - the properties to return for each station
  368. * @param {string} payload.sort - the sort object
  369. * @param {string} payload.queries - the queries array
  370. * @param {string} payload.operator - the operator for queries
  371. * @returns {Promise} - returns a promise (resolve, reject)
  372. */
  373. GET_DATA(payload) {
  374. return new Promise((resolve, reject) => {
  375. async.waterfall(
  376. [
  377. next => {
  378. const { queries, operator } = payload;
  379. const newQueries = queries.map(query => {
  380. const { data, filter, filterType } = query;
  381. const newQuery = {};
  382. if (filterType === "regex") {
  383. newQuery[filter.property] = new RegExp(`${data.slice(1, data.length - 1)}`, "i");
  384. } else if (filterType === "contains") {
  385. newQuery[filter.property] = new RegExp(
  386. `${data.replaceAll(/[.*+?^${}()|[\]\\]/g, "\\$&")}`,
  387. "i"
  388. );
  389. } else if (filterType === "exact") {
  390. newQuery[filter.property] = data.toString();
  391. }
  392. return newQuery;
  393. });
  394. const queryObject = {};
  395. if (newQueries.length > 0) {
  396. if (operator === "and") queryObject.$and = newQueries;
  397. else if (operator === "or") queryObject.$or = newQueries;
  398. else if (operator === "nor") queryObject.$nor = newQueries;
  399. }
  400. next(null, queryObject);
  401. },
  402. (queryObject, next) => {
  403. StationsModule.stationModel.find(queryObject).count((err, count) => {
  404. next(err, queryObject, count);
  405. });
  406. },
  407. (queryObject, count, next) => {
  408. const { page, pageSize, properties, sort } = payload;
  409. StationsModule.stationModel
  410. .find(queryObject)
  411. .sort(sort)
  412. .skip(pageSize * (page - 1))
  413. .limit(pageSize)
  414. .select(properties.join(" "))
  415. .exec((err, stations) => {
  416. next(err, count, stations);
  417. });
  418. }
  419. ],
  420. (err, count, stations) => {
  421. if (err && err !== true) return reject(new Error(err));
  422. return resolve({ data: stations, count });
  423. }
  424. );
  425. });
  426. }
  427. /**
  428. * Updates the station in cache from mongo or deletes station in cache if no longer in mongo.
  429. *
  430. * @param {object} payload - object that contains the payload
  431. * @param {string} payload.stationId - the id of the station to update
  432. * @returns {Promise} - returns a promise (resolve, reject)
  433. */
  434. UPDATE_STATION(payload) {
  435. return new Promise((resolve, reject) => {
  436. async.waterfall(
  437. [
  438. next => {
  439. StationsModule.stationModel.findOne({ _id: payload.stationId }, next);
  440. },
  441. (station, next) => {
  442. if (!station) {
  443. CacheModule.runJob("HDEL", {
  444. table: "stations",
  445. key: payload.stationId
  446. })
  447. .then()
  448. .catch();
  449. return next("Station not found");
  450. }
  451. return CacheModule.runJob(
  452. "HSET",
  453. {
  454. table: "stations",
  455. key: payload.stationId,
  456. value: station
  457. },
  458. this
  459. )
  460. .then(station => {
  461. next(null, station);
  462. })
  463. .catch(next);
  464. }
  465. ],
  466. async (err, station) => {
  467. if (err && err !== true) {
  468. err = await UtilsModule.runJob(
  469. "GET_ERROR",
  470. {
  471. error: err
  472. },
  473. this
  474. );
  475. reject(new Error(err));
  476. } else resolve(station);
  477. }
  478. );
  479. });
  480. }
  481. /**
  482. * Fills up the official station playlist queue using the songs from the official station playlist
  483. *
  484. * @param {object} payload - object that contains the payload
  485. * @param {string} payload.stationId - the id of the station
  486. * @param {string} payload.ignoreExistingQueue - ignore the existing queue songs, replacing the old queue with a completely fresh one
  487. * @returns {Promise} - returns a promise (resolve, reject)
  488. */
  489. FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST(payload) {
  490. return new Promise((resolve, reject) => {
  491. const { stationId, ignoreExistingQueue } = payload;
  492. async.waterfall(
  493. [
  494. next => {
  495. PlaylistsModule.runJob("GET_STATION_PLAYLIST", { stationId, includeSongs: true }, this)
  496. .then(response => {
  497. next(null, response.playlist);
  498. })
  499. .catch(next);
  500. },
  501. (playlist, next) => {
  502. StationsModule.runJob("GET_STATION", { stationId }, this)
  503. .then(station => {
  504. if (ignoreExistingQueue) station.queue = [];
  505. next(null, playlist, station);
  506. })
  507. .catch(next);
  508. },
  509. (playlist, station, next) => {
  510. if (station.playMode === "random") {
  511. UtilsModule.runJob("SHUFFLE", { array: playlist.songs }, this)
  512. .then(response => {
  513. next(null, response.array, station);
  514. })
  515. .catch(next);
  516. } else next(null, playlist.songs, station);
  517. },
  518. (_playlistSongs, station, next) => {
  519. let playlistSongs = JSON.parse(JSON.stringify(_playlistSongs));
  520. if (station.playMode === "sequential") {
  521. if (station.currentSongIndex <= playlistSongs.length) {
  522. const songsToAddToEnd = playlistSongs.splice(0, station.currentSongIndex);
  523. playlistSongs = [...playlistSongs, ...songsToAddToEnd];
  524. }
  525. }
  526. const songsStillNeeded = 50 - station.queue.length;
  527. const currentSongs = station.queue;
  528. const currentYoutubeIds = station.queue.map(song => song.youtubeId);
  529. const songsToAdd = [];
  530. let lastSongAdded = null;
  531. playlistSongs.every(song => {
  532. if (
  533. songsToAdd.length < songsStillNeeded &&
  534. currentYoutubeIds.indexOf(song.youtubeId) === -1
  535. ) {
  536. lastSongAdded = song;
  537. songsToAdd.push(song);
  538. return true;
  539. }
  540. if (songsToAdd.length >= songsStillNeeded) return false;
  541. return true;
  542. });
  543. let { currentSongIndex } = station;
  544. if (station.playMode === "sequential" && lastSongAdded) {
  545. const indexOfLastSong = _playlistSongs
  546. .map(song => song.youtubeId)
  547. .indexOf(lastSongAdded.youtubeId);
  548. if (indexOfLastSong !== -1) currentSongIndex = indexOfLastSong;
  549. }
  550. next(null, currentSongs, songsToAdd, currentSongIndex);
  551. },
  552. (currentSongs, songsToAdd, currentSongIndex, next) => {
  553. SongsModule.runJob("GET_SONGS", {
  554. songIds: songsToAdd.map(song => song._id),
  555. properties: [
  556. "youtubeId",
  557. "title",
  558. "duration",
  559. "skipDuration",
  560. "artists",
  561. "thumbnail",
  562. "status"
  563. ]
  564. })
  565. .then(response => {
  566. const newSongsToAdd = songsToAdd.map(song =>
  567. response.songs.find(newSong => newSong._id.toString() === song._id.toString())
  568. );
  569. next(null, currentSongs, newSongsToAdd, currentSongIndex);
  570. })
  571. .catch(err => next(err));
  572. },
  573. (currentSongs, songsToAdd, currentSongIndex, next) => {
  574. const newPlaylist = [...currentSongs, ...songsToAdd].map(song => {
  575. if (!song._id) song._id = null;
  576. return song;
  577. });
  578. next(null, newPlaylist, currentSongIndex);
  579. },
  580. (newPlaylist, currentSongIndex, next) => {
  581. StationsModule.stationModel.updateOne(
  582. { _id: stationId },
  583. { $set: { queue: newPlaylist, currentSongIndex } },
  584. { runValidators: true },
  585. err => {
  586. if (err) next(err);
  587. else
  588. StationsModule.runJob(
  589. "UPDATE_STATION",
  590. {
  591. stationId
  592. },
  593. this
  594. )
  595. .then(() => {
  596. next(null);
  597. })
  598. .catch(next);
  599. }
  600. );
  601. }
  602. ],
  603. err => {
  604. if (err) reject(err);
  605. else resolve();
  606. }
  607. );
  608. });
  609. }
  610. /**
  611. * Gets next station song
  612. *
  613. * @param {object} payload - object that contains the payload
  614. * @param {string} payload.stationId - the id of the station
  615. * @returns {Promise} - returns a promise (resolve, reject)
  616. */
  617. GET_NEXT_STATION_SONG(payload) {
  618. return new Promise((resolve, reject) => {
  619. const { stationId } = payload;
  620. async.waterfall(
  621. [
  622. next => {
  623. StationsModule.runJob("GET_STATION", { stationId }, this)
  624. .then(station => {
  625. next(null, station);
  626. })
  627. .catch(next);
  628. },
  629. (station, next) => {
  630. if (station.queue.length === 0) next("No songs available.");
  631. else {
  632. next(null, station.queue[0]);
  633. }
  634. },
  635. (queueSong, next) => {
  636. if (!queueSong._id) next(null, queueSong);
  637. else
  638. SongsModule.runJob("GET_SONG", { songId: queueSong._id }, this)
  639. .then(response => {
  640. const { song } = response;
  641. if (song) {
  642. const newSong = {
  643. _id: song._id,
  644. youtubeId: song.youtubeId,
  645. title: song.title,
  646. artists: song.artists,
  647. duration: song.duration,
  648. skipDuration: song.skipDuration,
  649. thumbnail: song.thumbnail,
  650. requestedAt: queueSong.requestedAt,
  651. requestedBy: queueSong.requestedBy,
  652. likes: song.likes,
  653. dislikes: song.dislikes,
  654. status: song.status
  655. };
  656. return next(null, newSong);
  657. }
  658. return next(null, song);
  659. })
  660. .catch(err => {
  661. next(err);
  662. });
  663. }
  664. ],
  665. (err, song) => {
  666. if (err) console.log(33333, err, payload);
  667. if (err) reject(err);
  668. else resolve({ song });
  669. }
  670. );
  671. });
  672. }
  673. /**
  674. * Removes first station queue song
  675. *
  676. * @param {object} payload - object that contains the payload
  677. * @param {string} payload.stationId - the id of the station
  678. * @returns {Promise} - returns a promise (resolve, reject)
  679. */
  680. REMOVE_FIRST_QUEUE_SONG(payload) {
  681. return new Promise((resolve, reject) => {
  682. const { stationId } = payload;
  683. async.waterfall(
  684. [
  685. next => {
  686. StationsModule.stationModel.updateOne(
  687. { _id: stationId },
  688. { $pop: { queue: -1 } },
  689. { runValidators: true },
  690. err => {
  691. if (err) next(err);
  692. else
  693. StationsModule.runJob(
  694. "UPDATE_STATION",
  695. {
  696. stationId
  697. },
  698. this
  699. )
  700. .then(() => {
  701. next(null);
  702. })
  703. .catch(next);
  704. }
  705. );
  706. }
  707. ],
  708. err => {
  709. if (err) reject(err);
  710. else resolve();
  711. }
  712. );
  713. });
  714. }
  715. /**
  716. * Skips a station
  717. *
  718. * @param {object} payload - object that contains the payload
  719. * @param {string} payload.stationId - the id of the station to skip
  720. * @param {string} payload.natural - whether to skip naturally or forcefully
  721. * @returns {Promise} - returns a promise (resolve, reject)
  722. */
  723. SKIP_STATION(payload) {
  724. return new Promise((resolve, reject) => {
  725. StationsModule.log("INFO", `Skipping station ${payload.stationId}.`);
  726. StationsModule.log("STATION_ISSUE", `SKIP_STATION_CB - Station ID: ${payload.stationId}.`);
  727. async.waterfall(
  728. [
  729. // Clears up any existing timers that would skip the station if the song ends
  730. next => {
  731. NotificationsModule.runJob("UNSCHEDULE", {
  732. name: `stations.nextSong?id=${payload.stationId}`
  733. })
  734. .then(() => {
  735. next();
  736. })
  737. .catch(next);
  738. },
  739. // Gets the station object
  740. next => {
  741. StationsModule.runJob(
  742. "GET_STATION",
  743. {
  744. stationId: payload.stationId
  745. },
  746. this
  747. )
  748. .then(station => next(null, station))
  749. .catch(next);
  750. },
  751. // eslint-disable-next-line consistent-return
  752. (station, next) => {
  753. if (!station) return next("Station not found.");
  754. if (station.type === "community" && station.partyMode && station.queue.length === 0)
  755. return next(null, null, station); // Community station with party mode enabled and no songs in the queue
  756. if (station.type === "community" && station.partyMode && station.queue.length > 0) {
  757. // Community station with party mode enabled and songs in the queue
  758. if (station.paused) return next(null, null, station);
  759. StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)
  760. .then(response => {
  761. StationsModule.runJob(
  762. "REMOVE_FIRST_QUEUE_SONG",
  763. { stationId: station._id },
  764. this
  765. ).then(() => {
  766. next(null, response.song, station);
  767. });
  768. })
  769. .catch(err => {
  770. if (err === "No songs available.") next(null, null, station);
  771. else next(err);
  772. });
  773. }
  774. if (station.type === "community" && !station.partyMode) {
  775. StationsModule.runJob(
  776. "FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST",
  777. { stationId: station._id },
  778. this
  779. )
  780. .then(() => {
  781. StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)
  782. .then(response => {
  783. StationsModule.runJob(
  784. "REMOVE_FIRST_QUEUE_SONG",
  785. { stationId: station._id },
  786. this
  787. ).then(() => {
  788. next(null, response.song, station);
  789. });
  790. })
  791. .catch(err => {
  792. if (err === "No songs available.") next(null, null, station);
  793. else next(err);
  794. });
  795. })
  796. .catch(next);
  797. }
  798. if (station.type === "official") {
  799. StationsModule.runJob(
  800. "FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST",
  801. { stationId: station._id },
  802. this
  803. )
  804. .then(() => {
  805. StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)
  806. .then(response => {
  807. StationsModule.runJob(
  808. "REMOVE_FIRST_QUEUE_SONG",
  809. { stationId: station._id },
  810. this
  811. )
  812. .then(() => {
  813. next(null, response.song, station);
  814. })
  815. .catch(next);
  816. })
  817. .catch(err => {
  818. if (err === "No songs available.") next(null, null, station);
  819. else next(err);
  820. });
  821. })
  822. .catch(next);
  823. }
  824. },
  825. (song, station, next) => {
  826. const $set = {};
  827. if (song === null) $set.currentSong = null;
  828. else {
  829. $set.currentSong = {
  830. _id: song._id,
  831. youtubeId: song.youtubeId,
  832. title: song.title,
  833. artists: song.artists,
  834. duration: song.duration,
  835. skipDuration: song.skipDuration,
  836. thumbnail: song.thumbnail,
  837. requestedAt: song.requestedAt,
  838. requestedBy: song.requestedBy,
  839. status: song.status
  840. };
  841. }
  842. $set.startedAt = Date.now();
  843. $set.timePaused = 0;
  844. if (station.paused) $set.pausedAt = Date.now();
  845. next(null, $set, song, station);
  846. },
  847. ($set, song, station, next) => {
  848. StationsModule.stationModel.updateOne({ _id: station._id }, { $set }, err => {
  849. if (err) return next(err);
  850. return StationsModule.runJob("UPDATE_STATION", { stationId: station._id }, this)
  851. .then(station => {
  852. CacheModule.runJob("PUB", {
  853. channel: "station.queueUpdate",
  854. value: payload.stationId
  855. })
  856. .then()
  857. .catch();
  858. next(null, station, song);
  859. })
  860. .catch(next);
  861. });
  862. },
  863. (station, song, next) => {
  864. if (station.currentSong !== null && station.currentSong.youtubeId !== undefined) {
  865. station.currentSong.likes = song.likes;
  866. station.currentSong.dislikes = song.dislikes;
  867. station.currentSong.skipVotes = 0;
  868. }
  869. next(null, station);
  870. }
  871. ],
  872. async (err, station) => {
  873. if (err) {
  874. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  875. StationsModule.log("ERROR", `Skipping station "${payload.stationId}" failed. "${err}"`);
  876. return reject(new Error(err));
  877. }
  878. // TODO Pub/Sub this
  879. const { currentSong } = station;
  880. WSModule.runJob("EMIT_TO_ROOM", {
  881. room: `station.${station._id}`,
  882. args: [
  883. "event:station.nextSong",
  884. {
  885. data: {
  886. currentSong,
  887. startedAt: station.startedAt,
  888. paused: station.paused,
  889. timePaused: 0,
  890. natural: payload.natural
  891. }
  892. }
  893. ]
  894. });
  895. WSModule.runJob("EMIT_TO_ROOM", {
  896. room: `manage-station.${station._id}`,
  897. args: ["event:station.nextSong", { data: { stationId: station._id, currentSong } }]
  898. });
  899. if (station.privacy === "public")
  900. WSModule.runJob("EMIT_TO_ROOM", {
  901. room: "home",
  902. args: ["event:station.nextSong", { data: { stationId: station._id, currentSong } }]
  903. });
  904. else {
  905. const sockets = await WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: "home" }, this);
  906. sockets.forEach(async socketId => {
  907. const socket = await WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId });
  908. if (!socket) return;
  909. const { session } = socket;
  910. if (session.sessionId) {
  911. CacheModule.runJob("HGET", { table: "sessions", key: session.sessionId }).then(
  912. session => {
  913. if (session) {
  914. DBModule.runJob("GET_MODEL", { modelName: "user" }).then(userModel => {
  915. userModel.findOne({ _id: session.userId }, (err, user) => {
  916. if (!err && user) {
  917. if (user.role === "admin")
  918. socket.dispatch("event:station.nextSong", {
  919. data: {
  920. stationId: station._id,
  921. currentSong
  922. }
  923. });
  924. else if (
  925. station.type === "community" &&
  926. station.owner === session.userId
  927. )
  928. socket.dispatch("event:station.nextSong", {
  929. data: {
  930. stationId: station._id,
  931. currentSong
  932. }
  933. });
  934. }
  935. });
  936. });
  937. }
  938. }
  939. );
  940. }
  941. });
  942. }
  943. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: `station.${station._id}` }).then(sockets => {
  944. if (station.currentSong !== null && station.currentSong.youtubeId !== undefined) {
  945. WSModule.runJob("SOCKETS_JOIN_SONG_ROOM", {
  946. sockets,
  947. room: `song.${station.currentSong.youtubeId}`
  948. });
  949. if (!station.paused) {
  950. NotificationsModule.runJob("SCHEDULE", {
  951. name: `stations.nextSong?id=${station._id}`,
  952. time: station.currentSong.duration * 1000,
  953. station
  954. });
  955. }
  956. } else WSModule.runJob("SOCKETS_LEAVE_SONG_ROOMS", { sockets });
  957. });
  958. return resolve({ station });
  959. }
  960. );
  961. });
  962. }
  963. /**
  964. * Checks if a user can view/access a station
  965. *
  966. * @param {object} payload - object that contains the payload
  967. * @param {object} payload.station - the station object of the station in question
  968. * @param {string} payload.userId - the id of the user in question
  969. * @param {boolean} payload.hideUnlisted - whether the user is allowed to see unlisted stations or not
  970. * @returns {Promise} - returns a promise (resolve, reject)
  971. */
  972. CAN_USER_VIEW_STATION(payload) {
  973. return new Promise((resolve, reject) => {
  974. async.waterfall(
  975. [
  976. next => {
  977. if (payload.station.privacy === "public") return next(true);
  978. if (payload.station.privacy === "unlisted")
  979. if (payload.hideUnlisted === true) return next();
  980. else return next(true);
  981. if (!payload.userId) return next("Not allowed");
  982. return next();
  983. },
  984. next => {
  985. DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(userModel => {
  986. userModel.findOne({ _id: payload.userId }, next);
  987. });
  988. },
  989. (user, next) => {
  990. if (!user) return next("Not allowed");
  991. if (user.role === "admin") return next(true);
  992. if (payload.station.type === "official") return next("Not allowed");
  993. if (payload.station.owner === payload.userId) return next(true);
  994. return next("Not allowed");
  995. }
  996. ],
  997. async errOrResult => {
  998. if (errOrResult !== true && errOrResult !== "Not allowed") {
  999. errOrResult = await UtilsModule.runJob(
  1000. "GET_ERROR",
  1001. {
  1002. error: errOrResult
  1003. },
  1004. this
  1005. );
  1006. reject(new Error(errOrResult));
  1007. } else {
  1008. resolve(errOrResult === true);
  1009. }
  1010. }
  1011. );
  1012. });
  1013. }
  1014. /**
  1015. * Checks if a user has favorited a station or not
  1016. *
  1017. * @param {object} payload - object that contains the payload
  1018. * @param {object} payload.stationId - the id of the station in question
  1019. * @param {string} payload.userId - the id of the user in question
  1020. * @returns {Promise} - returns a promise (resolve, reject)
  1021. */
  1022. HAS_USER_FAVORITED_STATION(payload) {
  1023. return new Promise((resolve, reject) => {
  1024. async.waterfall(
  1025. [
  1026. next => {
  1027. DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(userModel => {
  1028. userModel.findOne({ _id: payload.userId }, next);
  1029. });
  1030. },
  1031. (user, next) => {
  1032. if (!user) return next("User not found.");
  1033. if (user.favoriteStations.indexOf(payload.stationId) !== -1) return next(null, true);
  1034. return next(null, false);
  1035. }
  1036. ],
  1037. async (err, isStationFavorited) => {
  1038. if (err && err !== true) {
  1039. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1040. return reject(new Error(err));
  1041. }
  1042. return resolve(isStationFavorited);
  1043. }
  1044. );
  1045. });
  1046. }
  1047. /**
  1048. * Returns a list of sockets in a room that can and can't know about a station
  1049. *
  1050. * @param {object} payload - the payload object
  1051. * @param {object} payload.station - the station object
  1052. * @param {string} payload.room - the websockets room to get the sockets from
  1053. * @returns {Promise} - returns a promise (resolve, reject)
  1054. */
  1055. GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION(payload) {
  1056. return new Promise((resolve, reject) => {
  1057. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: payload.room }, this)
  1058. .then(socketIds => {
  1059. const sockets = [];
  1060. async.eachLimit(
  1061. socketIds,
  1062. 1,
  1063. (socketId, next) => {
  1064. WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this)
  1065. .then(socket => {
  1066. if (socket) sockets.push(socket);
  1067. next();
  1068. })
  1069. .catch(err => {
  1070. reject(err);
  1071. });
  1072. },
  1073. err => {
  1074. if (err) reject(err);
  1075. else {
  1076. let socketsThatCan = [];
  1077. const socketsThatCannot = [];
  1078. if (payload.station.privacy === "public") {
  1079. socketsThatCan = sockets;
  1080. resolve({ socketsThatCan, socketsThatCannot });
  1081. } else {
  1082. async.eachLimit(
  1083. sockets,
  1084. 1,
  1085. (socket, next) => {
  1086. const { session } = socket;
  1087. async.waterfall(
  1088. [
  1089. next => {
  1090. if (!session.sessionId) next("No session id");
  1091. else next();
  1092. },
  1093. next => {
  1094. CacheModule.runJob(
  1095. "HGET",
  1096. {
  1097. table: "sessions",
  1098. key: session.sessionId
  1099. },
  1100. this
  1101. )
  1102. .then(response => {
  1103. next(null, response);
  1104. })
  1105. .catch(next);
  1106. },
  1107. (session, next) => {
  1108. if (!session) next("No session");
  1109. else {
  1110. DBModule.runJob("GET_MODEL", { modelName: "user" }, this)
  1111. .then(userModel => {
  1112. next(null, userModel);
  1113. })
  1114. .catch(next);
  1115. }
  1116. },
  1117. (userModel, next) => {
  1118. if (!userModel) next("No user model");
  1119. else
  1120. userModel.findOne(
  1121. {
  1122. _id: session.userId
  1123. },
  1124. next
  1125. );
  1126. },
  1127. (user, next) => {
  1128. if (!user) next("No user found");
  1129. else if (user.role === "admin") {
  1130. socketsThatCan.push(socket);
  1131. next();
  1132. } else if (
  1133. payload.station.type === "community" &&
  1134. payload.station.owner === session.userId
  1135. ) {
  1136. socketsThatCan.push(socket);
  1137. next();
  1138. }
  1139. }
  1140. ],
  1141. err => {
  1142. if (err) socketsThatCannot.push(socket);
  1143. next();
  1144. }
  1145. );
  1146. },
  1147. err => {
  1148. if (err) reject(err);
  1149. else resolve({ socketsThatCan, socketsThatCannot });
  1150. }
  1151. );
  1152. }
  1153. }
  1154. }
  1155. );
  1156. })
  1157. .catch(reject);
  1158. });
  1159. }
  1160. /**
  1161. * Adds a playlist to be included in a station
  1162. *
  1163. * @param {object} payload - object that contains the payload
  1164. * @param {object} payload.stationId - the id of the station to include the playlist in
  1165. * @param {object} payload.playlistId - the id of the playlist to be included
  1166. * @returns {Promise} - returns a promise (resolve, reject)
  1167. */
  1168. INCLUDE_PLAYLIST(payload) {
  1169. return new Promise((resolve, reject) => {
  1170. async.waterfall(
  1171. [
  1172. next => {
  1173. if (!payload.stationId) next("Please specify a station id");
  1174. else if (!payload.playlistId) next("Please specify a playlist id");
  1175. else next();
  1176. },
  1177. next => {
  1178. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1179. .then(station => {
  1180. next(null, station);
  1181. })
  1182. .catch(next);
  1183. },
  1184. (station, next) => {
  1185. if (station.playlist === payload.playlistId) next("You cannot include the station playlist");
  1186. else if (station.includedPlaylists.indexOf(payload.playlistId) !== -1)
  1187. next("This playlist is already included");
  1188. else if (station.excludedPlaylists.indexOf(payload.playlistId) !== -1)
  1189. next(
  1190. "This playlist is currently excluded, please remove it from there before including it"
  1191. );
  1192. else
  1193. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)
  1194. .then(() => {
  1195. next(null);
  1196. })
  1197. .catch(next);
  1198. },
  1199. next => {
  1200. DBModule.runJob(
  1201. "GET_MODEL",
  1202. {
  1203. modelName: "station"
  1204. },
  1205. this
  1206. ).then(stationModel => {
  1207. stationModel.updateOne(
  1208. { _id: payload.stationId },
  1209. { $push: { includedPlaylists: payload.playlistId } },
  1210. next
  1211. );
  1212. });
  1213. },
  1214. (res, next) => {
  1215. StationsModule.runJob(
  1216. "UPDATE_STATION",
  1217. {
  1218. stationId: payload.stationId
  1219. },
  1220. this
  1221. )
  1222. .then(() => {
  1223. next();
  1224. })
  1225. .catch(next);
  1226. }
  1227. ],
  1228. async err => {
  1229. if (err && err !== true) {
  1230. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1231. return reject(new Error(err));
  1232. }
  1233. return resolve();
  1234. }
  1235. );
  1236. });
  1237. }
  1238. /**
  1239. * Removes a playlist that is included in a station
  1240. *
  1241. * @param {object} payload - object that contains the payload
  1242. * @param {object} payload.stationId - the id of the station
  1243. * @param {object} payload.playlistId - the id of the playlist
  1244. * @returns {Promise} - returns a promise (resolve, reject)
  1245. */
  1246. REMOVE_INCLUDED_PLAYLIST(payload) {
  1247. return new Promise((resolve, reject) => {
  1248. async.waterfall(
  1249. [
  1250. next => {
  1251. if (!payload.stationId) next("Please specify a station id");
  1252. else if (!payload.playlistId) next("Please specify a playlist id");
  1253. else next();
  1254. },
  1255. next => {
  1256. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1257. .then(station => {
  1258. next(null, station);
  1259. })
  1260. .catch(next);
  1261. },
  1262. (station, next) => {
  1263. if (station.includedPlaylists.indexOf(payload.playlistId) === -1)
  1264. next("This playlist isn't included");
  1265. else next();
  1266. },
  1267. next => {
  1268. DBModule.runJob(
  1269. "GET_MODEL",
  1270. {
  1271. modelName: "station"
  1272. },
  1273. this
  1274. ).then(stationModel => {
  1275. stationModel.updateOne(
  1276. { _id: payload.stationId },
  1277. { $pull: { includedPlaylists: payload.playlistId } },
  1278. next
  1279. );
  1280. });
  1281. },
  1282. (res, next) => {
  1283. StationsModule.runJob(
  1284. "UPDATE_STATION",
  1285. {
  1286. stationId: payload.stationId
  1287. },
  1288. this
  1289. )
  1290. .then(() => {
  1291. next();
  1292. })
  1293. .catch(next);
  1294. }
  1295. ],
  1296. async err => {
  1297. if (err && err !== true) {
  1298. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1299. return reject(new Error(err));
  1300. }
  1301. return resolve();
  1302. }
  1303. );
  1304. });
  1305. }
  1306. /**
  1307. * Adds a playlist to be excluded in a station
  1308. *
  1309. * @param {object} payload - object that contains the payload
  1310. * @param {object} payload.stationId - the id of the station
  1311. * @param {object} payload.playlistId - the id of the playlist
  1312. * @returns {Promise} - returns a promise (resolve, reject)
  1313. */
  1314. EXCLUDE_PLAYLIST(payload) {
  1315. return new Promise((resolve, reject) => {
  1316. async.waterfall(
  1317. [
  1318. next => {
  1319. if (!payload.stationId) next("Please specify a station id");
  1320. else if (!payload.playlistId) next("Please specify a playlist id");
  1321. else next();
  1322. },
  1323. next => {
  1324. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1325. .then(station => {
  1326. next(null, station);
  1327. })
  1328. .catch(next);
  1329. },
  1330. (station, next) => {
  1331. if (station.playlist === payload.playlistId) next("You cannot exclude the station playlist");
  1332. else if (station.excludedPlaylists.indexOf(payload.playlistId) !== -1)
  1333. next("This playlist is already excluded");
  1334. else if (station.includedPlaylists.indexOf(payload.playlistId) !== -1)
  1335. next(
  1336. "This playlist is currently included, please remove it from there before excluding it"
  1337. );
  1338. else
  1339. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)
  1340. .then(() => {
  1341. next(null);
  1342. })
  1343. .catch(next);
  1344. },
  1345. next => {
  1346. DBModule.runJob(
  1347. "GET_MODEL",
  1348. {
  1349. modelName: "station"
  1350. },
  1351. this
  1352. ).then(stationModel => {
  1353. stationModel.updateOne(
  1354. { _id: payload.stationId },
  1355. { $push: { excludedPlaylists: payload.playlistId } },
  1356. next
  1357. );
  1358. });
  1359. },
  1360. (res, next) => {
  1361. StationsModule.runJob(
  1362. "UPDATE_STATION",
  1363. {
  1364. stationId: payload.stationId
  1365. },
  1366. this
  1367. )
  1368. .then(() => {
  1369. next();
  1370. })
  1371. .catch(next);
  1372. }
  1373. ],
  1374. async err => {
  1375. if (err && err !== true) {
  1376. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1377. return reject(new Error(err));
  1378. }
  1379. return resolve();
  1380. }
  1381. );
  1382. });
  1383. }
  1384. /**
  1385. * Removes a playlist that is excluded in a station
  1386. *
  1387. * @param {object} payload - object that contains the payload
  1388. * @param {object} payload.stationId - the id of the station
  1389. * @param {object} payload.playlistId - the id of the playlist
  1390. * @returns {Promise} - returns a promise (resolve, reject)
  1391. */
  1392. REMOVE_EXCLUDED_PLAYLIST(payload) {
  1393. return new Promise((resolve, reject) => {
  1394. async.waterfall(
  1395. [
  1396. next => {
  1397. if (!payload.stationId) next("Please specify a station id");
  1398. else if (!payload.playlistId) next("Please specify a playlist id");
  1399. else next();
  1400. },
  1401. next => {
  1402. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1403. .then(station => {
  1404. next(null, station);
  1405. })
  1406. .catch(next);
  1407. },
  1408. (station, next) => {
  1409. if (station.excludedPlaylists.indexOf(payload.playlistId) === -1)
  1410. next("This playlist isn't excluded");
  1411. else next();
  1412. },
  1413. next => {
  1414. DBModule.runJob(
  1415. "GET_MODEL",
  1416. {
  1417. modelName: "station"
  1418. },
  1419. this
  1420. ).then(stationModel => {
  1421. stationModel.updateOne(
  1422. { _id: payload.stationId },
  1423. { $pull: { excludedPlaylists: payload.playlistId } },
  1424. next
  1425. );
  1426. });
  1427. },
  1428. (res, next) => {
  1429. StationsModule.runJob(
  1430. "UPDATE_STATION",
  1431. {
  1432. stationId: payload.stationId
  1433. },
  1434. this
  1435. )
  1436. .then(() => {
  1437. next();
  1438. })
  1439. .catch(next);
  1440. }
  1441. ],
  1442. async err => {
  1443. if (err && err !== true) {
  1444. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1445. return reject(new Error(err));
  1446. }
  1447. return resolve();
  1448. }
  1449. );
  1450. });
  1451. }
  1452. /**
  1453. * Removes included or excluded playlist from a station
  1454. *
  1455. * @param {object} payload - object that contains the payload
  1456. * @param {string} payload.playlistId - the playlist id
  1457. * @returns {Promise} - returns promise (reject, resolve)
  1458. */
  1459. REMOVE_INCLUDED_OR_EXCLUDED_PLAYLIST_FROM_STATIONS(payload) {
  1460. return new Promise((resolve, reject) => {
  1461. async.waterfall(
  1462. [
  1463. next => {
  1464. if (!payload.playlistId) next("Please specify a playlist id");
  1465. else next();
  1466. },
  1467. next => {
  1468. StationsModule.stationModel.updateMany(
  1469. {
  1470. $or: [
  1471. { includedPlaylists: payload.playlistId },
  1472. { excludedPlaylists: payload.playlistId }
  1473. ]
  1474. },
  1475. {
  1476. $pull: {
  1477. includedPlaylists: payload.playlistId,
  1478. excludedPlaylists: payload.playlistId
  1479. }
  1480. },
  1481. err => {
  1482. if (err) next(err);
  1483. else next();
  1484. }
  1485. );
  1486. }
  1487. ],
  1488. async err => {
  1489. if (err && err !== true) {
  1490. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1491. return reject(new Error(err));
  1492. }
  1493. return resolve();
  1494. }
  1495. );
  1496. });
  1497. }
  1498. /**
  1499. * Gets stations that include or exclude a specific playlist
  1500. *
  1501. * @param {object} payload - object that contains the payload
  1502. * @param {string} payload.playlistId - the playlist id
  1503. * @returns {Promise} - returns promise (reject, resolve)
  1504. */
  1505. GET_STATIONS_THAT_INCLUDE_OR_EXCLUDE_PLAYLIST(payload) {
  1506. return new Promise((resolve, reject) => {
  1507. DBModule.runJob(
  1508. "GET_MODEL",
  1509. {
  1510. modelName: "station"
  1511. },
  1512. this
  1513. ).then(stationModel => {
  1514. stationModel.find(
  1515. {
  1516. $or: [{ includedPlaylists: payload.playlistId }, { excludedPlaylists: payload.playlistId }]
  1517. },
  1518. (err, stations) => {
  1519. if (err) reject(err);
  1520. else resolve({ stationIds: stations.map(station => station._id) });
  1521. }
  1522. );
  1523. });
  1524. });
  1525. }
  1526. /**
  1527. * Clears every queue
  1528. *
  1529. * @returns {Promise} - returns a promise (resolve, reject)
  1530. */
  1531. CLEAR_EVERY_STATION_QUEUE() {
  1532. return new Promise((resolve, reject) => {
  1533. async.waterfall(
  1534. [
  1535. next => {
  1536. StationsModule.stationModel.updateMany({}, { $set: { queue: [] } }, err => {
  1537. if (err) next(err);
  1538. else {
  1539. StationsModule.stationModel.find({}, (err, stations) => {
  1540. if (err) next(err);
  1541. else {
  1542. async.eachLimit(
  1543. stations,
  1544. 1,
  1545. (station, next) => {
  1546. StationsModule.runJob("UPDATE_STATION", {
  1547. stationId: station._id
  1548. })
  1549. .then(() => next())
  1550. .catch(next);
  1551. CacheModule.runJob("PUB", {
  1552. channel: "station.queueUpdate",
  1553. value: station._id
  1554. })
  1555. .then()
  1556. .catch();
  1557. },
  1558. next
  1559. );
  1560. }
  1561. });
  1562. }
  1563. });
  1564. }
  1565. ],
  1566. err => {
  1567. if (err) reject(err);
  1568. else resolve();
  1569. }
  1570. );
  1571. });
  1572. }
  1573. /**
  1574. * Clears and refills a station queue
  1575. *
  1576. * @param {object} payload - object that contains the payload
  1577. * @param {string} payload.stationId - the station id
  1578. * @returns {Promise} - returns a promise (resolve, reject)
  1579. */
  1580. CLEAR_AND_REFILL_STATION_QUEUE(payload) {
  1581. return new Promise((resolve, reject) => {
  1582. async.waterfall(
  1583. [
  1584. next => {
  1585. StationsModule.runJob(
  1586. "FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST",
  1587. { stationId: payload.stationId, ignoreExistingQueue: true },
  1588. this
  1589. )
  1590. .then(() => {
  1591. CacheModule.runJob("PUB", {
  1592. channel: "station.queueUpdate",
  1593. value: payload.stationId
  1594. })
  1595. .then()
  1596. .catch();
  1597. next();
  1598. })
  1599. .catch(err => {
  1600. next(err);
  1601. });
  1602. }
  1603. ],
  1604. err => {
  1605. if (err) reject(err);
  1606. else resolve();
  1607. }
  1608. );
  1609. });
  1610. }
  1611. }
  1612. export default new _StationsModule();