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