stations.js 34 KB

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