stations.js 14 KB

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