stations.js 44 KB

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