stations.js 9.6 KB

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