stations.js 39 KB

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