stations.js 15 KB

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