stations.js 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. 'use strict';
  2. const async = require('async'),
  3. request = require('request'),
  4. config = require('config');
  5. const io = require('../io');
  6. const db = require('../db');
  7. const cache = require('../cache');
  8. const notifications = require('../notifications');
  9. const utils = require('../utils');
  10. const stations = require('../stations');
  11. const songs = require('../songs');
  12. const hooks = require('./hooks');
  13. cache.sub('station.locked', stationId => {
  14. io.io.to(`station.${stationId}`).emit("event:stations.locked");
  15. });
  16. cache.sub('station.unlocked', stationId => {
  17. io.io.to(`station.${stationId}`).emit("event:stations.unlocked");
  18. });
  19. cache.sub('station.pause', stationId => {
  20. io.io.to(`station.${stationId}`).emit("event:stations.pause");
  21. });
  22. cache.sub('station.resume', stationId => {
  23. stations.getStation(stationId, (err, station) => {
  24. io.io.to(`station.${stationId}`).emit("event:stations.resume", {timePaused: station.timePaused});
  25. });
  26. });
  27. cache.sub('station.create', stationId => {
  28. console.log(111);
  29. stations.initializeStation(stationId, (err, station) => {
  30. console.log(err, station);
  31. //TODO Emit to homepage and admin station page
  32. if (!err) {
  33. io.io.to('home').emit("event:stations.created", station);
  34. }
  35. });
  36. });
  37. module.exports = {
  38. /**
  39. * Get a list of all the stations
  40. *
  41. * @param session
  42. * @param cb
  43. * @return {{ status: String, stations: Array }}
  44. */
  45. index: (session, cb) => {
  46. // TODO: the logic should be a bit more personalized to the users preferred genres
  47. // and it should probably just a different cache table then 'stations'
  48. cache.hgetall('stations', (err, stations) => {
  49. if (err && err !== true) {
  50. return cb({
  51. status: 'error',
  52. message: 'An error occurred while obtaining the stations'
  53. });
  54. }
  55. let arr = [];
  56. for (let prop in stations) {
  57. arr.push(stations[prop]);
  58. }
  59. cb({ status: 'success', stations: arr });
  60. });
  61. },
  62. getPlaylist: (session, stationId, cb) => {
  63. let playlist = [];
  64. stations.getStation(stationId, (err, station) => {
  65. for (let s = 1; s < station.playlist.length; s++) {
  66. songs.getSong(station.playlist[s], (err, song) => {
  67. playlist.push(song);
  68. });
  69. }
  70. });
  71. cb({ status: 'success', data: playlist })
  72. },
  73. /**
  74. * Joins the station by its id
  75. *
  76. * @param session
  77. * @param stationId - the station id
  78. * @param cb
  79. * @return {{ status: String, userCount: Integer }}
  80. */
  81. join: (session, stationId, cb) => {
  82. stations.getStation(stationId, (err, station) => {
  83. if (err && err !== true) {
  84. return cb({ status: 'error', message: 'An error occurred while joining the station' });
  85. }
  86. if (station) {
  87. //TODO Loop through all sockets, see if socket with same session exists, and if so leave all other station rooms and join this stationRoom
  88. /*cache.client.hincrby('station.userCounts', stationId, 1, (err, userCount) => {
  89. if (err) return cb({ status: 'error', message: 'An error occurred while joining the station' });*/
  90. utils.socketJoinRoom(session.socketId, `station.${stationId}`);
  91. utils.socketJoinSongRoom(session.socketId, `song.${station.currentSong._id}`);
  92. //TODO Emit to cache, listen on cache
  93. songs.getSong(station.currentSong._id, (err, song) => {
  94. if (!err && song) {
  95. station.currentSong.likes = song.likes;
  96. station.currentSong.dislikes = song.dislikes;
  97. } else {
  98. station.currentSong.likes = -1;
  99. station.currentSong.dislikes = -1;
  100. }
  101. cb({ status: 'success', currentSong: station.currentSong, startedAt: station.startedAt, paused: station.paused, timePaused: station.timePaused });
  102. });
  103. //});
  104. }
  105. else {
  106. cb({ status: 'failure', message: `That station doesn't exist` });
  107. }
  108. });
  109. },
  110. /**
  111. * Skips the users current station
  112. *
  113. * @param session
  114. * @param stationId - the station id
  115. * @param cb
  116. * @return {{ status: String, skipCount: Integer }}
  117. */
  118. /*skip: (session, stationId, cb) => {
  119. if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });
  120. stations.getStation(stationId, (err, station) => {
  121. if (err && err !== true) {
  122. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  123. }
  124. if (station) {
  125. cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {
  126. session.skippedSong = true;
  127. if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  128. cache.hget('station.userCounts', session.stationId, (err, userCount) => {
  129. cb({ status: 'success', skipCount });
  130. });
  131. });
  132. }
  133. else {
  134. cb({ status: 'failure', message: `That station doesn't exist` });
  135. }
  136. });
  137. },*/
  138. forceSkip: hooks.adminRequired((session, stationId, cb) => {
  139. stations.getStation(stationId, (err, station) => {
  140. if (err && err !== true) {
  141. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  142. }
  143. if (station) {
  144. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  145. notifications.schedule(`stations.nextSong?id=${stationId}`, 100);
  146. }
  147. else {
  148. cb({ status: 'failure', message: `That station doesn't exist` });
  149. }
  150. });
  151. }),
  152. /**
  153. * Leaves the users current station
  154. *
  155. * @param session
  156. * @param stationId
  157. * @param cb
  158. * @return {{ status: String, userCount: Integer }}
  159. */
  160. leave: (session, stationId, cb) => {
  161. stations.getStation(stationId, (err, station) => {
  162. if (err && err !== true) {
  163. return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  164. }
  165. if (session) session.stationId = null;
  166. else if (station) {
  167. cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
  168. if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  169. utils.socketLeaveRooms(session);
  170. cb({ status: 'success', userCount });
  171. });
  172. } else {
  173. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  174. }
  175. });
  176. },
  177. lock: hooks.adminRequired((session, stationId, cb) => {
  178. stations.getStation(stationId, (err, station) => {
  179. if (err && err !== true) {
  180. return cb({ status: 'error', message: 'An error occurred while locking the station' });
  181. } else if (station) {
  182. // Add code to update Mongo and Redis
  183. cb({ status: 'success' });
  184. } else {
  185. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  186. }
  187. });
  188. }),
  189. unlock: hooks.adminRequired((session, stationId, cb) => {
  190. stations.getStation(stationId, (err, station) => {
  191. if (err && err !== true) {
  192. return cb({ status: 'error', message: 'An error occurred while unlocking the station' });
  193. } else if (station) {
  194. // Add code to update Mongo and Redis
  195. cb({ status: 'success' });
  196. } else {
  197. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  198. }
  199. });
  200. }),
  201. pause: hooks.adminRequired((session, stationId, cb) => {
  202. stations.getStation(stationId, (err, station) => {
  203. if (err && err !== true) {
  204. return cb({ status: 'error', message: 'An error occurred while pausing the station' });
  205. } else if (station) {
  206. if (!station.paused) {
  207. station.paused = true;
  208. station.pausedAt = Date.now();
  209. db.models.station.update({_id: stationId}, {$set: {paused: true, pausedAt: Date.now()}}, () => {
  210. if (err) return cb({ status: 'failure', message: 'An error occurred while pausing the station.' });
  211. stations.updateStation(stationId, () => {
  212. cache.pub('station.pause', stationId);
  213. notifications.unschedule(`stations.nextSong?id=${stationId}`);
  214. cb({ status: 'success' });
  215. });
  216. });
  217. } else {
  218. cb({ status: 'failure', message: 'That station was already paused.' });
  219. }
  220. cb({ status: 'success' });
  221. } else {
  222. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  223. }
  224. });
  225. }),
  226. resume: hooks.adminRequired((session, stationId, cb) => {
  227. stations.getStation(stationId, (err, station) => {
  228. if (err && err !== true) {
  229. return cb({ status: 'error', message: 'An error occurred while resuming the station' });
  230. } else if (station) {
  231. if (station.paused) {
  232. station.paused = false;
  233. station.timePaused += (Date.now() - station.pausedAt);
  234. db.models.station.update({_id: stationId}, {$set: {paused: false}, $inc: {timePaused: Date.now() - station.pausedAt}}, () => {
  235. stations.updateStation(stationId, (err, station) => {
  236. cache.pub('station.resume', stationId);
  237. cb({ status: 'success' });
  238. });
  239. });
  240. } else {
  241. cb({ status: 'failure', message: 'That station is not paused.' });
  242. }
  243. } else {
  244. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  245. }
  246. });
  247. }),
  248. remove: hooks.adminRequired((session, stationId, cb) => {
  249. db.models.station.remove({ _id: stationId });
  250. cache.hdel('stations', stationId, () => {
  251. return cb({ status: 'success', message: 'Station successfully removed' });
  252. });
  253. }),
  254. create: hooks.adminRequired((session, data, cb) => {
  255. async.waterfall([
  256. (next) => {
  257. return (data) ? next() : cb({ 'status': 'failure', 'message': 'Invalid data' });
  258. },
  259. // check the cache for the station
  260. (next) => cache.hget('stations', data._id, next),
  261. // if the cached version exist
  262. (station, next) => {
  263. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  264. db.models.station.findOne({ _id: data._id }, next);
  265. },
  266. (station, next) => {
  267. if (station) return next({ 'status': 'failure', 'message': 'A station with that id already exists' });
  268. const { _id, displayName, description, genres, playlist } = data;
  269. db.models.station.create({
  270. _id,
  271. displayName,
  272. description,
  273. type: "official",
  274. playlist,
  275. genres,
  276. currentSong: stations.defaultSong
  277. }, next);
  278. }
  279. ], (err, station) => {
  280. console.log(1234567);
  281. if (err) throw err;
  282. console.log(123456789);
  283. cache.pub('station.create', data._id);
  284. console.log(123456723213);
  285. return cb(null, { 'status': 'success', 'message': 'Successfully created station.' });
  286. });
  287. }),
  288. };