stations.js 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464
  1. 'use strict';
  2. const cache = require('./cache');
  3. const db = require('./db');
  4. const io = require('./io');
  5. const utils = require('./utils');
  6. const songs = require('./songs');
  7. const notifications = require('./notifications');
  8. const async = require('async');
  9. let skipTimeout = null;
  10. //TEMP
  11. cache.sub('station.pause', (stationId) => {
  12. notifications.remove(`stations.nextSong?id=${stationId}`);
  13. });
  14. cache.sub('station.resume', (stationId) => {
  15. module.exports.initializeStation(stationId)
  16. });
  17. cache.sub('station.queueUpdate', (stationId) => {
  18. module.exports.getStation(stationId, (err, station) => {
  19. if (!station.currentSong && station.queue.length > 0) {
  20. module.exports.initializeStation(stationId);
  21. }
  22. });
  23. });
  24. cache.sub('station.newOfficialPlaylist', (stationId) => {
  25. cache.hget("officialPlaylists", stationId, (err, playlistObj) => {
  26. if (!err && !playlistObj) {
  27. utils.emitToRoom(`station.${stationId}`, "event:newOfficialPlaylist", playlistObj.songs);
  28. }
  29. })
  30. });
  31. module.exports = {
  32. init: function(cb) {
  33. let _this = this;
  34. //TODO Add async waterfall
  35. db.models.station.find({}, (err, stations) => {
  36. if (!err) {
  37. stations.forEach((station) => {
  38. console.info("Initializing Station: " + station._id);
  39. _this.initializeStation(station._id);
  40. });
  41. cb();
  42. }
  43. });
  44. },
  45. initializeStation: function(stationId, cb) {
  46. if (typeof cb !== 'function') cb = ()=>{};
  47. let _this = this;
  48. _this.getStation(stationId, (err, station) => {
  49. if (!err) {
  50. if (station) {
  51. let notification = notifications.subscribe(`stations.nextSong?id=${station._id}`, _this.skipStation(station._id), true);
  52. if (!station.paused ) {
  53. /*if (!station.startedAt) {
  54. station.startedAt = Date.now();
  55. station.timePaused = 0;
  56. cache.hset('stations', stationId, station);
  57. }*/
  58. if (station.currentSong) {
  59. let timeLeft = ((station.currentSong.duration * 1000) - (Date.now() - station.startedAt - station.timePaused));
  60. if (isNaN(timeLeft)) timeLeft = -1;
  61. if (station.currentSong.duration * 1000 < timeLeft || timeLeft < 0) {
  62. this.skipStation(station._id)((err, station) => {
  63. cb(err, station);
  64. });
  65. } else {
  66. notifications.schedule(`stations.nextSong?id=${station._id}`, timeLeft);
  67. cb(null, station);
  68. }
  69. } else {
  70. _this.skipStation(station._id)((err, station) => {
  71. cb(err, station);
  72. });
  73. }
  74. } else {
  75. notifications.unschedule(`stations.nextSong?id${station._id}`);
  76. cb(null, station);
  77. }
  78. } else cb("Station not found");
  79. } else cb(err);
  80. });
  81. },
  82. calculateSongForStation: function(station, cb) {
  83. let _this = this;
  84. let songList = [];
  85. async.waterfall([
  86. (next) => {
  87. let genresDone = [];
  88. station.genres.forEach((genre) => {
  89. db.models.song.find({genres: genre}, (err, songs) => {
  90. if (!err) {
  91. songs.forEach((song) => {
  92. if (songList.indexOf(song._id) === -1) {
  93. let found = false;
  94. song.genres.forEach((songGenre) => {
  95. if (station.blacklistedGenres.indexOf(songGenre) !== -1) found = true;
  96. });
  97. if (!found) {
  98. songList.push(song._id);
  99. }
  100. }
  101. });
  102. }
  103. genresDone.push(genre);
  104. if (genresDone.length === station.genres.length) next();
  105. });
  106. });
  107. },
  108. (next) => {
  109. let playlist = [];
  110. songList.forEach(function(songId) {
  111. if(station.playlist.indexOf(songId) === -1) playlist.push(songId);
  112. });
  113. station.playlist.filter((songId) => {
  114. if (songList.indexOf(songId) !== -1) playlist.push(songId);
  115. });
  116. _this.calculateOfficialPlaylistList(station._id, playlist, () => {
  117. next(null, playlist);
  118. });
  119. },
  120. (playlist, next) => {
  121. db.models.station.update({_id: station._id}, {$set: {playlist: playlist}}, (err) => {
  122. _this.updateStation(station._id, () => {
  123. next(err, playlist);
  124. });
  125. });
  126. }
  127. ], (err, newPlaylist) => {
  128. cb(err, newPlaylist);
  129. });
  130. },
  131. // Attempts to get the station from Redis. If it's not in Redis, get it from Mongo and add it to Redis.
  132. getStation: function(stationId, cb) {
  133. let _this = this;
  134. async.waterfall([
  135. (next) => {
  136. cache.hget('stations', stationId, next);
  137. },
  138. (station, next) => {
  139. if (station) return next(true, station);
  140. db.models.station.findOne({ _id: stationId }, next);
  141. },
  142. (station, next) => {
  143. if (station) {
  144. if (station.type === 'official') {
  145. _this.calculateOfficialPlaylistList(station._id, station.playlist, ()=>{});
  146. }
  147. station = cache.schemas.station(station);
  148. cache.hset('stations', stationId, station);
  149. next(true, station);
  150. } else next('Station not found');
  151. },
  152. ], (err, station) => {
  153. if (err && err !== true) cb(err);
  154. cb(null, station);
  155. });
  156. },
  157. updateStation: function(stationId, cb) {
  158. let _this = this;
  159. async.waterfall([
  160. (next) => {
  161. db.models.station.findOne({ _id: stationId }, next);
  162. },
  163. (station, next) => {
  164. if (!station) return next('Station not found');
  165. if (station.type === 'official') {
  166. _this.calculateOfficialPlaylistList(station._id, station.playlist, ()=>{});
  167. }
  168. cache.hset('stations', stationId, station, (err) => {
  169. if (err) return next(err);
  170. next(null, station);
  171. });
  172. }
  173. ], (err, station) => {
  174. if (err && err !== true) cb(err);
  175. cb(null, station);
  176. });
  177. },
  178. calculateOfficialPlaylistList: (stationId, songList, cb) => {
  179. let lessInfoPlaylist = [];
  180. function getSongInfo(index) {
  181. if (songList.length > index) {
  182. songs.getSong(songList[index], (err, song) => {
  183. if (!err && song) {
  184. let newSong = {
  185. _id: song._id,
  186. title: song.title,
  187. artists: song.artists,
  188. duration: song.duration
  189. };
  190. lessInfoPlaylist.push(newSong);
  191. }
  192. getSongInfo(index + 1);
  193. })
  194. } else {
  195. cache.hset("officialPlaylists", stationId, cache.schemas.officialPlaylist(stationId, lessInfoPlaylist), () => {
  196. cache.pub("station.newOfficialPlaylist", stationId);
  197. cb();
  198. });
  199. }
  200. }
  201. getSongInfo(0);
  202. },
  203. skipStation: function(stationId) {
  204. let _this = this;
  205. return (cb) => {
  206. if (typeof cb !== 'function') cb = ()=>{};
  207. _this.getStation(stationId, (err, station) => {
  208. if (station) {
  209. // notify all the sockets on this station to go to the next song
  210. async.waterfall([
  211. (next) => {
  212. if (station.type === "official") {
  213. if (station.playlist.length > 0) {
  214. function setCurrentSong() {
  215. if (station.currentSongIndex < station.playlist.length - 1) {
  216. songs.getSong(station.playlist[station.currentSongIndex + 1], (err, song) => {
  217. if (!err) {
  218. let $set = {};
  219. $set.currentSong = {
  220. _id: song._id,
  221. title: song.title,
  222. artists: song.artists,
  223. duration: song.duration,
  224. likes: song.likes,
  225. dislikes: song.dislikes,
  226. skipDuration: song.skipDuration,
  227. thumbnail: song.thumbnail
  228. };
  229. $set.startedAt = Date.now();
  230. $set.timePaused = 0;
  231. $set.currentSongIndex = station.currentSongIndex + 1;
  232. next(null, $set);
  233. } else {
  234. db.models.station.update({ _id: station._id }, { $inc: { currentSongIndex: 1 } }, (err) => {
  235. _this.updateStation(station._id, () => {
  236. setCurrentSong();
  237. });
  238. });
  239. }
  240. });
  241. } else {
  242. db.models.station.update({_id: station._id}, {$set: {currentSongIndex: 0}}, (err) => {
  243. _this.updateStation(station._id, (err, station) => {
  244. _this.calculateSongForStation(station, (err, newPlaylist) => {
  245. if (!err) {
  246. songs.getSong(newPlaylist[0], (err, song) => {
  247. let $set = {};
  248. if (song) {
  249. $set.currentSong = {
  250. _id: song._id,
  251. title: song.title,
  252. artists: song.artists,
  253. duration: song.duration,
  254. likes: song.likes,
  255. dislikes: song.dislikes,
  256. skipDuration: song.skipDuration,
  257. thumbnail: song.thumbnail
  258. };
  259. station.playlist = newPlaylist;
  260. } else $set.currentSong = _this.defaultSong;
  261. $set.startedAt = Date.now();
  262. $set.timePaused = 0;
  263. next(null, $set);
  264. });
  265. } else {
  266. let $set = {};
  267. $set.currentSong = _this.defaultSong;
  268. $set.startedAt = Date.now();
  269. $set.timePaused = 0;
  270. next(null, $set);
  271. }
  272. })
  273. });
  274. });
  275. }
  276. }
  277. func();
  278. } else {
  279. _this.calculateSongForStation(station, (err, playlist) => {
  280. if (!err && playlist.length === 0) {
  281. let $set = {};
  282. $set.currentSongIndex = 0;
  283. $set.currentSong = _this.defaultSong;
  284. $set.startedAt = Date.now();
  285. $set.timePaused = 0;
  286. next(null, $set);
  287. } else {
  288. songs.getSong(playlist[0], (err, song) => {
  289. let $set = {};
  290. if (!err) {
  291. $set.currentSong = {
  292. _id: song._id,
  293. title: song.title,
  294. artists: song.artists,
  295. duration: song.duration,
  296. likes: song.likes,
  297. dislikes: song.dislikes,
  298. skipDuration: song.skipDuration,
  299. thumbnail: song.thumbnail
  300. };
  301. } else {
  302. $set.currentSong = _this.defaultSong;
  303. }
  304. $set.currentSongIndex = 0;
  305. $set.startedAt = Date.now();
  306. $set.timePaused = 0;
  307. next(null, $set);
  308. });
  309. }
  310. });
  311. }
  312. } else {
  313. if (station.partyMode === true) if (station.queue.length > 0) {
  314. db.models.station.update({ _id: stationId }, { $pull: { queue: { _id: station.queue[0]._id } } }, (err) => {
  315. if (err) return next(err);
  316. let $set = {};
  317. $set.currentSong = station.queue[0];
  318. $set.startedAt = Date.now();
  319. $set.timePaused = 0;
  320. if (station.paused) $set.pausedAt = Date.now();
  321. next(null, $set);
  322. });
  323. } else next(null, {currentSong: null});
  324. else {
  325. db.models.playlist.findOne({_id: station.privatePlaylist}, (err, playlist) => {
  326. if (err || !playlist) return next(null, {currentSong: null});
  327. playlist = playlist.songs;
  328. if (playlist.length > 0) {
  329. let $set = {};
  330. if (station.currentSongIndex < playlist.length - 1) $set.currentSongIndex = station.currentSongIndex + 1;
  331. else $set.currentSongIndex = 0;
  332. songs.getSong(playlist[$set.currentSongIndex]._id, (err, song) => {
  333. if (!err && song) {
  334. $set.currentSong = {
  335. _id: song._id,
  336. title: song.title,
  337. artists: song.artists,
  338. duration: song.duration,
  339. likes: song.likes,
  340. dislikes: song.dislikes,
  341. skipDuration: song.skipDuration,
  342. thumbnail: song.thumbnail
  343. };
  344. } else {
  345. let song = playlist[$set.currentSongIndex];
  346. $set.currentSong = {
  347. _id: song._id,
  348. title: song.title,
  349. duration: song.duration,
  350. likes: -1,
  351. dislikes: -1
  352. };
  353. }
  354. $set.startedAt = Date.now();
  355. $set.timePaused = 0;
  356. next(null, $set);
  357. });
  358. } else next(null, {currentSong: null});
  359. });
  360. }
  361. }
  362. },
  363. ($set, next) => {
  364. db.models.station.update({_id: station._id}, {$set}, (err) => {
  365. _this.updateStation(station._id, (err, station) => {
  366. if (station.type === 'community' && station.partyMode === true)
  367. cache.pub('station.queueUpdate', stationId);
  368. next(null, station);
  369. });
  370. });
  371. },
  372. ], (err, station) => {
  373. if (!err) {
  374. if (station.currentSong !== null && station.currentSong._id !== undefined) {
  375. station.currentSong.skipVotes = 0;
  376. }
  377. //TODO Pub/Sub this
  378. utils.emitToRoom(`station.${station._id}`, "event:songs.next", {
  379. currentSong: station.currentSong,
  380. startedAt: station.startedAt,
  381. paused: station.paused,
  382. timePaused: 0
  383. });
  384. if (station.privacy === 'public') utils.emitToRoom('home', "event:station.nextSong", station._id, station.currentSong);
  385. else {
  386. let sockets = utils.getRoomSockets('home');
  387. for (let socketId in sockets) {
  388. let socket = sockets[socketId];
  389. let session = sockets[socketId].session;
  390. if (session.sessionId) {
  391. cache.hget('sessions', session.sessionId, (err, session) => {
  392. if (!err && session) {
  393. db.models.user.findOne({_id: session.userId}, (err, user) => {
  394. if (user.role === 'admin') socket.emit("event:station.nextSong", station._id, station.currentSong);
  395. else if (station.type === "community" && station.owner === session.userId) socket.emit("event:station.nextSong", station._id, station.currentSong);
  396. });
  397. }
  398. });
  399. }
  400. }
  401. }
  402. if (station.currentSong !== null && station.currentSong._id !== undefined) {
  403. utils.socketsJoinSongRoom(utils.getRoomSockets(`station.${station._id}`), `song.${station.currentSong._id}`);
  404. if (!station.paused) {
  405. notifications.schedule(`stations.nextSong?id=${station._id}`, station.currentSong.duration * 1000);
  406. }
  407. } else {
  408. utils.socketsLeaveSongRooms(utils.getRoomSockets(`station.${station._id}`), `song.${station.currentSong._id}`);
  409. }
  410. cb(null, station);
  411. } else cb(err);
  412. });
  413. }
  414. // the station doesn't exist anymore, unsubscribe from it
  415. else {
  416. cb("Station not found.");
  417. }
  418. });
  419. }
  420. },
  421. defaultSong: {
  422. _id: '60ItHLz5WEA',
  423. title: 'Faded - Alan Walker',
  424. duration: 212,
  425. likes: -1,
  426. dislikes: -1
  427. }
  428. };