stations.js 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423
  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. console.log(111);
  36. stations.initializeStation(stationId, (err, station) => {
  37. console.log(err, station);
  38. //TODO Emit to homepage and admin station page
  39. if (!err) {
  40. io.io.to('home').emit("event:stations.created", station);
  41. }
  42. });
  43. });
  44. module.exports = {
  45. /**
  46. * Get a list of all the stations
  47. *
  48. * @param session
  49. * @param cb
  50. * @return {{ status: String, stations: Array }}
  51. */
  52. index: (session, cb) => {
  53. // TODO: the logic should be a bit more personalized to the users preferred genres
  54. // and it should probably just a different cache table then 'stations'
  55. cache.hgetall('stations', (err, stations) => {
  56. if (err && err !== true) {
  57. return cb({
  58. status: 'error',
  59. message: 'An error occurred while obtaining the stations'
  60. });
  61. }
  62. let arr = [];
  63. for (let prop in stations) {
  64. arr.push(stations[prop]);
  65. }
  66. cb({ status: 'success', stations: arr });
  67. });
  68. },
  69. getPlaylist: (session, stationId, cb) => {
  70. let playlist = [];
  71. stations.getStation(stationId, (err, station) => {
  72. for (let s = 1; s < station.playlist.length; s++) {
  73. songs.getSong(station.playlist[s], (err, song) => {
  74. playlist.push(song);
  75. });
  76. }
  77. });
  78. cb({ status: 'success', data: playlist })
  79. },
  80. /**
  81. * Joins the station by its id
  82. *
  83. * @param session
  84. * @param stationId - the station id
  85. * @param cb
  86. * @return {{ status: String, userCount: Integer }}
  87. */
  88. join: (session, stationId, cb) => {
  89. stations.getStation(stationId, (err, station) => {
  90. if (err && err !== true) {
  91. return cb({ status: 'error', message: 'An error occurred while joining the station' });
  92. }
  93. if (station) {
  94. //TODO Loop through all sockets, see if socket with same session exists, and if so leave all other station rooms and join this stationRoom
  95. /*cache.client.hincrby('station.userCounts', stationId, 1, (err, userCount) => {
  96. if (err) return cb({ status: 'error', message: 'An error occurred while joining the station' });*/
  97. utils.socketJoinRoom(session.socketId, `station.${stationId}`);
  98. utils.socketJoinSongRoom(session.socketId, `song.${station.currentSong._id}`);
  99. //TODO Emit to cache, listen on cache
  100. songs.getSong(station.currentSong._id, (err, song) => {
  101. if (!err && song) {
  102. station.currentSong.likes = song.likes;
  103. station.currentSong.dislikes = song.dislikes;
  104. } else {
  105. station.currentSong.likes = -1;
  106. station.currentSong.dislikes = -1;
  107. }
  108. cb({ status: 'success', currentSong: station.currentSong, startedAt: station.startedAt, paused: station.paused, timePaused: station.timePaused });
  109. });
  110. //});
  111. }
  112. else {
  113. cb({ status: 'failure', message: `That station doesn't exist` });
  114. }
  115. });
  116. },
  117. /**
  118. * Skips the users current station
  119. *
  120. * @param session
  121. * @param stationId - the station id
  122. * @param cb
  123. * @return {{ status: String, skipCount: Integer }}
  124. */
  125. /*skip: (session, stationId, cb) => {
  126. if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });
  127. stations.getStation(stationId, (err, station) => {
  128. if (err && err !== true) {
  129. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  130. }
  131. if (station) {
  132. cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {
  133. session.skippedSong = true;
  134. if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  135. cache.hget('station.userCounts', session.stationId, (err, userCount) => {
  136. cb({ status: 'success', skipCount });
  137. });
  138. });
  139. }
  140. else {
  141. cb({ status: 'failure', message: `That station doesn't exist` });
  142. }
  143. });
  144. },*/
  145. forceSkip: hooks.adminRequired((session, stationId, cb) => {
  146. stations.getStation(stationId, (err, station) => {
  147. if (err && err !== true) {
  148. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  149. }
  150. if (station) {
  151. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  152. notifications.schedule(`stations.nextSong?id=${stationId}`, 100);
  153. }
  154. else {
  155. cb({ status: 'failure', message: `That station doesn't exist` });
  156. }
  157. });
  158. }),
  159. /**
  160. * Leaves the users current station
  161. *
  162. * @param session
  163. * @param stationId
  164. * @param cb
  165. * @return {{ status: String, userCount: Integer }}
  166. */
  167. leave: (session, stationId, cb) => {
  168. stations.getStation(stationId, (err, station) => {
  169. if (err && err !== true) {
  170. return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  171. }
  172. if (session) session.stationId = null;
  173. else if (station) {
  174. cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
  175. if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  176. utils.socketLeaveRooms(session);
  177. cb({ status: 'success', userCount });
  178. });
  179. } else {
  180. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  181. }
  182. });
  183. },
  184. lock: hooks.adminRequired((session, stationId, cb) => {
  185. stations.getStation(stationId, (err, station) => {
  186. if (err && err !== true) {
  187. return cb({ status: 'error', message: 'An error occurred while locking the station' });
  188. } else if (station) {
  189. // Add code to update Mongo and Redis
  190. cb({ status: 'success' });
  191. } else {
  192. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  193. }
  194. });
  195. }),
  196. unlock: hooks.adminRequired((session, stationId, cb) => {
  197. stations.getStation(stationId, (err, station) => {
  198. if (err && err !== true) {
  199. return cb({ status: 'error', message: 'An error occurred while unlocking the station' });
  200. } else if (station) {
  201. // Add code to update Mongo and Redis
  202. cb({ status: 'success' });
  203. } else {
  204. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  205. }
  206. });
  207. }),
  208. pause: hooks.adminRequired((session, stationId, cb) => {
  209. stations.getStation(stationId, (err, station) => {
  210. if (err && err !== true) {
  211. return cb({ status: 'error', message: 'An error occurred while pausing the station' });
  212. } else if (station) {
  213. if (!station.paused) {
  214. station.paused = true;
  215. station.pausedAt = Date.now();
  216. db.models.station.update({_id: stationId}, {$set: {paused: true, pausedAt: Date.now()}}, () => {
  217. if (err) return cb({ status: 'failure', message: 'An error occurred while pausing the station.' });
  218. stations.updateStation(stationId, () => {
  219. cache.pub('station.pause', stationId);
  220. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  221. cb({ status: 'success' });
  222. });
  223. });
  224. } else {
  225. cb({ status: 'failure', message: 'That station was already paused.' });
  226. }
  227. cb({ status: 'success' });
  228. } else {
  229. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  230. }
  231. });
  232. }),
  233. resume: hooks.adminRequired((session, stationId, cb) => {
  234. stations.getStation(stationId, (err, station) => {
  235. if (err && err !== true) {
  236. return cb({ status: 'error', message: 'An error occurred while resuming the station' });
  237. } else if (station) {
  238. if (station.paused) {
  239. station.paused = false;
  240. station.timePaused += (Date.now() - station.pausedAt);
  241. db.models.station.update({_id: stationId}, {$set: {paused: false}, $inc: {timePaused: Date.now() - station.pausedAt}}, () => {
  242. stations.updateStation(stationId, (err, station) => {
  243. cache.pub('station.resume', stationId);
  244. cb({ status: 'success' });
  245. });
  246. });
  247. } else {
  248. cb({ status: 'failure', message: 'That station is not paused.' });
  249. }
  250. } else {
  251. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  252. }
  253. });
  254. }),
  255. remove: hooks.adminRequired((session, stationId, cb) => {
  256. db.models.station.remove({ _id: stationId });
  257. cache.hdel('stations', stationId, () => {
  258. return cb({ status: 'success', message: 'Station successfully removed' });
  259. });
  260. }),
  261. create: hooks.adminRequired((session, data, cb) => {
  262. async.waterfall([
  263. (next) => {
  264. return (data) ? next() : cb({ 'status': 'failure', 'message': 'Invalid data' });
  265. },
  266. // check the cache for the station
  267. (next) => cache.hget('stations', data._id, next),
  268. // if the cached version exist
  269. (station, next) => {
  270. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  271. db.models.station.findOne({ _id: data._id }, next);
  272. },
  273. (station, next) => {
  274. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  275. const { _id, displayName, description, genres, playlist } = data;
  276. db.models.station.create({
  277. _id,
  278. displayName,
  279. description,
  280. type: "official",
  281. playlist,
  282. genres,
  283. currentSong: stations.defaultSong
  284. }, next);
  285. }
  286. ], (err, station) => {
  287. console.log(1234567);
  288. if (err) throw err;
  289. console.log(123456789);
  290. cache.pub('station.create', data._id);
  291. console.log(123456723213);
  292. return cb(null, { 'status': 'success', 'message': 'Successfully created station.' });
  293. });
  294. }),
  295. createCommunity: hooks.loginRequired((session, data, cb) => {
  296. async.waterfall([
  297. (next) => {
  298. return (data) ? next() : cb({ 'status': 'failure', 'message': 'Invalid data' });
  299. },
  300. // check the cache for the station
  301. (next) => cache.hget('stations', data._id, next),
  302. // if the cached version exist
  303. (station, next) => {
  304. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  305. db.models.station.findOne({ _id: data._id }, next);
  306. },
  307. (station, next) => {
  308. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  309. const { _id, displayName, description } = data;
  310. db.models.station.create({
  311. _id,
  312. displayName,
  313. description,
  314. type: "community",
  315. queue: [],
  316. currentSong: stations.defaultSong
  317. }, next);
  318. }
  319. ], (err, station) => {
  320. if (err) throw err;
  321. cache.pub('station.create', data._id);
  322. return cb(null, { 'status': 'success', 'message': 'Successfully created station.' });
  323. });
  324. }),
  325. addToQueue: hooks.loginRequired((session, stationId, songId, cb, userId) => {
  326. stations.getStation(stationId, (err, station) => {
  327. if (err) return cb(err);
  328. if (station.type === 'community') {
  329. let has = false;
  330. station.queue.forEach((queueSong) => {
  331. if (queueSong.songId === songId) {
  332. has = true;
  333. }
  334. });
  335. if (has) return cb({'status': 'failure', 'message': 'That song has already been added to the queue.'});
  336. db.models.update({_id: stationId}, {$push: {queue: {_id: songId, title: "Title", duration: 100, requestedBy: userId}}}, (err) => {
  337. if (err) return cb({'status': 'failure', 'message': 'Something went wrong.'});
  338. stations.updateStation(stationId, (err, station) => {
  339. if (err) return cb(err);
  340. cache.pub('station.queueUpdate', stationId);
  341. });
  342. });
  343. } else cb({'status': 'failure', 'message': 'That station is not a community station.'});
  344. });
  345. }),
  346. removeFromQueue: hooks.adminRequired((session, stationId, songId, cb, userId) => {
  347. stations.getStation(stationId, (err, station) => {
  348. if (err) return cb(err);
  349. if (station.type === 'community') {
  350. let has = false;
  351. station.queue.forEach((queueSong) => {
  352. if (queueSong._id === songId) {
  353. has = true;
  354. }
  355. });
  356. if (!has) return cb({'status': 'failure', 'message': 'That song is not in the queue.'});
  357. db.models.update({_id: stationId}, {$pull: {queue: {_id: songId}}}, (err) => {
  358. if (err) return cb({'status': 'failure', 'message': 'Something went wrong.'});
  359. stations.updateStation(stationId, (err, station) => {
  360. if (err) return cb(err);
  361. cache.pub('station.queueUpdate', stationId);
  362. });
  363. });
  364. } else cb({'status': 'failure', 'message': 'That station is not a community station.'});
  365. });
  366. }),
  367. };