stations.js 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229
  1. 'use strict';
  2. const async = require('async');
  3. const request = require('request');
  4. const io = require('../io');
  5. const db = require('../db');
  6. const cache = require('../cache');
  7. const notifications = require('../notifications');
  8. const utils = require('../utils');
  9. /**
  10. * Loads a station into the cache, and sets up all the related logic
  11. *
  12. * @param {String} stationId - the id of the station
  13. * @param {Function} cb - gets called when this function completes
  14. */
  15. function initializeAndReturnStation (stationId, cb) {
  16. async.waterfall([
  17. // first check the cache for the station
  18. (next) => cache.hget('stations', stationId, next),
  19. // if the cached version exist
  20. (station, next) => {
  21. if (station) return next(true, station);
  22. db.models.station.find({ id: stationId }, next);
  23. },
  24. // if the station exists in the DB, add it to the cache
  25. (station, next) => {
  26. if (!station) return cb('Station by that id does not exist');
  27. station = cache.schemas.station(station);
  28. cache.hset('stations', station.id, station, (err) => next(err, station));
  29. }
  30. ], (err, station) => {
  31. if (err && err !== true) return cb(err);
  32. // get notified when the next song for this station should play, so that we can notify our sockets
  33. let notification = notifications.subscribe(`stations.nextSong?id=${station.id}`, () => {
  34. // get the station from the cache
  35. cache.hget('stations', station.id, (err, station) => {
  36. if (station) {
  37. // notify all the sockets on this station to go to the next song
  38. async.waterfall(io.sockets.clients().map((socket) => (next) => {
  39. // fetch the sockets session
  40. cache.hget('sessions', socket.sessionId, (err, session) => {
  41. if (session.stationId == station.id) {
  42. socket.emit('notification:stations.nextSong');
  43. }
  44. next();
  45. });
  46. }), (err) => {
  47. // schedule a notification to be dispatched when the next song ends
  48. notifications.schedule(`stations.nextSong?id=${station.id}`, 5000);
  49. });
  50. }
  51. // the station doesn't exist anymore, unsubscribe from it
  52. else {
  53. notifications.remove(notification);
  54. }
  55. });
  56. }, true);
  57. cb(null, station);
  58. });
  59. }
  60. module.exports = {
  61. /**
  62. * Get a list of all the stations
  63. *
  64. * @param session
  65. * @param cb
  66. * @return {{ status: String, stations: Array }}
  67. */
  68. index: (session, cb) => {
  69. // TODO: the logic should be a bit more personalized to the users preferred genres
  70. // and it should probably just a different cache table then 'stations'
  71. cache.hgetall('stations', (err, stations) => {
  72. if (err && err !== true) {
  73. return cb({
  74. status: 'error',
  75. message: 'An error occurred while obtaining the stations'
  76. });
  77. }
  78. cb({ status: 'success', stations });
  79. });
  80. },
  81. /**
  82. * Joins the station by its id
  83. *
  84. * @param session
  85. * @param stationId - the station join
  86. * @param cb
  87. * @return {{ status: String, userCount: Integer }}
  88. */
  89. join: (session, stationId, cb) => {
  90. initializeAndReturnStation(stationId, (err, station) => {
  91. if (err && err !== true) {
  92. return cb({ status: 'error', message: 'An error occurred while joining the station' });
  93. }
  94. if (station) {
  95. if (session) session.stationId = stationId;
  96. cache.client.hincrby('station.userCounts', stationId, 1, (err, userCount) => {
  97. if (err) return cb({ status: 'error', message: 'An error occurred while joining the station' });
  98. cb({ status: 'success', userCount });
  99. });
  100. }
  101. else {
  102. cb({ status: 'failure', message: `That station doesn't exist` });
  103. }
  104. });
  105. },
  106. /**
  107. * Skips the users current station
  108. *
  109. * @param session
  110. * @param cb
  111. * @return {{ status: String, skipCount: Integer }}
  112. */
  113. skip: (session, cb) => {
  114. if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });
  115. initializeAndReturnStation(stationId, (err, station) => {
  116. if (err && err !== true) {
  117. return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  118. }
  119. if (station) {
  120. cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {
  121. session.skippedSong = true;
  122. if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });
  123. cache.hget('station.userCounts', session.stationId, (err, userCount) => {
  124. cb({ status: 'success', skipCount });
  125. });
  126. });
  127. }
  128. else {
  129. cb({ status: 'failure', message: `That station doesn't exist` });
  130. }
  131. });
  132. },
  133. /**
  134. * Leaves the users current station
  135. *
  136. * @param session
  137. * @param cb
  138. * @return {{ status: String, userCount: Integer }}
  139. */
  140. leave: (session, cb) => {
  141. initializeAndReturnStation(stationId, (err, station) => {
  142. if (err && err !== true) {
  143. return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  144. }
  145. if (session) session.stationId = null;
  146. if (station) {
  147. cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
  148. if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
  149. cb({ status: 'success', userCount });
  150. });
  151. }
  152. else {
  153. cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
  154. }
  155. });
  156. },
  157. addSong: (session, station, song, cb) => {
  158. // if (!session.logged_in) return cb({ status: 'failure', message: 'You must be logged in to add a song' });
  159. const params = [
  160. 'part=snippet,contentDetails,statistics,status',
  161. `id=${encodeURIComponent(song.id)}`,
  162. `key=${config.get('apis.youtube.key')}`
  163. ].join('&');
  164. request(`https://www.googleapis.com/youtube/v3/videos?${params}`, (err, res, body) => {
  165. if (err) {
  166. console.error(err);
  167. return cb({ status: 'error', message: 'Failed to find song from youtube' });
  168. }
  169. const newSong = new db.models.song({
  170. id: json.items[0].id,
  171. title: json.items[0].snippet.title,
  172. duration: utils.convertTime(json.items[0].contentDetails.duration),
  173. thumbnail: json.items[0].snippet.thumbnails.high.url
  174. });
  175. // save the song to the database
  176. newSong.save(err => {
  177. if (err) {
  178. console.error(err);
  179. return cb({ status: 'error', message: 'Failed to save song from youtube to the database' });
  180. }
  181. stations.getStation(station).playlist.push(newSong);
  182. cb({ status: 'success', data: stations.getStation(station.playlist) });
  183. });
  184. });
  185. }
  186. };