stations.js 15 KB

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