stations.js 39 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234
  1. 'use strict';
  2. const async = require('async'),
  3. request = require('request'),
  4. config = require('config'),
  5. _ = require('underscore')._;
  6. const hooks = require('./hooks');
  7. const moduleManager = require("../../index");
  8. const db = moduleManager.modules["db"];
  9. const cache = moduleManager.modules["cache"];
  10. const notifications = moduleManager.modules["notifications"];
  11. const utils = moduleManager.modules["utils"];
  12. const logger = moduleManager.modules["logger"];
  13. const stations = moduleManager.modules["stations"];
  14. const songs = moduleManager.modules["songs"];
  15. let userList = {};
  16. let usersPerStation = {};
  17. let usersPerStationCount = {};
  18. setInterval(() => {
  19. let stationsCountUpdated = [];
  20. let stationsUpdated = [];
  21. let oldUsersPerStation = usersPerStation;
  22. usersPerStation = {};
  23. let oldUsersPerStationCount = usersPerStationCount;
  24. usersPerStationCount = {};
  25. async.each(Object.keys(userList), function(socketId, next) {
  26. utils.socketFromSession(socketId).then((socket) => {
  27. let stationId = userList[socketId];
  28. if (!socket || Object.keys(socket.rooms).indexOf(`station.${stationId}`) === -1) {
  29. if (stationsCountUpdated.indexOf(stationId) === -1) stationsCountUpdated.push(stationId);
  30. if (stationsUpdated.indexOf(stationId) === -1) stationsUpdated.push(stationId);
  31. delete userList[socketId];
  32. return next();
  33. }
  34. if (!usersPerStationCount[stationId]) usersPerStationCount[stationId] = 0;
  35. usersPerStationCount[stationId]++;
  36. if (!usersPerStation[stationId]) usersPerStation[stationId] = [];
  37. async.waterfall([
  38. (next) => {
  39. if (!socket.session || !socket.session.sessionId) return next('No session found.');
  40. cache.hget('sessions', socket.session.sessionId, next);
  41. },
  42. (session, next) => {
  43. if (!session) return next('Session not found.');
  44. db.models.user.findOne({_id: session.userId}, next);
  45. },
  46. (user, next) => {
  47. if (!user) return next('User not found.');
  48. if (usersPerStation[stationId].indexOf(user.username) !== -1) return next('User already in the list.');
  49. next(null, user.username);
  50. }
  51. ], (err, username) => {
  52. if (!err) {
  53. usersPerStation[stationId].push(username);
  54. }
  55. next();
  56. });
  57. });
  58. //TODO Code to show users
  59. }, (err) => {
  60. for (let stationId in usersPerStationCount) {
  61. if (oldUsersPerStationCount[stationId] !== usersPerStationCount[stationId]) {
  62. if (stationsCountUpdated.indexOf(stationId) === -1) stationsCountUpdated.push(stationId);
  63. }
  64. }
  65. for (let stationId in usersPerStation) {
  66. if (_.difference(usersPerStation[stationId], oldUsersPerStation[stationId]).length > 0 || _.difference(oldUsersPerStation[stationId], usersPerStation[stationId]).length > 0) {
  67. if (stationsUpdated.indexOf(stationId) === -1) stationsUpdated.push(stationId);
  68. }
  69. }
  70. stationsCountUpdated.forEach((stationId) => {
  71. //logger.info("UPDATE_STATION_USER_COUNT", `Updating user count of ${stationId}.`);
  72. cache.pub('station.updateUserCount', stationId);
  73. });
  74. stationsUpdated.forEach((stationId) => {
  75. //logger.info("UPDATE_STATION_USER_LIST", `Updating user list of ${stationId}.`);
  76. cache.pub('station.updateUsers', stationId);
  77. });
  78. //console.log("Userlist", usersPerStation);
  79. });
  80. }, 3000);
  81. cache.sub('station.updateUsers', stationId => {
  82. let list = usersPerStation[stationId] || [];
  83. utils.emitToRoom(`station.${stationId}`, "event:users.updated", list);
  84. });
  85. cache.sub('station.updateUserCount', stationId => {
  86. let count = usersPerStationCount[stationId] || 0;
  87. utils.emitToRoom(`station.${stationId}`, "event:userCount.updated", count);
  88. stations.getStation(stationId, async (err, station) => {
  89. if (station.privacy === 'public') utils.emitToRoom('home', "event:userCount.updated", stationId, count);
  90. else {
  91. let sockets = await utils.getRoomSockets('home');
  92. for (let socketId in sockets) {
  93. let socket = sockets[socketId];
  94. let session = sockets[socketId].session;
  95. if (session.sessionId) {
  96. cache.hget('sessions', session.sessionId, (err, session) => {
  97. if (!err && session) {
  98. db.models.user.findOne({_id: session.userId}, (err, user) => {
  99. if (user.role === 'admin') socket.emit("event:userCount.updated", stationId, count);
  100. else if (station.type === "community" && station.owner === session.userId) socket.emit("event:userCount.updated", stationId, count);
  101. });
  102. }
  103. });
  104. }
  105. }
  106. }
  107. })
  108. });
  109. cache.sub('station.queueLockToggled', data => {
  110. utils.emitToRoom(`station.${data.stationId}`, "event:queueLockToggled", data.locked)
  111. });
  112. cache.sub('station.updatePartyMode', data => {
  113. utils.emitToRoom(`station.${data.stationId}`, "event:partyMode.updated", data.partyMode);
  114. });
  115. cache.sub('privatePlaylist.selected', data => {
  116. utils.emitToRoom(`station.${data.stationId}`, "event:privatePlaylist.selected", data.playlistId);
  117. });
  118. cache.sub('station.pause', stationId => {
  119. stations.getStation(stationId, (err, station) => {
  120. utils.emitToRoom(`station.${stationId}`, "event:stations.pause", { pausedAt: station.pausedAt });
  121. });
  122. });
  123. cache.sub('station.resume', stationId => {
  124. stations.getStation(stationId, (err, station) => {
  125. utils.emitToRoom(`station.${stationId}`, "event:stations.resume", { timePaused: station.timePaused });
  126. });
  127. });
  128. cache.sub('station.queueUpdate', stationId => {
  129. stations.getStation(stationId, (err, station) => {
  130. if (!err) utils.emitToRoom(`station.${stationId}`, "event:queue.update", station.queue);
  131. });
  132. });
  133. cache.sub('station.voteSkipSong', stationId => {
  134. utils.emitToRoom(`station.${stationId}`, "event:song.voteSkipSong");
  135. });
  136. cache.sub('station.remove', stationId => {
  137. utils.emitToRoom(`station.${stationId}`, 'event:stations.remove');
  138. utils.emitToRoom('admin.stations', 'event:admin.station.removed', stationId);
  139. });
  140. cache.sub('station.create', stationId => {
  141. stations.initializeStation(stationId, async (err, station) => {
  142. station.userCount = usersPerStationCount[stationId] || 0;
  143. if (err) console.error(err);
  144. utils.emitToRoom('admin.stations', 'event:admin.station.added', station);
  145. // TODO If community, check if on whitelist
  146. if (station.privacy === 'public') utils.emitToRoom('home', "event:stations.created", station);
  147. else {
  148. let sockets = await utils.getRoomSockets('home');
  149. for (let socketId in sockets) {
  150. let socket = sockets[socketId];
  151. let session = sockets[socketId].session;
  152. if (session.sessionId) {
  153. cache.hget('sessions', session.sessionId, (err, session) => {
  154. if (!err && session) {
  155. db.models.user.findOne({_id: session.userId}, (err, user) => {
  156. if (user.role === 'admin') socket.emit("event:stations.created", station);
  157. else if (station.type === "community" && station.owner === session.userId) socket.emit("event:stations.created", station);
  158. });
  159. }
  160. });
  161. }
  162. }
  163. }
  164. });
  165. });
  166. module.exports = {
  167. /**
  168. * Get a list of all the stations
  169. *
  170. * @param session
  171. * @param cb
  172. * @return {{ status: String, stations: Array }}
  173. */
  174. index: (session, cb) => {
  175. async.waterfall([
  176. (next) => {
  177. cache.hgetall('stations', next);
  178. },
  179. (stations, next) => {
  180. let resultStations = [];
  181. for (let id in stations) {
  182. resultStations.push(stations[id]);
  183. }
  184. next(null, stations);
  185. },
  186. (stationsArray, next) => {
  187. let resultStations = [];
  188. async.each(stationsArray, (station, next) => {
  189. async.waterfall([
  190. (next) => {
  191. stations.canUserViewStation(station, session.userId, (err, exists) => {
  192. next(err, exists);
  193. });
  194. }
  195. ], (err, exists) => {
  196. station.userCount = usersPerStationCount[station._id] || 0;
  197. if (exists) resultStations.push(station);
  198. next();
  199. });
  200. }, () => {
  201. next(null, resultStations);
  202. });
  203. }
  204. ], async (err, stations) => {
  205. if (err) {
  206. err = await utils.getError(err);
  207. logger.error("STATIONS_INDEX", `Indexing stations failed. "${err}"`);
  208. return cb({'status': 'failure', 'message': err});
  209. }
  210. logger.success("STATIONS_INDEX", `Indexing stations successful.`, false);
  211. return cb({'status': 'success', 'stations': stations});
  212. });
  213. },
  214. /**
  215. * Verifies that a station exists
  216. *
  217. * @param session
  218. * @param stationName - the station name
  219. * @param cb
  220. */
  221. existsByName: (session, stationName, cb) => {
  222. async.waterfall([
  223. (next) => {
  224. stations.getStationByName(stationName, next);
  225. },
  226. (station, next) => {
  227. if (!station) return next(null, false);
  228. stations.canUserViewStation(station, session.userId, (err, exists) => {
  229. next(err, exists);
  230. });
  231. }
  232. ], async (err, exists) => {
  233. if (err) {
  234. err = await utils.getError(err);
  235. logger.error("STATION_EXISTS_BY_NAME", `Checking if station "${stationName}" exists failed. "${err}"`);
  236. return cb({'status': 'failure', 'message': err});
  237. }
  238. logger.success("STATION_EXISTS_BY_NAME", `Station "${stationName}" exists successfully.`/*, false*/);
  239. cb({status: 'success', exists});
  240. });
  241. },
  242. /**
  243. * Gets the official playlist for a station
  244. *
  245. * @param session
  246. * @param stationId - the station id
  247. * @param cb
  248. */
  249. getPlaylist: (session, stationId, cb) => {
  250. async.waterfall([
  251. (next) => {
  252. stations.getStation(stationId, next);
  253. },
  254. (station, next) => {
  255. stations.canUserViewStation(station, session.userId, (err, canView) => {
  256. if (err) return next(err);
  257. if (canView) return next(null, station);
  258. return next('Insufficient permissions.');
  259. });
  260. },
  261. (station, next) => {
  262. if (!station) return next('Station not found.');
  263. else if (station.type !== 'official') return next('This is not an official station.');
  264. else next();
  265. },
  266. (next) => {
  267. cache.hget('officialPlaylists', stationId, next);
  268. },
  269. (playlist, next) => {
  270. if (!playlist) return next('Playlist not found.');
  271. next(null, playlist);
  272. }
  273. ], async (err, playlist) => {
  274. if (err) {
  275. err = await utils.getError(err);
  276. logger.error("STATIONS_GET_PLAYLIST", `Getting playlist for station "${stationId}" failed. "${err}"`);
  277. return cb({ status: 'failure', message: err });
  278. } else {
  279. logger.success("STATIONS_GET_PLAYLIST", `Got playlist for station "${stationId}" successfully.`, false);
  280. cb({ status: 'success', data: playlist.songs });
  281. }
  282. });
  283. },
  284. /**
  285. * Joins the station by its name
  286. *
  287. * @param session
  288. * @param stationName - the station name
  289. * @param cb
  290. * @return {{ status: String, userCount: Integer }}
  291. */
  292. join: (session, stationName, cb) => {
  293. async.waterfall([
  294. (next) => {
  295. stations.getStationByName(stationName, next);
  296. },
  297. (station, next) => {
  298. if (!station) return next('Station not found.');
  299. stations.canUserViewStation(station, session.userId, (err, canView) => {
  300. if (err) return next(err);
  301. if (!canView) next("Not allowed to join station.");
  302. else next(null, station);
  303. });
  304. },
  305. (station, next) => {
  306. utils.socketJoinRoom(session.socketId, `station.${station._id}`);
  307. let data = {
  308. _id: station._id,
  309. type: station.type,
  310. currentSong: station.currentSong,
  311. startedAt: station.startedAt,
  312. paused: station.paused,
  313. timePaused: station.timePaused,
  314. pausedAt: station.pausedAt,
  315. description: station.description,
  316. displayName: station.displayName,
  317. privacy: station.privacy,
  318. locked: station.locked,
  319. partyMode: station.partyMode,
  320. owner: station.owner,
  321. privatePlaylist: station.privatePlaylist
  322. };
  323. userList[session.socketId] = station._id;
  324. next(null, data);
  325. },
  326. (data, next) => {
  327. data = JSON.parse(JSON.stringify(data));
  328. data.userCount = usersPerStationCount[data._id] || 0;
  329. data.users = usersPerStation[data._id] || [];
  330. if (!data.currentSong || !data.currentSong.title) return next(null, data);
  331. utils.socketJoinSongRoom(session.socketId, `song.${data.currentSong.songId}`);
  332. data.currentSong.skipVotes = data.currentSong.skipVotes.length;
  333. songs.getSongFromId(data.currentSong.songId, (err, song) => {
  334. if (!err && song) {
  335. data.currentSong.likes = song.likes;
  336. data.currentSong.dislikes = song.dislikes;
  337. } else {
  338. data.currentSong.likes = -1;
  339. data.currentSong.dislikes = -1;
  340. }
  341. next(null, data);
  342. });
  343. }
  344. ], async (err, data) => {
  345. if (err) {
  346. err = await utils.getError(err);
  347. logger.error("STATIONS_JOIN", `Joining station "${stationName}" failed. "${err}"`);
  348. return cb({'status': 'failure', 'message': err});
  349. }
  350. logger.success("STATIONS_JOIN", `Joined station "${data._id}" successfully.`);
  351. cb({status: 'success', data});
  352. });
  353. },
  354. /**
  355. * Toggles if a station is locked
  356. *
  357. * @param session
  358. * @param stationId - the station id
  359. * @param cb
  360. */
  361. toggleLock: hooks.ownerRequired((session, stationId, cb) => {
  362. async.waterfall([
  363. (next) => {
  364. stations.getStation(stationId, next);
  365. },
  366. (station, next) => {
  367. db.models.station.updateOne({ _id: stationId }, { $set: { locked: !station.locked} }, next);
  368. },
  369. (res, next) => {
  370. stations.updateStation(stationId, next);
  371. }
  372. ], async (err, station) => {
  373. if (err) {
  374. err = await utils.getError(err);
  375. logger.error("STATIONS_UPDATE_LOCKED_STATUS", `Toggling the queue lock for station "${stationId}" failed. "${err}"`);
  376. return cb({ status: 'failure', message: err });
  377. } else {
  378. logger.success("STATIONS_UPDATE_LOCKED_STATUS", `Toggled the queue lock for station "${stationId}" successfully to "${station.locked}".`);
  379. cache.pub('station.queueLockToggled', {stationId, locked: station.locked});
  380. return cb({ status: 'success', data: station.locked });
  381. }
  382. });
  383. }),
  384. /**
  385. * Votes to skip a station
  386. *
  387. * @param session
  388. * @param stationId - the station id
  389. * @param cb
  390. */
  391. voteSkip: hooks.loginRequired((session, stationId, cb) => {
  392. let skipVotes = 0;
  393. let shouldSkip = false;
  394. async.waterfall([
  395. (next) => {
  396. stations.getStation(stationId, next);
  397. },
  398. (station, next) => {
  399. if (!station) return next('Station not found.');
  400. stations.canUserViewStation(station, session.userId, (err, canView) => {
  401. if (err) return next(err);
  402. if (canView) return next(null, station);
  403. return next('Insufficient permissions.');
  404. });
  405. },
  406. (station, next) => {
  407. if (!station.currentSong) return next('There is currently no song to skip.');
  408. if (station.currentSong.skipVotes.indexOf(session.userId) !== -1) return next('You have already voted to skip this song.');
  409. next(null, station);
  410. },
  411. (station, next) => {
  412. db.models.station.updateOne({_id: stationId}, {$push: {"currentSong.skipVotes": session.userId}}, next)
  413. },
  414. (res, next) => {
  415. stations.updateStation(stationId, next);
  416. },
  417. (station, next) => {
  418. if (!station) return next('Station not found.');
  419. next(null, station);
  420. },
  421. (station, next) => {
  422. skipVotes = station.currentSong.skipVotes.length;
  423. utils.getRoomSockets(`station.${stationId}`).then(sockets => {
  424. next(null, sockets);
  425. }).catch(next);
  426. },
  427. (sockets, next) => {
  428. if (sockets.length <= skipVotes) shouldSkip = true;
  429. next();
  430. }
  431. ], async (err, station) => {
  432. if (err) {
  433. err = await utils.getError(err);
  434. logger.error("STATIONS_VOTE_SKIP", `Vote skipping station "${stationId}" failed. "${err}"`);
  435. return cb({'status': 'failure', 'message': err});
  436. }
  437. logger.success("STATIONS_VOTE_SKIP", `Vote skipping "${stationId}" successful.`);
  438. cache.pub('station.voteSkipSong', stationId);
  439. cb({ status: 'success', message: 'Successfully voted to skip the song.' });
  440. if (shouldSkip) stations.skipStation(stationId)();
  441. });
  442. }),
  443. /**
  444. * Force skips a station
  445. *
  446. * @param session
  447. * @param stationId - the station id
  448. * @param cb
  449. */
  450. forceSkip: hooks.ownerRequired((session, stationId, cb) => {
  451. async.waterfall([
  452. (next) => {
  453. stations.getStation(stationId, next);
  454. },
  455. (station, next) => {
  456. if (!station) return next('Station not found.');
  457. next();
  458. }
  459. ], async (err) => {
  460. if (err) {
  461. err = await utils.getError(err);
  462. logger.error("STATIONS_FORCE_SKIP", `Force skipping station "${stationId}" failed. "${err}"`);
  463. return cb({'status': 'failure', 'message': err});
  464. }
  465. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  466. stations.skipStation(stationId)();
  467. logger.success("STATIONS_FORCE_SKIP", `Force skipped station "${stationId}" successfully.`);
  468. return cb({'status': 'success', 'message': 'Successfully skipped station.'});
  469. });
  470. }),
  471. /**
  472. * Leaves the user's current station
  473. *
  474. * @param session
  475. * @param stationId
  476. * @param cb
  477. * @return {{ status: String, userCount: Integer }}
  478. */
  479. leave: (session, stationId, cb) => {
  480. async.waterfall([
  481. (next) => {
  482. stations.getStation(stationId, next);
  483. },
  484. (station, next) => {
  485. if (!station) return next('Station not found.');
  486. next();
  487. }
  488. ], async (err, userCount) => {
  489. if (err) {
  490. err = await utils.getError(err);
  491. logger.error("STATIONS_LEAVE", `Leaving station "${stationId}" failed. "${err}"`);
  492. return cb({'status': 'failure', 'message': err});
  493. }
  494. logger.success("STATIONS_LEAVE", `Left station "${stationId}" successfully.`);
  495. utils.socketLeaveRooms(session);
  496. delete userList[session.socketId];
  497. return cb({'status': 'success', 'message': 'Successfully left station.', userCount});
  498. });
  499. },
  500. /**
  501. * Updates a station's name
  502. *
  503. * @param session
  504. * @param stationId - the station id
  505. * @param newName - the new station name
  506. * @param cb
  507. */
  508. updateName: hooks.ownerRequired((session, stationId, newName, cb) => {
  509. async.waterfall([
  510. (next) => {
  511. db.models.station.updateOne({_id: stationId}, {$set: {name: newName}}, {runValidators: true}, next);
  512. },
  513. (res, next) => {
  514. stations.updateStation(stationId, next);
  515. }
  516. ], async (err) => {
  517. if (err) {
  518. err = await utils.getError(err);
  519. logger.error("STATIONS_UPDATE_NAME", `Updating station "${stationId}" name to "${newName}" failed. "${err}"`);
  520. return cb({'status': 'failure', 'message': err});
  521. }
  522. logger.success("STATIONS_UPDATE_NAME", `Updated station "${stationId}" name to "${newName}" successfully.`);
  523. return cb({'status': 'success', 'message': 'Successfully updated the name.'});
  524. });
  525. }),
  526. /**
  527. * Updates a station's display name
  528. *
  529. * @param session
  530. * @param stationId - the station id
  531. * @param newDisplayName - the new station display name
  532. * @param cb
  533. */
  534. updateDisplayName: hooks.ownerRequired((session, stationId, newDisplayName, cb) => {
  535. async.waterfall([
  536. (next) => {
  537. db.models.station.updateOne({_id: stationId}, {$set: {displayName: newDisplayName}}, {runValidators: true}, next);
  538. },
  539. (res, next) => {
  540. stations.updateStation(stationId, next);
  541. }
  542. ], async (err) => {
  543. if (err) {
  544. err = await utils.getError(err);
  545. logger.error("STATIONS_UPDATE_DISPLAY_NAME", `Updating station "${stationId}" displayName to "${newDisplayName}" failed. "${err}"`);
  546. return cb({'status': 'failure', 'message': err});
  547. }
  548. logger.success("STATIONS_UPDATE_DISPLAY_NAME", `Updated station "${stationId}" displayName to "${newDisplayName}" successfully.`);
  549. return cb({'status': 'success', 'message': 'Successfully updated the display name.'});
  550. });
  551. }),
  552. /**
  553. * Updates a station's description
  554. *
  555. * @param session
  556. * @param stationId - the station id
  557. * @param newDescription - the new station description
  558. * @param cb
  559. */
  560. updateDescription: hooks.ownerRequired((session, stationId, newDescription, cb) => {
  561. async.waterfall([
  562. (next) => {
  563. db.models.station.updateOne({_id: stationId}, {$set: {description: newDescription}}, {runValidators: true}, next);
  564. },
  565. (res, next) => {
  566. stations.updateStation(stationId, next);
  567. }
  568. ], async (err) => {
  569. if (err) {
  570. err = await utils.getError(err);
  571. logger.error("STATIONS_UPDATE_DESCRIPTION", `Updating station "${stationId}" description to "${newDescription}" failed. "${err}"`);
  572. return cb({'status': 'failure', 'message': err});
  573. }
  574. logger.success("STATIONS_UPDATE_DESCRIPTION", `Updated station "${stationId}" description to "${newDescription}" successfully.`);
  575. return cb({'status': 'success', 'message': 'Successfully updated the description.'});
  576. });
  577. }),
  578. /**
  579. * Updates a station's privacy
  580. *
  581. * @param session
  582. * @param stationId - the station id
  583. * @param newPrivacy - the new station privacy
  584. * @param cb
  585. */
  586. updatePrivacy: hooks.ownerRequired((session, stationId, newPrivacy, cb) => {
  587. async.waterfall([
  588. (next) => {
  589. db.models.station.updateOne({_id: stationId}, {$set: {privacy: newPrivacy}}, {runValidators: true}, next);
  590. },
  591. (res, next) => {
  592. stations.updateStation(stationId, next);
  593. }
  594. ], async (err) => {
  595. if (err) {
  596. err = await utils.getError(err);
  597. logger.error("STATIONS_UPDATE_PRIVACY", `Updating station "${stationId}" privacy to "${newPrivacy}" failed. "${err}"`);
  598. return cb({'status': 'failure', 'message': err});
  599. }
  600. logger.success("STATIONS_UPDATE_PRIVACY", `Updated station "${stationId}" privacy to "${newPrivacy}" successfully.`);
  601. return cb({'status': 'success', 'message': 'Successfully updated the privacy.'});
  602. });
  603. }),
  604. /**
  605. * Updates a station's genres
  606. *
  607. * @param session
  608. * @param stationId - the station id
  609. * @param newGenres - the new station genres
  610. * @param cb
  611. */
  612. updateGenres: hooks.ownerRequired((session, stationId, newGenres, cb) => {
  613. async.waterfall([
  614. (next) => {
  615. db.models.station.updateOne({_id: stationId}, {$set: {genres: newGenres}}, {runValidators: true}, next);
  616. },
  617. (res, next) => {
  618. stations.updateStation(stationId, next);
  619. }
  620. ], async (err) => {
  621. if (err) {
  622. err = await utils.getError(err);
  623. logger.error("STATIONS_UPDATE_GENRES", `Updating station "${stationId}" genres to "${newGenres}" failed. "${err}"`);
  624. return cb({'status': 'failure', 'message': err});
  625. }
  626. logger.success("STATIONS_UPDATE_GENRES", `Updated station "${stationId}" genres to "${newGenres}" successfully.`);
  627. return cb({'status': 'success', 'message': 'Successfully updated the genres.'});
  628. });
  629. }),
  630. /**
  631. * Updates a station's blacklisted genres
  632. *
  633. * @param session
  634. * @param stationId - the station id
  635. * @param newBlacklistedGenres - the new station blacklisted genres
  636. * @param cb
  637. */
  638. updateBlacklistedGenres: hooks.ownerRequired((session, stationId, newBlacklistedGenres, cb) => {
  639. async.waterfall([
  640. (next) => {
  641. db.models.station.updateOne({_id: stationId}, {$set: {blacklistedGenres: newBlacklistedGenres}}, {runValidators: true}, next);
  642. },
  643. (res, next) => {
  644. stations.updateStation(stationId, next);
  645. }
  646. ], async (err) => {
  647. if (err) {
  648. err = await utils.getError(err);
  649. logger.error("STATIONS_UPDATE_BLACKLISTED_GENRES", `Updating station "${stationId}" blacklisted genres to "${newBlacklistedGenres}" failed. "${err}"`);
  650. return cb({'status': 'failure', 'message': err});
  651. }
  652. logger.success("STATIONS_UPDATE_BLACKLISTED_GENRES", `Updated station "${stationId}" blacklisted genres to "${newBlacklistedGenres}" successfully.`);
  653. return cb({'status': 'success', 'message': 'Successfully updated the blacklisted genres.'});
  654. });
  655. }),
  656. /**
  657. * Updates a station's party mode
  658. *
  659. * @param session
  660. * @param stationId - the station id
  661. * @param newPartyMode - the new station party mode
  662. * @param cb
  663. */
  664. updatePartyMode: hooks.ownerRequired((session, stationId, newPartyMode, cb) => {
  665. async.waterfall([
  666. (next) => {
  667. stations.getStation(stationId, next);
  668. },
  669. (station, next) => {
  670. if (!station) return next('Station not found.');
  671. if (station.partyMode === newPartyMode) return next('The party mode was already ' + ((newPartyMode) ? 'enabled.' : 'disabled.'));
  672. db.models.station.updateOne({_id: stationId}, {$set: {partyMode: newPartyMode}}, {runValidators: true}, next);
  673. },
  674. (res, next) => {
  675. stations.updateStation(stationId, next);
  676. }
  677. ], async (err) => {
  678. if (err) {
  679. err = await utils.getError(err);
  680. logger.error("STATIONS_UPDATE_PARTY_MODE", `Updating station "${stationId}" party mode to "${newPartyMode}" failed. "${err}"`);
  681. return cb({'status': 'failure', 'message': err});
  682. }
  683. logger.success("STATIONS_UPDATE_PARTY_MODE", `Updated station "${stationId}" party mode to "${newPartyMode}" successfully.`);
  684. cache.pub('station.updatePartyMode', {stationId: stationId, partyMode: newPartyMode});
  685. stations.skipStation(stationId)();
  686. return cb({'status': 'success', 'message': 'Successfully updated the party mode.'});
  687. });
  688. }),
  689. /**
  690. * Pauses a station
  691. *
  692. * @param session
  693. * @param stationId - the station id
  694. * @param cb
  695. */
  696. pause: hooks.ownerRequired((session, stationId, cb) => {
  697. async.waterfall([
  698. (next) => {
  699. stations.getStation(stationId, next);
  700. },
  701. (station, next) => {
  702. if (!station) return next('Station not found.');
  703. if (station.paused) return next('That station was already paused.');
  704. db.models.station.updateOne({_id: stationId}, {$set: {paused: true, pausedAt: Date.now()}}, next);
  705. },
  706. (res, next) => {
  707. stations.updateStation(stationId, next);
  708. }
  709. ], async (err) => {
  710. if (err) {
  711. err = await utils.getError(err);
  712. logger.error("STATIONS_PAUSE", `Pausing station "${stationId}" failed. "${err}"`);
  713. return cb({'status': 'failure', 'message': err});
  714. }
  715. logger.success("STATIONS_PAUSE", `Paused station "${stationId}" successfully.`);
  716. cache.pub('station.pause', stationId);
  717. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  718. return cb({'status': 'success', 'message': 'Successfully paused.'});
  719. });
  720. }),
  721. /**
  722. * Resumes a station
  723. *
  724. * @param session
  725. * @param stationId - the station id
  726. * @param cb
  727. */
  728. resume: hooks.ownerRequired((session, stationId, cb) => {
  729. async.waterfall([
  730. (next) => {
  731. stations.getStation(stationId, next);
  732. },
  733. (station, next) => {
  734. if (!station) return next('Station not found.');
  735. if (!station.paused) return next('That station is not paused.');
  736. station.timePaused += (Date.now() - station.pausedAt);
  737. db.models.station.updateOne({_id: stationId}, {$set: {paused: false}, $inc: {timePaused: Date.now() - station.pausedAt}}, next);
  738. },
  739. (res, next) => {
  740. stations.updateStation(stationId, next);
  741. }
  742. ], async (err) => {
  743. if (err) {
  744. err = await utils.getError(err);
  745. logger.error("STATIONS_RESUME", `Resuming station "${stationId}" failed. "${err}"`);
  746. return cb({'status': 'failure', 'message': err});
  747. }
  748. logger.success("STATIONS_RESUME", `Resuming station "${stationId}" successfully.`);
  749. cache.pub('station.resume', stationId);
  750. return cb({'status': 'success', 'message': 'Successfully resumed.'});
  751. });
  752. }),
  753. /**
  754. * Removes a station
  755. *
  756. * @param session
  757. * @param stationId - the station id
  758. * @param cb
  759. */
  760. remove: hooks.ownerRequired((session, stationId, cb) => {
  761. async.waterfall([
  762. (next) => {
  763. db.models.station.deleteOne({ _id: stationId }, err => next(err));
  764. },
  765. (next) => {
  766. cache.hdel('stations', stationId, err => next(err));
  767. }
  768. ], async (err) => {
  769. if (err) {
  770. err = await utils.getError(err);
  771. logger.error("STATIONS_REMOVE", `Removing station "${stationId}" failed. "${err}"`);
  772. return cb({ 'status': 'failure', 'message': err });
  773. }
  774. logger.success("STATIONS_REMOVE", `Removing station "${stationId}" successfully.`);
  775. cache.pub('station.remove', stationId);
  776. return cb({ 'status': 'success', 'message': 'Successfully removed.' });
  777. });
  778. }),
  779. /**
  780. * Create a station
  781. *
  782. * @param session
  783. * @param data - the station data
  784. * @param cb
  785. */
  786. create: hooks.loginRequired((session, data, cb) => {
  787. data.name = data.name.toLowerCase();
  788. let blacklist = ["country", "edm", "musare", "hip-hop", "rap", "top-hits", "todays-hits", "old-school", "christmas", "about", "support", "staff", "help", "news", "terms", "privacy", "profile", "c", "community", "tos", "login", "register", "p", "official", "o", "trap", "faq", "team", "donate", "buy", "shop", "forums", "explore", "settings", "admin", "auth", "reset_password"];
  789. async.waterfall([
  790. (next) => {
  791. if (!data) return next('Invalid data.');
  792. next();
  793. },
  794. (next) => {
  795. db.models.station.findOne({ $or: [{name: data.name}, {displayName: new RegExp(`^${data.displayName}$`, 'i')}] }, next);
  796. },
  797. (station, next) => {
  798. if (station) return next('A station with that name or display name already exists.');
  799. const { name, displayName, description, genres, playlist, type, blacklistedGenres } = data;
  800. if (type === 'official') {
  801. db.models.user.findOne({_id: session.userId}, (err, user) => {
  802. if (err) return next(err);
  803. if (!user) return next('User not found.');
  804. if (user.role !== 'admin') return next('Admin required.');
  805. db.models.station.create({
  806. name,
  807. displayName,
  808. description,
  809. type,
  810. privacy: 'private',
  811. playlist,
  812. genres,
  813. blacklistedGenres,
  814. currentSong: stations.defaultSong
  815. }, next);
  816. });
  817. } else if (type === 'community') {
  818. if (blacklist.indexOf(name) !== -1) return next('That name is blacklisted. Please use a different name.');
  819. db.models.station.create({
  820. name,
  821. displayName,
  822. description,
  823. type,
  824. privacy: 'private',
  825. owner: session.userId,
  826. queue: [],
  827. currentSong: null
  828. }, next);
  829. }
  830. }
  831. ], async (err, station) => {
  832. if (err) {
  833. err = await utils.getError(err);
  834. logger.error("STATIONS_CREATE", `Creating station failed. "${err}"`);
  835. return cb({'status': 'failure', 'message': err});
  836. }
  837. logger.success("STATIONS_CREATE", `Created station "${station._id}" successfully.`);
  838. cache.pub('station.create', station._id);
  839. return cb({'status': 'success', 'message': 'Successfully created station.'});
  840. });
  841. }),
  842. /**
  843. * Adds song to station queue
  844. *
  845. * @param session
  846. * @param stationId - the station id
  847. * @param songId - the song id
  848. * @param cb
  849. */
  850. addToQueue: hooks.loginRequired((session, stationId, songId, cb) => {
  851. async.waterfall([
  852. (next) => {
  853. stations.getStation(stationId, next);
  854. },
  855. (station, next) => {
  856. if (!station) return next('Station not found.');
  857. if (station.locked) {
  858. db.models.user.findOne({ _id: session.userId }, (err, user) => {
  859. if (user.role !== 'admin' && station.owner !== session.userId) return next('Only owners and admins can add songs to a locked queue.');
  860. else return next(null, station);
  861. });
  862. } else {
  863. return next(null, station);
  864. }
  865. },
  866. (station, next) => {
  867. if (station.type !== 'community') return next('That station is not a community station.');
  868. stations.canUserViewStation(station, session.userId, (err, canView) => {
  869. if (err) return next(err);
  870. if (canView) return next(null, station);
  871. return next('Insufficient permissions.');
  872. });
  873. },
  874. (station, next) => {
  875. if (station.currentSong && station.currentSong.songId === songId) return next('That song is currently playing.');
  876. async.each(station.queue, (queueSong, next) => {
  877. if (queueSong.songId === songId) return next('That song is already in the queue.');
  878. next();
  879. }, (err) => {
  880. next(err, station);
  881. });
  882. },
  883. (station, next) => {
  884. songs.getSong(songId, (err, song) => {
  885. if (!err && song) return next(null, song, station);
  886. utils.getSongFromYouTube(songId, (song) => {
  887. song.artists = [];
  888. song.skipDuration = 0;
  889. song.likes = -1;
  890. song.dislikes = -1;
  891. song.thumbnail = "empty";
  892. song.explicit = false;
  893. next(null, song, station);
  894. });
  895. });
  896. },
  897. (song, station, next) => {
  898. let queue = station.queue;
  899. song.requestedBy = session.userId;
  900. queue.push(song);
  901. let totalDuration = 0;
  902. queue.forEach((song) => {
  903. totalDuration += song.duration;
  904. });
  905. if (totalDuration >= 3600 * 3) return next('The max length of the queue is 3 hours.');
  906. next(null, song, station);
  907. },
  908. (song, station, next) => {
  909. let queue = station.queue;
  910. if (queue.length === 0) return next(null, song, station);
  911. let totalDuration = 0;
  912. const userId = queue[queue.length - 1].requestedBy;
  913. station.queue.forEach((song) => {
  914. if (userId === song.requestedBy) {
  915. totalDuration += song.duration;
  916. }
  917. });
  918. if(totalDuration >= 900) return next('The max length of songs per user is 15 minutes.');
  919. next(null, song, station);
  920. },
  921. (song, station, next) => {
  922. let queue = station.queue;
  923. if (queue.length === 0) return next(null, song);
  924. let totalSongs = 0;
  925. const userId = queue[queue.length - 1].requestedBy;
  926. queue.forEach((song) => {
  927. if (userId === song.requestedBy) {
  928. totalSongs++;
  929. }
  930. });
  931. if (totalSongs <= 2) return next(null, song);
  932. if (totalSongs > 3) return next('The max amount of songs per user is 3, and only 2 in a row is allowed.');
  933. if (queue[queue.length - 2].requestedBy !== userId || queue[queue.length - 3] !== userId) return next('The max amount of songs per user is 3, and only 2 in a row is allowed.');
  934. next(null, song);
  935. },
  936. (song, next) => {
  937. db.models.station.updateOne({_id: stationId}, {$push: {queue: song}}, {runValidators: true}, next);
  938. },
  939. (res, next) => {
  940. stations.updateStation(stationId, next);
  941. }
  942. ], async (err, station) => {
  943. if (err) {
  944. err = await utils.getError(err);
  945. logger.error("STATIONS_ADD_SONG_TO_QUEUE", `Adding song "${songId}" to station "${stationId}" queue failed. "${err}"`);
  946. return cb({'status': 'failure', 'message': err});
  947. }
  948. logger.success("STATIONS_ADD_SONG_TO_QUEUE", `Added song "${songId}" to station "${stationId}" successfully.`);
  949. cache.pub('station.queueUpdate', stationId);
  950. return cb({'status': 'success', 'message': 'Successfully added song to queue.'});
  951. });
  952. }),
  953. /**
  954. * Removes song from station queue
  955. *
  956. * @param session
  957. * @param stationId - the station id
  958. * @param songId - the song id
  959. * @param cb
  960. */
  961. removeFromQueue: hooks.ownerRequired((session, stationId, songId, cb) => {
  962. async.waterfall([
  963. (next) => {
  964. if (!songId) return next('Invalid song id.');
  965. stations.getStation(stationId, next);
  966. },
  967. (station, next) => {
  968. if (!station) return next('Station not found.');
  969. if (station.type !== 'community') return next('Station is not a community station.');
  970. async.each(station.queue, (queueSong, next) => {
  971. if (queueSong.songId === songId) return next(true);
  972. next();
  973. }, (err) => {
  974. if (err === true) return next();
  975. next('Song is not currently in the queue.');
  976. });
  977. },
  978. (next) => {
  979. db.models.station.updateOne({_id: stationId}, {$pull: {queue: {songId: songId}}}, next);
  980. },
  981. (res, next) => {
  982. stations.updateStation(stationId, next);
  983. }
  984. ], async (err, station) => {
  985. if (err) {
  986. err = await utils.getError(err);
  987. logger.error("STATIONS_REMOVE_SONG_TO_QUEUE", `Removing song "${songId}" from station "${stationId}" queue failed. "${err}"`);
  988. return cb({'status': 'failure', 'message': err});
  989. }
  990. logger.success("STATIONS_REMOVE_SONG_TO_QUEUE", `Removed song "${songId}" from station "${stationId}" successfully.`);
  991. cache.pub('station.queueUpdate', stationId);
  992. return cb({'status': 'success', 'message': 'Successfully removed song from queue.'});
  993. });
  994. }),
  995. /**
  996. * Gets the queue from a station
  997. *
  998. * @param session
  999. * @param stationId - the station id
  1000. * @param cb
  1001. */
  1002. getQueue: (session, stationId, cb) => {
  1003. async.waterfall([
  1004. (next) => {
  1005. stations.getStation(stationId, next);
  1006. },
  1007. (station, next) => {
  1008. if (!station) return next('Station not found.');
  1009. if (station.type !== 'community') return next('Station is not a community station.');
  1010. next(null, station);
  1011. },
  1012. (station, next) => {
  1013. stations.canUserViewStation(station, session.userId, (err, canView) => {
  1014. if (err) return next(err);
  1015. if (canView) return next(null, station);
  1016. return next('Insufficient permissions.');
  1017. });
  1018. }
  1019. ], async (err, station) => {
  1020. if (err) {
  1021. err = await utils.getError(err);
  1022. logger.error("STATIONS_GET_QUEUE", `Getting queue for station "${stationId}" failed. "${err}"`);
  1023. return cb({'status': 'failure', 'message': err});
  1024. }
  1025. logger.success("STATIONS_GET_QUEUE", `Got queue for station "${stationId}" successfully.`);
  1026. return cb({'status': 'success', 'message': 'Successfully got queue.', queue: station.queue});
  1027. });
  1028. },
  1029. /**
  1030. * Selects a private playlist for a station
  1031. *
  1032. * @param session
  1033. * @param stationId - the station id
  1034. * @param playlistId - the private playlist id
  1035. * @param cb
  1036. */
  1037. selectPrivatePlaylist: hooks.ownerRequired((session, stationId, playlistId, cb) => {
  1038. async.waterfall([
  1039. (next) => {
  1040. stations.getStation(stationId, next);
  1041. },
  1042. (station, next) => {
  1043. if (!station) return next('Station not found.');
  1044. if (station.type !== 'community') return next('Station is not a community station.');
  1045. if (station.privatePlaylist === playlistId) return next('That private playlist is already selected.');
  1046. db.models.playlist.findOne({_id: playlistId}, next);
  1047. },
  1048. (playlist, next) => {
  1049. if (!playlist) return next('Playlist not found.');
  1050. let currentSongIndex = (playlist.songs.length > 0) ? playlist.songs.length - 1 : 0;
  1051. db.models.station.updateOne({_id: stationId}, {$set: {privatePlaylist: playlistId, currentSongIndex: currentSongIndex}}, {runValidators: true}, next);
  1052. },
  1053. (res, next) => {
  1054. stations.updateStation(stationId, next);
  1055. }
  1056. ], async (err, station) => {
  1057. if (err) {
  1058. err = await utils.getError(err);
  1059. logger.error("STATIONS_SELECT_PRIVATE_PLAYLIST", `Selecting private playlist "${playlistId}" for station "${stationId}" failed. "${err}"`);
  1060. return cb({'status': 'failure', 'message': err});
  1061. }
  1062. logger.success("STATIONS_SELECT_PRIVATE_PLAYLIST", `Selected private playlist "${playlistId}" for station "${stationId}" successfully.`);
  1063. notifications.unschedule(`stations.nextSong?id${stationId}`);
  1064. if (!station.partyMode) stations.skipStation(stationId)();
  1065. cache.pub('privatePlaylist.selected', {playlistId, stationId});
  1066. return cb({'status': 'success', 'message': 'Successfully selected playlist.'});
  1067. });
  1068. }),
  1069. favoriteStation: hooks.loginRequired((session, stationId, cb) => {
  1070. async.waterfall([
  1071. (next) => {
  1072. stations.getStation(stationId, next);
  1073. },
  1074. (station, next) => {
  1075. if (!station) return next('Station not found.');
  1076. stations.canUserViewStation(station, session.userId, (err, canView) => {
  1077. if (err) return next(err);
  1078. if (canView) return next();
  1079. return next('Insufficient permissions.');
  1080. });
  1081. },
  1082. (next) => {
  1083. db.models.user.updateOne({ _id: session.userId }, { $addToSet: { favoriteStations: stationId } }, next);
  1084. },
  1085. (res, next) => {
  1086. if (res.nModified === 0) return next("The station was already favorited.");
  1087. next();
  1088. }
  1089. ], async (err) => {
  1090. if (err) {
  1091. err = await utils.getError(err);
  1092. logger.error("FAVORITE_STATION", `Favoriting station "${stationId}" failed. "${err}"`);
  1093. return cb({'status': 'failure', 'message': err});
  1094. }
  1095. logger.success("FAVORITE_STATION", `Favorited station "${stationId}" successfully.`);
  1096. cache.pub('user.favoritedStation', { userId: session.userId, stationId });
  1097. return cb({'status': 'success', 'message': 'Succesfully favorited station.'});
  1098. });
  1099. }),
  1100. unfavoriteStation: hooks.loginRequired((session, stationId, cb) => {
  1101. async.waterfall([
  1102. (next) => {
  1103. db.models.user.updateOne({ _id: session.userId }, { $pull: { favoriteStations: stationId } }, next);
  1104. },
  1105. (res, next) => {
  1106. if (res.nModified === 0) return next("The station wasn't favorited.");
  1107. next();
  1108. }
  1109. ], async (err) => {
  1110. if (err) {
  1111. err = await utils.getError(err);
  1112. logger.error("UNFAVORITE_STATION", `Unfavoriting station "${stationId}" failed. "${err}"`);
  1113. return cb({'status': 'failure', 'message': err});
  1114. }
  1115. logger.success("UNFAVORITE_STATION", `Unfavorited station "${stationId}" successfully.`);
  1116. cache.pub('user.unfavoritedStation', { userId: session.userId, stationId });
  1117. return cb({'status': 'success', 'message': 'Succesfully unfavorited station.'});
  1118. });
  1119. }),
  1120. };