stations.js 9.2 KB

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