stations.js 15 KB

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