stations.js 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524
  1. 'use strict';
  2. const async = require('async'),
  3. request = require('request'),
  4. config = require('config');
  5. const io = require('../io');
  6. const db = require('../db');
  7. const cache = require('../cache');
  8. const notifications = require('../notifications');
  9. const utils = require('../utils');
  10. const stations = require('../stations');
  11. const songs = require('../songs');
  12. const hooks = require('./hooks');
  13. cache.sub('station.locked', stationId => {
  14. io.io.to(`station.${stationId}`).emit("event:stations.locked");
  15. });
  16. cache.sub('station.unlocked', stationId => {
  17. io.io.to(`station.${stationId}`).emit("event:stations.unlocked");
  18. });
  19. cache.sub('station.pause', stationId => {
  20. io.io.to(`station.${stationId}`).emit("event:stations.pause");
  21. });
  22. cache.sub('station.resume', stationId => {
  23. stations.getStation(stationId, (err, station) => {
  24. io.io.to(`station.${stationId}`).emit("event:stations.resume", {timePaused: station.timePaused});
  25. });
  26. });
  27. cache.sub('station.queueUpdate', stationId => {
  28. stations.getStation(stationId, (err, station) => {
  29. if (!err) {
  30. io.io.to(`station.${stationId}`).emit("event:queue.update", station.queue);
  31. }
  32. });
  33. });
  34. cache.sub('station.create', stationId => {
  35. stations.initializeStation(stationId, (err, station) => {
  36. //TODO Emit to admin station page
  37. io.io.to('home').emit("event:stations.created", station);
  38. });
  39. });
  40. module.exports = {
  41. /**
  42. * Get a list of all the stations
  43. *
  44. * @param session
  45. * @param cb
  46. * @return {{ status: String, stations: Array }}
  47. */
  48. index: (session, cb) => {
  49. cache.hgetall('stations', (err, stations) => {
  50. if (err && err !== true) {
  51. return cb({
  52. status: 'error',
  53. message: 'An error occurred while obtaining the stations'
  54. });
  55. }
  56. let arr = [];
  57. let done = 0;
  58. for (let prop in stations) {
  59. // TODO If community, check if on whitelist
  60. let station = stations[prop];
  61. console.log(station)
  62. if (station.privacy === 'public') add(true, station);
  63. else if (!session.sessionId) add(false);
  64. else {
  65. cache.hget('sessions', session.sessionId, (err, session) => {
  66. if (err || !session) {
  67. add(false);
  68. } else {
  69. db.models.user.findOne({_id: session.userId}, (err, user) => {
  70. if (err || !user) add(false);
  71. else if (user.role === 'admin') add(true, station);
  72. else if (station.type === 'official') add(false);
  73. else if (station.owner === session.userId) add(true, station);
  74. else add(false);
  75. });
  76. }
  77. });
  78. }
  79. }
  80. function add(add, station) {
  81. console.log("ADD!", add, station);
  82. if (add) arr.push(station);
  83. done++;
  84. if (done === Object.keys(stations).length) {
  85. console.log("DONE!", done);
  86. cb({ status: 'success', stations: arr });
  87. }
  88. }
  89. });
  90. },
  91. getPlaylist: (session, stationId, cb) => {
  92. let playlist = [];
  93. stations.getStation(stationId, (err, station) => {
  94. for (let s = 1; s < station.playlist.length; s++) {
  95. songs.getSong(station.playlist[s], (err, song) => {
  96. playlist.push(song);
  97. });
  98. }
  99. });
  100. cb({ status: 'success', data: playlist })
  101. },
  102. /**
  103. * Joins the station by its id
  104. *
  105. * @param session
  106. * @param stationId - the station id
  107. * @param cb
  108. * @return {{ status: String, userCount: Integer }}
  109. */
  110. join: (session, stationId, cb) => {
  111. stations.getStation(stationId, (err, station) => {
  112. if (err && err !== true) {
  113. return cb({ status: 'error', message: 'An error occurred while joining the station' });
  114. }
  115. if (station) {
  116. if (station.privacy !== 'private') {
  117. func();
  118. } else {
  119. // TODO If community, check if on whitelist
  120. if (!session.userId) return cb({ status: 'error', message: 'An error occurred while joining the station1' });
  121. db.models.user.findOne({_id: session.userId}, (err, user) => {
  122. if (err || !user) return cb({ status: 'error', message: 'An error occurred while joining the station2' });
  123. if (user.role === 'admin') return func();
  124. if (station.type === 'official') return cb({ status: 'error', message: 'An error occurred while joining the station3' });
  125. if (station.owner === session.userId) return func();
  126. return cb({ status: 'error', message: 'An error occurred while joining the station4' });
  127. });
  128. }
  129. function func() {
  130. utils.socketJoinRoom(session.socketId, `station.${stationId}`);
  131. if (station.currentSong) {
  132. utils.socketJoinSongRoom(session.socketId, `song.${station.currentSong._id}`);
  133. //TODO Emit to cache, listen on cache
  134. songs.getSong(station.currentSong._id, (err, song) => {
  135. if (!err && song) {
  136. station.currentSong.likes = song.likes;
  137. station.currentSong.dislikes = song.dislikes;
  138. } else {
  139. station.currentSong.likes = -1;
  140. station.currentSong.dislikes = -1;
  141. }
  142. cb({
  143. status: 'success',
  144. data: {
  145. type: station.type,
  146. currentSong: station.currentSong,
  147. startedAt: station.startedAt,
  148. paused: station.paused,
  149. timePaused: station.timePaused,
  150. description: station.description,
  151. displayName: station.displayName,
  152. privacy: station.privacy
  153. }
  154. });
  155. });
  156. } else {
  157. cb({
  158. status: 'success',
  159. data: {
  160. type: station.type,
  161. currentSong: null,
  162. startedAt: station.startedAt,
  163. paused: station.paused,
  164. timePaused: station.timePaused,
  165. description: station.description,
  166. displayName: station.displayName,
  167. privacy: station.privacy
  168. }
  169. });
  170. }
  171. }
  172. } else {
  173. cb({ status: 'failure', message: `That station doesn't exist` });
  174. }
  175. });
  176. },
  177. /**
  178. * Skips the users current station
  179. *
  180. * @param session
  181. * @param stationId - the station id
  182. * @param cb
  183. * @return {{ status: String, skipCount: Integer }}
  184. */
  185. /*skip: (session, stationId, cb) => {
  186. if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });
  187. stations.getStation(stationId, (err, station) => {
  188. if (err && err !== true) {
  189. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  190. }
  191. if (station) {
  192. cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {
  193. session.skippedSong = true;
  194. if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  195. cache.hget('station.userCounts', session.stationId, (err, userCount) => {
  196. cb({ status: 'success', skipCount });
  197. });
  198. });
  199. }
  200. else {
  201. cb({ status: 'failure', message: `That station doesn't exist` });
  202. }
  203. });
  204. },*/
  205. forceSkip: hooks.adminRequired((session, stationId, cb) => {
  206. stations.getStation(stationId, (err, station) => {
  207. if (err && err !== true) {
  208. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  209. }
  210. if (station) {
  211. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  212. //notifications.schedule(`stations.nextSong?id=${stationId}`, 100);
  213. stations.skipStation(stationId)();
  214. }
  215. else {
  216. cb({ status: 'failure', message: `That station doesn't exist` });
  217. }
  218. });
  219. }),
  220. /**
  221. * Leaves the users current station
  222. *
  223. * @param session
  224. * @param stationId
  225. * @param cb
  226. * @return {{ status: String, userCount: Integer }}
  227. */
  228. leave: (session, stationId, cb) => {
  229. stations.getStation(stationId, (err, station) => {
  230. if (err && err !== true) {
  231. return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  232. }
  233. if (session) session.stationId = null;
  234. else if (station) {
  235. cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
  236. if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  237. utils.socketLeaveRooms(session);
  238. cb({ status: 'success', userCount });
  239. });
  240. } else {
  241. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  242. }
  243. });
  244. },
  245. updateDisplayName: hooks.adminRequired((session, stationId, newDisplayName, cb) => {
  246. db.models.station.update({_id: stationId}, {$set: {displayName: newDisplayName}}, (err) => {
  247. if (err) return cb({ status: 'failure', message: 'Something went wrong when saving the station.' });
  248. stations.updateStation(stationId, () => {
  249. //TODO Pub/sub for displayName change
  250. cb({ status: 'success', message: 'Successfully updated the display name.' });
  251. })
  252. });
  253. }),
  254. updateDescription: hooks.adminRequired((session, stationId, newDescription, cb) => {
  255. db.models.station.update({_id: stationId}, {$set: {description: newDescription}}, (err) => {
  256. if (err) return cb({ status: 'failure', message: 'Something went wrong when saving the station.' });
  257. stations.updateStation(stationId, () => {
  258. //TODO Pub/sub for description change
  259. cb({ status: 'success', message: 'Successfully updated the description.' });
  260. })
  261. });
  262. }),
  263. updatePrivacy: hooks.adminRequired((session, stationId, newPrivacy, cb) => {
  264. db.models.station.update({_id: stationId}, {$set: {privacy: newPrivacy}}, (err) => {
  265. if (err) return cb({ status: 'failure', message: 'Something went wrong when saving the station.' });
  266. stations.updateStation(stationId, () => {
  267. //TODO Pub/sub for privacy change
  268. cb({ status: 'success', message: 'Successfully updated the privacy.' });
  269. })
  270. });
  271. }),
  272. pause: hooks.adminRequired((session, stationId, cb) => {
  273. stations.getStation(stationId, (err, station) => {
  274. if (err && err !== true) {
  275. return cb({ status: 'error', message: 'An error occurred while pausing the station' });
  276. } else if (station) {
  277. if (!station.paused) {
  278. station.paused = true;
  279. station.pausedAt = Date.now();
  280. db.models.station.update({_id: stationId}, {$set: {paused: true, pausedAt: Date.now()}}, () => {
  281. if (err) return cb({ status: 'failure', message: 'An error occurred while pausing the station.' });
  282. stations.updateStation(stationId, () => {
  283. cache.pub('station.pause', stationId);
  284. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  285. cb({ status: 'success' });
  286. });
  287. });
  288. } else {
  289. cb({ status: 'failure', message: 'That station was already paused.' });
  290. }
  291. cb({ status: 'success' });
  292. } else {
  293. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  294. }
  295. });
  296. }),
  297. resume: hooks.adminRequired((session, stationId, cb) => {
  298. stations.getStation(stationId, (err, station) => {
  299. if (err && err !== true) {
  300. return cb({ status: 'error', message: 'An error occurred while resuming the station' });
  301. } else if (station) {
  302. if (station.paused) {
  303. station.paused = false;
  304. station.timePaused += (Date.now() - station.pausedAt);
  305. console.log("&&&", station.timePaused, station.pausedAt, Date.now(), station.timePaused);
  306. db.models.station.update({_id: stationId}, {$set: {paused: false}, $inc: {timePaused: Date.now() - station.pausedAt}}, () => {
  307. stations.updateStation(stationId, (err, station) => {
  308. cache.pub('station.resume', stationId);
  309. cb({ status: 'success' });
  310. });
  311. });
  312. } else {
  313. cb({ status: 'failure', message: 'That station is not paused.' });
  314. }
  315. } else {
  316. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  317. }
  318. });
  319. }),
  320. remove: hooks.adminRequired((session, stationId, cb) => {
  321. db.models.station.remove({ _id: stationId });
  322. cache.hdel('stations', stationId, () => {
  323. return cb({ status: 'success', message: 'Station successfully removed' });
  324. });
  325. }),
  326. create: hooks.adminRequired((session, data, cb) => {
  327. async.waterfall([
  328. (next) => {
  329. return (data) ? next() : cb({ 'status': 'failure', 'message': 'Invalid data' });
  330. },
  331. // check the cache for the station
  332. (next) => cache.hget('stations', data._id, next),
  333. // if the cached version exist
  334. (station, next) => {
  335. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  336. db.models.station.findOne({ _id: data._id }, next);
  337. },
  338. (station, next) => {
  339. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  340. const { _id, displayName, description, genres, playlist } = data;
  341. db.models.station.create({
  342. _id,
  343. displayName,
  344. description,
  345. type: 'official',
  346. privacy: 'private',
  347. playlist,
  348. genres,
  349. currentSong: stations.defaultSong
  350. }, next);
  351. }
  352. ], (err, station) => {
  353. if (err) {console.log(err); return cb({ 'status': 'failure', 'message': 'Something went wrong.'});}
  354. cache.pub('station.create', data._id);
  355. return cb(null, { 'status': 'success', 'message': 'Successfully created station.' });
  356. });
  357. }),
  358. createCommunity: hooks.loginRequired((session, data, cb) => {
  359. async.waterfall([
  360. (next) => {
  361. return (data) ? next() : cb({ 'status': 'failure', 'message': 'Invalid data' });
  362. },
  363. // check the cache for the station
  364. (next) => cache.hget('stations', data._id, next),
  365. // if the cached version exist
  366. (station, next) => {
  367. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  368. db.models.station.findOne({ _id: data._id }, next);
  369. },
  370. (station, next) => {
  371. cache.hget('sessions', session.sessionId, (err, session) => {
  372. next(null, station, session.userId);
  373. });
  374. },
  375. (station, userId, next) => {
  376. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  377. const { _id, displayName, description } = data;
  378. db.models.station.create({
  379. _id,
  380. displayName,
  381. owner: userId,
  382. privacy: 'private',
  383. description,
  384. type: "community",
  385. queue: [],
  386. currentSong: null
  387. }, next);
  388. }
  389. ], (err, station) => {
  390. if (err) console.error(err);
  391. cache.pub('station.create', data._id);
  392. return cb({ 'status': 'success', 'message': 'Successfully created station.' });
  393. });
  394. }),
  395. addToQueue: hooks.loginRequired((session, stationId, songId, cb, userId) => {
  396. stations.getStation(stationId, (err, station) => {
  397. if (err) return cb(err);
  398. if (station.type === 'community') {
  399. let has = false;
  400. station.queue.forEach((queueSong) => {
  401. if (queueSong._id === songId) {
  402. has = true;
  403. }
  404. });
  405. if (has) return cb({'status': 'failure', 'message': 'That song has already been added to the queue.'});
  406. if (station.currentSong && station.currentSong._id === songId) return cb({'status': 'failure', 'message': 'That song is currently playing.'});
  407. songs.getSong(songId, (err, song) => {
  408. if (err) {
  409. utils.getSongFromYouTube(songId, (song) => {
  410. song.artists = [];
  411. song.skipDuration = 0;
  412. song.likes = -1;
  413. song.dislikes = -1;
  414. song.thumbnail = "empty";
  415. song.explicit = false;
  416. cont(song);
  417. });
  418. } else cont(song);
  419. function cont(song) {
  420. db.models.station.update({_id: stationId}, {$push: {queue: song}}, (err) => {
  421. console.log(err);
  422. if (err) return cb({'status': 'failure', 'message': 'Something went wrong.'});
  423. stations.updateStation(stationId, (err, station) => {
  424. if (err) return cb(err);
  425. cache.pub('station.queueUpdate', stationId);
  426. cb({'status': 'success', 'message': 'Added that song to the queue.'});
  427. });
  428. });
  429. }
  430. });
  431. } else cb({'status': 'failure', 'message': 'That station is not a community station.'});
  432. });
  433. }),
  434. removeFromQueue: hooks.adminRequired((session, stationId, songId, cb, userId) => {
  435. stations.getStation(stationId, (err, station) => {
  436. if (err) return cb(err);
  437. if (station.type === 'community') {
  438. let has = false;
  439. station.queue.forEach((queueSong) => {
  440. if (queueSong._id === songId) {
  441. has = true;
  442. }
  443. });
  444. if (!has) return cb({'status': 'failure', 'message': 'That song is not in the queue.'});
  445. db.models.update({_id: stationId}, {$pull: {queue: {songId: songId}}}, (err) => {
  446. if (err) return cb({'status': 'failure', 'message': 'Something went wrong.'});
  447. stations.updateStation(stationId, (err, station) => {
  448. if (err) return cb(err);
  449. cache.pub('station.queueUpdate', stationId);
  450. });
  451. });
  452. } else cb({'status': 'failure', 'message': 'That station is not a community station.'});
  453. });
  454. }),
  455. getQueue: hooks.adminRequired((session, stationId, cb) => {
  456. stations.getStation(stationId, (err, station) => {
  457. if (err) return cb(err);
  458. if (!station) return cb({'status': 'failure', 'message': 'Station not found.'});
  459. if (station.type === 'community') {
  460. cb({'status': 'success', queue: station.queue});
  461. } else cb({'status': 'failure', 'message': 'That station is not a community station.'});
  462. });
  463. }),
  464. };