stations.js 11 KB

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