stations.js 14 KB

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