stations.js 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356
  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) {
  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. console.log(112233, stationId);
  40. if (typeof cb !== 'function') cb = ()=>{};
  41. let _this = this;
  42. _this.getStation(stationId, (err, station) => {
  43. if (!err) {
  44. if (station) {
  45. let notification = notifications.subscribe(`stations.nextSong?id=${station._id}`, () => {
  46. console.log("NOTIFICATION!!!");
  47. _this.getStation(stationId, (err, station) => {
  48. if (station) {
  49. // notify all the sockets on this station to go to the next song
  50. async.waterfall([
  51. (next) => {
  52. if (station.type === "official") {
  53. if (station.playlist.length > 0) {
  54. function func() {
  55. if (station.currentSongIndex < station.playlist.length - 1) {
  56. station.currentSongIndex++;
  57. songs.getSong(station.playlist[station.currentSongIndex], (err, song) => {
  58. if (!err) {
  59. let $set = {};
  60. $set.currentSong = {
  61. _id: song._id,
  62. title: song.title,
  63. artists: song.artists,
  64. duration: song.duration,
  65. likes: song.likes,
  66. dislikes: song.dislikes,
  67. skipDuration: song.skipDuration,
  68. thumbnail: song.thumbnail
  69. };
  70. $set.startedAt = Date.now();
  71. $set.timePaused = 0;
  72. next(null, $set);
  73. } else {
  74. db.models.station.update({_id: station._id}, {$inc: {currentSongIndex: 1}}, (err) => {
  75. _this.updateStation(station._id, () => {
  76. func();
  77. });
  78. });
  79. }
  80. });
  81. } else {
  82. db.models.station.update({_id: station._id}, {$set: {currentSongIndex: 0}}, (err) => {
  83. _this.updateStation(station._id, (err, station) => {
  84. console.log(12345678, err, station);
  85. _this.calculateSongForStation(station, (err, newPlaylist) => {
  86. console.log('New playlist: ', newPlaylist);
  87. if (!err) {
  88. songs.getSong(newPlaylist[0], (err, song) => {
  89. let $set = {};
  90. if (song) {
  91. $set.currentSong = {
  92. _id: song._id,
  93. title: song.title,
  94. artists: song.artists,
  95. duration: song.duration,
  96. likes: song.likes,
  97. dislikes: song.dislikes,
  98. skipDuration: song.skipDuration,
  99. thumbnail: song.thumbnail
  100. };
  101. station.playlist = newPlaylist;
  102. } else {
  103. $set.currentSong = _this.defaultSong;
  104. }
  105. $set.startedAt = Date.now();
  106. $set.timePaused = 0;
  107. next(null, $set);
  108. });
  109. } else {
  110. let $set = {};
  111. $set.currentSong = _this.defaultSong;
  112. $set.startedAt = Date.now();
  113. $set.timePaused = 0;
  114. next(null, $set);
  115. }
  116. })
  117. });
  118. });
  119. }
  120. }
  121. func();
  122. } else {
  123. _this.calculateSongForStation(station, (err, playlist) => {
  124. if (!err && playlist.length === 0) {
  125. let $set = {};
  126. $set.currentSongIndex = 0;
  127. $set.currentSong = _this.defaultSong;
  128. $set.startedAt = Date.now();
  129. $set.timePaused = 0;
  130. next(null, $set);
  131. } else {
  132. songs.getSong(playlist[0], (err, song) => {
  133. let $set = {};
  134. if (!err) {
  135. $set.currentSong = {
  136. _id: song._id,
  137. title: song.title,
  138. artists: song.artists,
  139. duration: song.duration,
  140. likes: song.likes,
  141. dislikes: song.dislikes,
  142. skipDuration: song.skipDuration,
  143. thumbnail: song.thumbnail
  144. };
  145. } else {
  146. $set.currentSong = _this.defaultSong;
  147. }
  148. $set.currentSongIndex = 0;
  149. $set.startedAt = Date.now();
  150. $set.timePaused = 0;
  151. next(null, $set);
  152. });
  153. }
  154. });
  155. }
  156. } else {
  157. if (station.queue.length > 0) {
  158. db.models.station.update({_id: stationId}, {$pull: {_id: station.queue[0]._id}}, (err) => {
  159. if (err) return next(err);
  160. let $set = {};
  161. $set.currentSong = {
  162. _id: station.queue[0]._id,
  163. title: station.queue[0].title,
  164. duration: station.queue[0].duration
  165. };
  166. $set.startedAt = Date.now();
  167. $set.timePaused = 0;
  168. next(null, $set);
  169. });
  170. func();
  171. } else {
  172. next(null, {currentSong: null});
  173. }
  174. }
  175. },
  176. ($set, next) => {
  177. db.models.station.update({_id: station._id}, {$set}, (err) => {
  178. _this.updateStation(station._id, (err, station) => {
  179. next(null, station);
  180. });
  181. });
  182. },
  183. ], (err, station) => {
  184. console.log(err);
  185. if (!err) {
  186. io.io.to(`station.${station._id}`).emit("event:songs.next", {
  187. currentSong: station.currentSong,
  188. startedAt: station.startedAt,
  189. paused: station.paused,
  190. timePaused: 0
  191. });
  192. if (station.currentSong !== null && station.currentSong._id !== undefined) {
  193. utils.socketsJoinSongRoom(io.io.to(`station.${station._id}`).sockets, `song.${station.currentSong._id}`);
  194. console.log("NEXT SONG!!!", station.currentSong);
  195. if (!station.paused) {
  196. notifications.schedule(`stations.nextSong?id=${station._id}`, station.currentSong.duration * 1000);
  197. }
  198. } else {
  199. console.log("22", !!(station.currentSong));
  200. utils.socketsLeaveSongRooms(io.io.to(`station.${station._id}`).sockets, `song.${station.currentSong._id}`);
  201. }
  202. }
  203. });
  204. }
  205. // the station doesn't exist anymore, unsubscribe from it
  206. else {
  207. console.log(112233445566, "REMOVE NOTIFICATION");
  208. notifications.remove(notification);
  209. }
  210. });
  211. }, true);
  212. if (!station.paused ) {
  213. /*if (!station.startedAt) {
  214. station.startedAt = Date.now();
  215. station.timePaused = 0;
  216. cache.hset('stations', stationId, station);
  217. }*/
  218. if (station.currentSong) {
  219. let timeLeft = ((station.currentSong.duration * 1000) - (Date.now() - station.startedAt - station.timePaused));
  220. if (isNaN(timeLeft)) timeLeft = -1;
  221. if (station.currentSong.duration * 1000 < timeLeft || timeLeft < 0) {
  222. console.log("Test");
  223. notifications.schedule(`stations.nextSong?id=${station._id}`, 1);
  224. } else {
  225. notifications.schedule(`stations.nextSong?id=${station._id}`, timeLeft);
  226. }
  227. } else {
  228. notifications.schedule(`stations.nextSong?id=${station._id}`, 1);
  229. }
  230. } else {
  231. notifications.unschedule(`stations.nextSong?id${station._id}`);
  232. }
  233. }
  234. }
  235. });
  236. },
  237. calculateSongForStation: function(station, cb) {
  238. let _this = this;
  239. let songList = [];
  240. async.waterfall([
  241. (next) => {
  242. let genresDone = [];
  243. station.genres.forEach((genre) => {
  244. db.models.song.find({genres: genre}, (err, songs) => {
  245. if (!err) {
  246. songs.forEach((song) => {
  247. if (songList.indexOf(song._id) === -1) songList.push(song._id);
  248. });
  249. }
  250. genresDone.push(genre);
  251. if (genresDone.length === station.genres.length) {
  252. next();
  253. }
  254. });
  255. });
  256. },
  257. (next) => {
  258. let playlist = [];
  259. songList.forEach(function(songId) {
  260. if(station.playlist.indexOf(songId) === -1) playlist.push(songId);
  261. });
  262. station.playlist.filter((songId) => {
  263. if (songList.indexOf(songId) !== -1) playlist.push(songId);
  264. });
  265. db.models.station.update({_id: station._id}, {$set: {playlist: playlist}}, (err) => {
  266. _this.updateStation(station._id, () => {
  267. next(err, playlist);
  268. });
  269. });
  270. }
  271. ], (err, newPlaylist) => {
  272. cb(err, newPlaylist);
  273. });
  274. },
  275. // Attempts to get the station from Redis. If it's not in Redis, get it from Mongo and add it to Redis.
  276. getStation: function(stationId, cb) {
  277. async.waterfall([
  278. (next) => {
  279. cache.hget('stations', stationId, next);
  280. },
  281. (station, next) => {
  282. if (station) return next(true, station);
  283. db.models.station.findOne({ _id: stationId }, next);
  284. },
  285. (station, next) => {
  286. if (station) {
  287. station = cache.schemas.station(station);
  288. console.log(1234321, stationId);
  289. cache.hset('stations', stationId, station);
  290. next(true, station);
  291. } else next('Station not found.');
  292. },
  293. ], (err, station) => {
  294. if (err && err !== true) cb(err);
  295. cb(null, station);
  296. });
  297. },
  298. updateStation: (stationId, cb) => {
  299. async.waterfall([
  300. (next) => {
  301. db.models.station.findOne({ _id: stationId }, next);
  302. },
  303. (station, next) => {
  304. if (!station) return next('Station not found.');
  305. console.log(123444321, stationId);
  306. cache.hset('stations', stationId, station, (err) => {
  307. if (err) return next(err);
  308. next(null, station);
  309. });
  310. }
  311. ], (err, station) => {
  312. if (err && err !== true) cb(err);
  313. cb(null, station);
  314. });
  315. },
  316. defaultSong: {
  317. _id: '60ItHLz5WEA',
  318. title: 'Faded',
  319. artists: ['Alan Walker'],
  320. duration: 212,
  321. skipDuration: 0,
  322. thumbnail: 'https://i.scdn.co/image/2ddde58427f632037093857ebb71a67ddbdec34b'
  323. }
  324. };