stations.js 53 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149
  1. import async from "async";
  2. import { isLoginRequired, isOwnerRequired } from "./hooks";
  3. import db from "../db";
  4. import utils from "../utils";
  5. import songs from "../songs";
  6. import cache from "../cache";
  7. import notifications from "../notifications";
  8. import stations from "../stations";
  9. import activities from "../activities";
  10. // const logger = moduleManager.modules["logger"];
  11. const userList = {};
  12. const usersPerStation = {};
  13. const usersPerStationCount = {};
  14. // Temporarily disabled until the messages in console can be limited
  15. // setInterval(async () => {
  16. // const stationsCountUpdated = [];
  17. // const stationsUpdated = [];
  18. // const oldUsersPerStation = usersPerStation;
  19. // usersPerStation = {};
  20. // const oldUsersPerStationCount = usersPerStationCount;
  21. // usersPerStationCount = {};
  22. // const userModel = await db.runJob("GET_MODEL", {
  23. // modelName: "user"
  24. // });
  25. // async.each(
  26. // Object.keys(userList),
  27. // (socketId, next) => {
  28. // utils.runJob("SOCKET_FROM_SESSION", { socketId }, { isQuiet: true }).then(socket => {
  29. // const stationId = userList[socketId];
  30. // if (!socket || Object.keys(socket.rooms).indexOf(`station.${stationId}`) === -1) {
  31. // if (stationsCountUpdated.indexOf(stationId) === -1) stationsCountUpdated.push(stationId);
  32. // if (stationsUpdated.indexOf(stationId) === -1) stationsUpdated.push(stationId);
  33. // delete userList[socketId];
  34. // return next();
  35. // }
  36. // if (!usersPerStationCount[stationId]) usersPerStationCount[stationId] = 0;
  37. // usersPerStationCount[stationId] += 1;
  38. // if (!usersPerStation[stationId]) usersPerStation[stationId] = [];
  39. // return async.waterfall(
  40. // [
  41. // next => {
  42. // if (!socket.session || !socket.session.sessionId) return next("No session found.");
  43. // return cache
  44. // .runJob("HGET", {
  45. // table: "sessions",
  46. // key: socket.session.sessionId
  47. // })
  48. // .then(session => {
  49. // next(null, session);
  50. // })
  51. // .catch(next);
  52. // },
  53. // (session, next) => {
  54. // if (!session) return next("Session not found.");
  55. // return userModel.findOne({ _id: session.userId }, next);
  56. // },
  57. // (user, next) => {
  58. // if (!user) return next("User not found.");
  59. // if (usersPerStation[stationId].indexOf(user.username) !== -1)
  60. // return next("User already in the list.");
  61. // return next(null, user.username);
  62. // }
  63. // ],
  64. // (err, username) => {
  65. // if (!err) {
  66. // usersPerStation[stationId].push(username);
  67. // }
  68. // next();
  69. // }
  70. // );
  71. // });
  72. // // TODO Code to show users
  73. // },
  74. // () => {
  75. // Object.keys(usersPerStationCount).forEach(stationId => {
  76. // if (oldUsersPerStationCount[stationId] !== usersPerStationCount[stationId]) {
  77. // if (stationsCountUpdated.indexOf(stationId) === -1) stationsCountUpdated.push(stationId);
  78. // }
  79. // })
  80. //
  81. // Object.keys(usersPerStation).forEach(stationId => {
  82. // if (
  83. // _.difference(usersPerStation[stationId], oldUsersPerStation[stationId]).length > 0 ||
  84. // _.difference(oldUsersPerStation[stationId], usersPerStation[stationId]).length > 0
  85. // ) {
  86. // if (stationsUpdated.indexOf(stationId) === -1) stationsUpdated.push(stationId);
  87. // }
  88. // });
  89. //
  90. // stationsCountUpdated.forEach(stationId => {
  91. // console.log("INFO", "UPDATE_STATION_USER_COUNT", `Updating user count of ${stationId}.`);
  92. // cache.runJob("PUB", {
  93. // table: "station.updateUserCount",
  94. // value: stationId
  95. // });
  96. // });
  97. // stationsUpdated.forEach(stationId => {
  98. // console.log("INFO", "UPDATE_STATION_USER_LIST", `Updating user list of ${stationId}.`);
  99. // cache.runJob("PUB", {
  100. // table: "station.updateUsers",
  101. // value: stationId
  102. // });
  103. // });
  104. // // console.log("Userlist", usersPerStation);
  105. // }
  106. // );
  107. // }, 3000);
  108. cache.runJob("SUB", {
  109. channel: "station.updateUsers",
  110. cb: stationId => {
  111. const list = usersPerStation[stationId] || [];
  112. utils.runJob("EMIT_TO_ROOM", {
  113. room: `station.${stationId}`,
  114. args: ["event:users.updated", list]
  115. });
  116. }
  117. });
  118. cache.runJob("SUB", {
  119. channel: "station.updateUserCount",
  120. cb: stationId => {
  121. const count = usersPerStationCount[stationId] || 0;
  122. utils.runJob("EMIT_TO_ROOM", {
  123. room: `station.${stationId}`,
  124. args: ["event:userCount.updated", count]
  125. });
  126. stations.runJob("GET_STATION", { stationId }).then(async station => {
  127. if (station.privacy === "public")
  128. utils.runJob("EMIT_TO_ROOM", {
  129. room: "home",
  130. args: ["event:userCount.updated", stationId, count]
  131. });
  132. else {
  133. const sockets = await utils.runJob("GET_ROOM_SOCKETS", {
  134. room: "home"
  135. });
  136. Object.keys(sockets).forEach(socketKey => {
  137. const socket = sockets[socketKey];
  138. const { session } = socket;
  139. if (session.sessionId) {
  140. cache
  141. .runJob("HGET", {
  142. table: "sessions",
  143. key: session.sessionId
  144. })
  145. .then(session => {
  146. if (session)
  147. db.runJob("GET_MODEL", {
  148. modelName: "user"
  149. }).then(userModel =>
  150. userModel.findOne({ _id: session.userId }, (err, user) => {
  151. if (user.role === "admin")
  152. socket.emit("event:userCount.updated", stationId, count);
  153. else if (station.type === "community" && station.owner === session.userId)
  154. socket.emit("event:userCount.updated", stationId, count);
  155. })
  156. );
  157. });
  158. }
  159. });
  160. }
  161. });
  162. }
  163. });
  164. cache.runJob("SUB", {
  165. channel: "station.queueLockToggled",
  166. cb: data => {
  167. utils.runJob("EMIT_TO_ROOM", {
  168. room: `station.${data.stationId}`,
  169. args: ["event:queueLockToggled", data.locked]
  170. });
  171. }
  172. });
  173. cache.runJob("SUB", {
  174. channel: "station.updatePartyMode",
  175. cb: data => {
  176. utils.runJob("EMIT_TO_ROOM", {
  177. room: `station.${data.stationId}`,
  178. args: ["event:partyMode.updated", data.partyMode]
  179. });
  180. }
  181. });
  182. cache.runJob("SUB", {
  183. channel: "privatePlaylist.selected",
  184. cb: data => {
  185. utils.runJob("EMIT_TO_ROOM", {
  186. room: `station.${data.stationId}`,
  187. args: ["event:privatePlaylist.selected", data.playlistId]
  188. });
  189. }
  190. });
  191. cache.runJob("SUB", {
  192. channel: "station.pause",
  193. cb: stationId => {
  194. stations.runJob("GET_STATION", { stationId }).then(station => {
  195. utils.runJob("EMIT_TO_ROOM", {
  196. room: `station.${stationId}`,
  197. args: ["event:stations.pause", { pausedAt: station.pausedAt }]
  198. });
  199. });
  200. }
  201. });
  202. cache.runJob("SUB", {
  203. channel: "station.resume",
  204. cb: stationId => {
  205. stations.runJob("GET_STATION", { stationId }).then(station => {
  206. utils.runJob("EMIT_TO_ROOM", {
  207. room: `station.${stationId}`,
  208. args: ["event:stations.resume", { timePaused: station.timePaused }]
  209. });
  210. });
  211. }
  212. });
  213. cache.runJob("SUB", {
  214. channel: "station.queueUpdate",
  215. cb: stationId => {
  216. stations.runJob("GET_STATION", { stationId }).then(station => {
  217. utils.runJob("EMIT_TO_ROOM", {
  218. room: `station.${stationId}`,
  219. args: ["event:queue.update", station.queue]
  220. });
  221. });
  222. }
  223. });
  224. cache.runJob("SUB", {
  225. channel: "station.voteSkipSong",
  226. cb: stationId => {
  227. utils.runJob("EMIT_TO_ROOM", {
  228. room: `station.${stationId}`,
  229. args: ["event:song.voteSkipSong"]
  230. });
  231. }
  232. });
  233. cache.runJob("SUB", {
  234. channel: "station.remove",
  235. cb: stationId => {
  236. utils.runJob("EMIT_TO_ROOM", {
  237. room: `station.${stationId}`,
  238. args: ["event:stations.remove"]
  239. });
  240. utils.runJob("EMIT_TO_ROOM", {
  241. room: "admin.stations",
  242. args: ["event:admin.station.removed", stationId]
  243. });
  244. }
  245. });
  246. cache.runJob("SUB", {
  247. channel: "station.create",
  248. cb: async stationId => {
  249. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  250. stations.runJob("INITIALIZE_STATION", { stationId }).then(async response => {
  251. const { station } = response;
  252. station.userCount = usersPerStationCount[stationId] || 0;
  253. utils.runJob("EMIT_TO_ROOM", {
  254. room: "admin.stations",
  255. args: ["event:admin.station.added", station]
  256. });
  257. // TODO If community, check if on whitelist
  258. if (station.privacy === "public")
  259. utils.runJob("EMIT_TO_ROOM", {
  260. room: "home",
  261. args: ["event:stations.created", station]
  262. });
  263. else {
  264. const sockets = await utils.runJob("GET_ROOM_SOCKETS", {
  265. room: "home"
  266. });
  267. Object.keys(sockets).forEach(socketKey => {
  268. const socket = sockets[socketKey];
  269. const { session } = socket;
  270. if (session.sessionId) {
  271. cache
  272. .runJob("HGET", {
  273. table: "sessions",
  274. key: session.sessionId
  275. })
  276. .then(session => {
  277. if (session) {
  278. userModel.findOne({ _id: session.userId }, (err, user) => {
  279. if (user.role === "admin") socket.emit("event:stations.created", station);
  280. else if (station.type === "community" && station.owner === session.userId)
  281. socket.emit("event:stations.created", station);
  282. });
  283. }
  284. });
  285. }
  286. });
  287. }
  288. });
  289. }
  290. });
  291. export default {
  292. /**
  293. * Get a list of all the stations
  294. *
  295. * @param {object} session - user session
  296. * @param {Function} cb - callback
  297. */
  298. index: (session, cb) => {
  299. async.waterfall(
  300. [
  301. next => {
  302. cache.runJob("HGETALL", { table: "stations" }).then(stations => {
  303. next(null, stations);
  304. });
  305. },
  306. (items, next) => {
  307. const filteredStations = [];
  308. async.each(
  309. items,
  310. (station, next) => {
  311. async.waterfall(
  312. [
  313. next => {
  314. stations
  315. .runJob("CAN_USER_VIEW_STATION", {
  316. station,
  317. userId: session.userId,
  318. hideUnlisted: true
  319. })
  320. .then(exists => {
  321. next(null, exists);
  322. })
  323. .catch(next);
  324. }
  325. ],
  326. (err, exists) => {
  327. if (err) console.log(err);
  328. station.userCount = usersPerStationCount[station._id] || 0;
  329. if (exists) filteredStations.push(station);
  330. next();
  331. }
  332. );
  333. },
  334. () => next(null, filteredStations)
  335. );
  336. }
  337. ],
  338. async (err, stations) => {
  339. if (err) {
  340. err = await utils.runJob("GET_ERROR", { error: err });
  341. console.log("ERROR", "STATIONS_INDEX", `Indexing stations failed. "${err}"`);
  342. return cb({ status: "failure", message: err });
  343. }
  344. console.log("SUCCESS", "STATIONS_INDEX", `Indexing stations successful.`, false);
  345. return cb({ status: "success", stations });
  346. }
  347. );
  348. },
  349. /**
  350. * Obtains basic metadata of a station in order to format an activity
  351. *
  352. * @param {object} session - user session
  353. * @param {string} stationId - the station id
  354. * @param {Function} cb - callback
  355. */
  356. getStationForActivity: (session, stationId, cb) => {
  357. async.waterfall(
  358. [
  359. next => {
  360. stations
  361. .runJob("GET_STATION", { stationId })
  362. .then(station => {
  363. next(null, station);
  364. })
  365. .catch(next);
  366. }
  367. ],
  368. async (err, station) => {
  369. if (err) {
  370. err = await utils.runJob("GET_ERROR", { error: err });
  371. console.log(
  372. "ERROR",
  373. "STATIONS_GET_STATION_FOR_ACTIVITY",
  374. `Failed to obtain metadata of station ${stationId} for activity formatting. "${err}"`
  375. );
  376. return cb({ status: "failure", message: err });
  377. }
  378. console.log(
  379. "SUCCESS",
  380. "STATIONS_GET_STATION_FOR_ACTIVITY",
  381. `Obtained metadata of station ${stationId} for activity formatting successfully.`
  382. );
  383. return cb({
  384. status: "success",
  385. data: {
  386. title: station.displayName,
  387. thumbnail: station.currentSong ? station.currentSong.thumbnail : ""
  388. }
  389. });
  390. }
  391. );
  392. },
  393. /**
  394. * Verifies that a station exists
  395. *
  396. * @param {object} session - user session
  397. * @param {string} stationName - the station name
  398. * @param {Function} cb - callback
  399. */
  400. existsByName: (session, stationName, cb) => {
  401. async.waterfall(
  402. [
  403. next => {
  404. stations
  405. .runJob("GET_STATION_BY_NAME", { stationName })
  406. .then(station => {
  407. next(null, station);
  408. })
  409. .catch(next);
  410. },
  411. (station, next) => {
  412. if (!station) return next(null, false);
  413. return stations
  414. .runJob("CAN_USER_VIEW_STATION", {
  415. station,
  416. userId: session.userId
  417. })
  418. .then(exists => {
  419. next(null, exists);
  420. })
  421. .catch(next);
  422. }
  423. ],
  424. async (err, exists) => {
  425. if (err) {
  426. err = await utils.runJob("GET_ERROR", { error: err });
  427. console.log(
  428. "ERROR",
  429. "STATION_EXISTS_BY_NAME",
  430. `Checking if station "${stationName}" exists failed. "${err}"`
  431. );
  432. return cb({ status: "failure", message: err });
  433. }
  434. console.log(
  435. "SUCCESS",
  436. "STATION_EXISTS_BY_NAME",
  437. `Station "${stationName}" exists successfully.` /* , false */
  438. );
  439. return cb({ status: "success", exists });
  440. }
  441. );
  442. },
  443. /**
  444. * Gets the official playlist for a station
  445. *
  446. * @param {object} session - user session
  447. * @param {string} stationId - the station id
  448. * @param {Function} cb - callback
  449. */
  450. getPlaylist: (session, stationId, cb) => {
  451. async.waterfall(
  452. [
  453. next => {
  454. stations
  455. .runJob("GET_STATION", { stationId })
  456. .then(station => {
  457. next(null, station);
  458. })
  459. .catch(next);
  460. },
  461. (station, next) => {
  462. stations
  463. .runJob("CAN_USER_VIEW_STATION", {
  464. station,
  465. userId: session.userId
  466. })
  467. .then(canView => {
  468. if (canView) return next(null, station);
  469. return next("Insufficient permissions.");
  470. })
  471. .catch(err => next(err));
  472. },
  473. (station, next) => {
  474. if (!station) return next("Station not found.");
  475. if (station.type !== "official") return next("This is not an official station.");
  476. return next();
  477. },
  478. next => {
  479. cache
  480. .runJob("HGET", {
  481. table: "officialPlaylists",
  482. key: stationId
  483. })
  484. .then(playlist => {
  485. next(null, playlist);
  486. })
  487. .catch(next);
  488. },
  489. (playlist, next) => {
  490. if (!playlist) return next("Playlist not found.");
  491. return next(null, playlist);
  492. }
  493. ],
  494. async (err, playlist) => {
  495. if (err) {
  496. err = await utils.runJob("GET_ERROR", { error: err });
  497. console.log(
  498. "ERROR",
  499. "STATIONS_GET_PLAYLIST",
  500. `Getting playlist for station "${stationId}" failed. "${err}"`
  501. );
  502. return cb({ status: "failure", message: err });
  503. }
  504. console.log(
  505. "SUCCESS",
  506. "STATIONS_GET_PLAYLIST",
  507. `Got playlist for station "${stationId}" successfully.`,
  508. false
  509. );
  510. return cb({ status: "success", data: playlist.songs });
  511. }
  512. );
  513. },
  514. /**
  515. * Joins the station by its name
  516. *
  517. * @param {object} session - user session
  518. * @param {string} stationName - the station name
  519. * @param {Function} cb - callback
  520. */
  521. join: (session, stationName, cb) => {
  522. async.waterfall(
  523. [
  524. next => {
  525. stations
  526. .runJob("GET_STATION_BY_NAME", { stationName })
  527. .then(station => {
  528. next(null, station);
  529. })
  530. .catch(next);
  531. },
  532. (station, next) => {
  533. if (!station) return next("Station not found.");
  534. return stations
  535. .runJob("CAN_USER_VIEW_STATION", {
  536. station,
  537. userId: session.userId
  538. })
  539. .then(canView => {
  540. if (!canView) next("Not allowed to join station.");
  541. else next(null, station);
  542. })
  543. .catch(err => next(err));
  544. },
  545. (station, next) => {
  546. utils.runJob("SOCKET_JOIN_ROOM", {
  547. socketId: session.socketId,
  548. room: `station.${station._id}`
  549. });
  550. const data = {
  551. _id: station._id,
  552. type: station.type,
  553. currentSong: station.currentSong,
  554. startedAt: station.startedAt,
  555. paused: station.paused,
  556. timePaused: station.timePaused,
  557. pausedAt: station.pausedAt,
  558. description: station.description,
  559. displayName: station.displayName,
  560. privacy: station.privacy,
  561. locked: station.locked,
  562. partyMode: station.partyMode,
  563. owner: station.owner,
  564. privatePlaylist: station.privatePlaylist
  565. };
  566. userList[session.socketId] = station._id;
  567. next(null, data);
  568. },
  569. (data, next) => {
  570. data = JSON.parse(JSON.stringify(data));
  571. data.userCount = usersPerStationCount[data._id] || 0;
  572. data.users = usersPerStation[data._id] || [];
  573. if (!data.currentSong || !data.currentSong.title) return next(null, data);
  574. utils.runJob("SOCKET_JOIN_SONG_ROOM", {
  575. socketId: session.socketId,
  576. room: `song.${data.currentSong.songId}`
  577. });
  578. data.currentSong.skipVotes = data.currentSong.skipVotes.length;
  579. return songs
  580. .runJob("GET_SONG_FROM_ID", {
  581. songId: data.currentSong.songId
  582. })
  583. .then(response => {
  584. const { song } = response;
  585. if (song) {
  586. data.currentSong.likes = song.likes;
  587. data.currentSong.dislikes = song.dislikes;
  588. } else {
  589. data.currentSong.likes = -1;
  590. data.currentSong.dislikes = -1;
  591. }
  592. })
  593. .catch(() => {
  594. data.currentSong.likes = -1;
  595. data.currentSong.dislikes = -1;
  596. })
  597. .finally(() => {
  598. next(null, data);
  599. });
  600. }
  601. ],
  602. async (err, data) => {
  603. if (err) {
  604. err = await utils.runJob("GET_ERROR", { error: err });
  605. console.log("ERROR", "STATIONS_JOIN", `Joining station "${stationName}" failed. "${err}"`);
  606. return cb({ status: "failure", message: err });
  607. }
  608. console.log("SUCCESS", "STATIONS_JOIN", `Joined station "${data._id}" successfully.`);
  609. return cb({ status: "success", data });
  610. }
  611. );
  612. },
  613. /**
  614. * Toggles if a station is locked
  615. *
  616. * @param session
  617. * @param stationId - the station id
  618. * @param cb
  619. */
  620. toggleLock: isOwnerRequired(async (session, stationId, cb) => {
  621. const stationModel = await db.runJob("GET_MODEL", {
  622. modelName: "station"
  623. });
  624. async.waterfall(
  625. [
  626. next => {
  627. stations
  628. .runJob("GET_STATION", { stationId })
  629. .then(station => {
  630. next(null, station);
  631. })
  632. .catch(next);
  633. },
  634. (station, next) => {
  635. stationModel.updateOne({ _id: stationId }, { $set: { locked: !station.locked } }, next);
  636. },
  637. (res, next) => {
  638. stations
  639. .runJob("UPDATE_STATION", { stationId })
  640. .then(station => {
  641. next(null, station);
  642. })
  643. .catch(next);
  644. }
  645. ],
  646. async (err, station) => {
  647. if (err) {
  648. err = await utils.runJob("GET_ERROR", { error: err });
  649. console.log(
  650. "ERROR",
  651. "STATIONS_UPDATE_LOCKED_STATUS",
  652. `Toggling the queue lock for station "${stationId}" failed. "${err}"`
  653. );
  654. return cb({ status: "failure", message: err });
  655. }
  656. console.log(
  657. "SUCCESS",
  658. "STATIONS_UPDATE_LOCKED_STATUS",
  659. `Toggled the queue lock for station "${stationId}" successfully to "${station.locked}".`
  660. );
  661. cache.runJob("PUB", {
  662. channel: "station.queueLockToggled",
  663. value: {
  664. stationId,
  665. locked: station.locked
  666. }
  667. });
  668. return cb({ status: "success", data: station.locked });
  669. }
  670. );
  671. }),
  672. /**
  673. * Votes to skip a station
  674. *
  675. * @param session
  676. * @param stationId - the station id
  677. * @param cb
  678. */
  679. voteSkip: isLoginRequired(async (session, stationId, cb) => {
  680. const stationModel = await db.runJob("GET_MODEL", {
  681. modelName: "station"
  682. });
  683. let skipVotes = 0;
  684. let shouldSkip = false;
  685. async.waterfall(
  686. [
  687. next => {
  688. stations
  689. .runJob("GET_STATION", { stationId })
  690. .then(station => {
  691. next(null, station);
  692. })
  693. .catch(next);
  694. },
  695. (station, next) => {
  696. if (!station) return next("Station not found.");
  697. return stations
  698. .runJob("CAN_USER_VIEW_STATION", {
  699. station,
  700. userId: session.userId
  701. })
  702. .then(canView => {
  703. if (canView) return next(null, station);
  704. return next("Insufficient permissions.");
  705. })
  706. .catch(err => next(err));
  707. },
  708. (station, next) => {
  709. if (!station.currentSong) return next("There is currently no song to skip.");
  710. if (station.currentSong.skipVotes.indexOf(session.userId) !== -1)
  711. return next("You have already voted to skip this song.");
  712. return next(null, station);
  713. },
  714. (station, next) => {
  715. stationModel.updateOne(
  716. { _id: stationId },
  717. { $push: { "currentSong.skipVotes": session.userId } },
  718. next
  719. );
  720. },
  721. (res, next) => {
  722. stations
  723. .runJob("UPDATE_STATION", { stationId })
  724. .then(station => {
  725. next(null, station);
  726. })
  727. .catch(next);
  728. },
  729. (station, next) => {
  730. if (!station) return next("Station not found.");
  731. return next(null, station);
  732. },
  733. (station, next) => {
  734. skipVotes = station.currentSong.skipVotes.length;
  735. utils
  736. .runJob("GET_ROOM_SOCKETS", {
  737. room: `station.${stationId}`
  738. })
  739. .then(sockets => {
  740. next(null, sockets);
  741. })
  742. .catch(next);
  743. },
  744. (sockets, next) => {
  745. if (sockets.length <= skipVotes) shouldSkip = true;
  746. next();
  747. }
  748. ],
  749. async err => {
  750. if (err) {
  751. err = await utils.runJob("GET_ERROR", { error: err });
  752. console.log("ERROR", "STATIONS_VOTE_SKIP", `Vote skipping station "${stationId}" failed. "${err}"`);
  753. return cb({ status: "failure", message: err });
  754. }
  755. console.log("SUCCESS", "STATIONS_VOTE_SKIP", `Vote skipping "${stationId}" successful.`);
  756. cache.runJob("PUB", {
  757. channel: "station.voteSkipSong",
  758. value: stationId
  759. });
  760. if (shouldSkip) stations.runJob("SKIP_STATION", { stationId });
  761. return cb({
  762. status: "success",
  763. message: "Successfully voted to skip the song."
  764. });
  765. }
  766. );
  767. }),
  768. /**
  769. * Force skips a station
  770. *
  771. * @param session
  772. * @param stationId - the station id
  773. * @param cb
  774. */
  775. forceSkip: isOwnerRequired((session, stationId, cb) => {
  776. async.waterfall(
  777. [
  778. next => {
  779. stations
  780. .runJob("GET_STATION", { stationId })
  781. .then(station => {
  782. next(null, station);
  783. })
  784. .catch(next);
  785. },
  786. (station, next) => {
  787. if (!station) return next("Station not found.");
  788. return next();
  789. }
  790. ],
  791. async err => {
  792. if (err) {
  793. err = await utils.runJob("GET_ERROR", { error: err });
  794. console.log(
  795. "ERROR",
  796. "STATIONS_FORCE_SKIP",
  797. `Force skipping station "${stationId}" failed. "${err}"`
  798. );
  799. return cb({ status: "failure", message: err });
  800. }
  801. notifications.runJob("UNSCHEDULE", {
  802. name: `stations.nextSong?id=${stationId}`
  803. });
  804. stations.runJob("SKIP_STATION", { stationId });
  805. console.log("SUCCESS", "STATIONS_FORCE_SKIP", `Force skipped station "${stationId}" successfully.`);
  806. return cb({
  807. status: "success",
  808. message: "Successfully skipped station."
  809. });
  810. }
  811. );
  812. }),
  813. /**
  814. * Leaves the user's current station
  815. *
  816. * @param {object} session - user session
  817. * @param {string} stationId - id of station to leave
  818. * @param {Function} cb - callback
  819. */
  820. leave: (session, stationId, cb) => {
  821. async.waterfall(
  822. [
  823. next => {
  824. stations
  825. .runJob("GET_STATION", { stationId })
  826. .then(station => {
  827. next(null, station);
  828. })
  829. .catch(next);
  830. },
  831. (station, next) => {
  832. if (!station) return next("Station not found.");
  833. return next();
  834. }
  835. ],
  836. async (err, userCount) => {
  837. if (err) {
  838. err = await utils.runJob("GET_ERROR", { error: err });
  839. console.log("ERROR", "STATIONS_LEAVE", `Leaving station "${stationId}" failed. "${err}"`);
  840. return cb({ status: "failure", message: err });
  841. }
  842. console.log("SUCCESS", "STATIONS_LEAVE", `Left station "${stationId}" successfully.`);
  843. utils.runJob("SOCKET_LEAVE_ROOMS", { socketId: session });
  844. delete userList[session.socketId];
  845. return cb({
  846. status: "success",
  847. message: "Successfully left station.",
  848. userCount
  849. });
  850. }
  851. );
  852. },
  853. /**
  854. * Updates a station's name
  855. *
  856. * @param session
  857. * @param stationId - the station id
  858. * @param newName - the new station name
  859. * @param cb
  860. */
  861. updateName: isOwnerRequired(async (session, stationId, newName, cb) => {
  862. const stationModel = await db.runJob("GET_MODEL", {
  863. modelName: "station"
  864. });
  865. async.waterfall(
  866. [
  867. next => {
  868. stationModel.updateOne(
  869. { _id: stationId },
  870. { $set: { name: newName } },
  871. { runValidators: true },
  872. next
  873. );
  874. },
  875. (res, next) => {
  876. stations
  877. .runJob("UPDATE_STATION", { stationId })
  878. .then(station => {
  879. next(null, station);
  880. })
  881. .catch(next);
  882. }
  883. ],
  884. async err => {
  885. if (err) {
  886. err = await utils.runJob("GET_ERROR", { error: err });
  887. console.log(
  888. "ERROR",
  889. "STATIONS_UPDATE_NAME",
  890. `Updating station "${stationId}" name to "${newName}" failed. "${err}"`
  891. );
  892. return cb({ status: "failure", message: err });
  893. }
  894. console.log(
  895. "SUCCESS",
  896. "STATIONS_UPDATE_NAME",
  897. `Updated station "${stationId}" name to "${newName}" successfully.`
  898. );
  899. return cb({
  900. status: "success",
  901. message: "Successfully updated the name."
  902. });
  903. }
  904. );
  905. }),
  906. /**
  907. * Updates a station's display name
  908. *
  909. * @param session
  910. * @param stationId - the station id
  911. * @param newDisplayName - the new station display name
  912. * @param cb
  913. */
  914. updateDisplayName: isOwnerRequired(async (session, stationId, newDisplayName, cb) => {
  915. const stationModel = await db.runJob("GET_MODEL", {
  916. modelName: "station"
  917. });
  918. async.waterfall(
  919. [
  920. next => {
  921. stationModel.updateOne(
  922. { _id: stationId },
  923. { $set: { displayName: newDisplayName } },
  924. { runValidators: true },
  925. next
  926. );
  927. },
  928. (res, next) => {
  929. stations
  930. .runJob("UPDATE_STATION", { stationId })
  931. .then(station => {
  932. next(null, station);
  933. })
  934. .catch(next);
  935. }
  936. ],
  937. async err => {
  938. if (err) {
  939. err = await utils.runJob("GET_ERROR", { error: err });
  940. console.log(
  941. "ERROR",
  942. "STATIONS_UPDATE_DISPLAY_NAME",
  943. `Updating station "${stationId}" displayName to "${newDisplayName}" failed. "${err}"`
  944. );
  945. return cb({ status: "failure", message: err });
  946. }
  947. console.log(
  948. "SUCCESS",
  949. "STATIONS_UPDATE_DISPLAY_NAME",
  950. `Updated station "${stationId}" displayName to "${newDisplayName}" successfully.`
  951. );
  952. return cb({
  953. status: "success",
  954. message: "Successfully updated the display name."
  955. });
  956. }
  957. );
  958. }),
  959. /**
  960. * Updates a station's description
  961. *
  962. * @param session
  963. * @param stationId - the station id
  964. * @param newDescription - the new station description
  965. * @param cb
  966. */
  967. updateDescription: isOwnerRequired(async (session, stationId, newDescription, cb) => {
  968. const stationModel = await db.runJob("GET_MODEL", {
  969. modelName: "station"
  970. });
  971. async.waterfall(
  972. [
  973. next => {
  974. stationModel.updateOne(
  975. { _id: stationId },
  976. { $set: { description: newDescription } },
  977. { runValidators: true },
  978. next
  979. );
  980. },
  981. (res, next) => {
  982. stations
  983. .runJob("UPDATE_STATION", { stationId })
  984. .then(station => {
  985. next(null, station);
  986. })
  987. .catch(next);
  988. }
  989. ],
  990. async err => {
  991. if (err) {
  992. err = await utils.runJob("GET_ERROR", { error: err });
  993. console.log(
  994. "ERROR",
  995. "STATIONS_UPDATE_DESCRIPTION",
  996. `Updating station "${stationId}" description to "${newDescription}" failed. "${err}"`
  997. );
  998. return cb({ status: "failure", message: err });
  999. }
  1000. console.log(
  1001. "SUCCESS",
  1002. "STATIONS_UPDATE_DESCRIPTION",
  1003. `Updated station "${stationId}" description to "${newDescription}" successfully.`
  1004. );
  1005. return cb({
  1006. status: "success",
  1007. message: "Successfully updated the description."
  1008. });
  1009. }
  1010. );
  1011. }),
  1012. /**
  1013. * Updates a station's privacy
  1014. *
  1015. * @param session
  1016. * @param stationId - the station id
  1017. * @param newPrivacy - the new station privacy
  1018. * @param cb
  1019. */
  1020. updatePrivacy: isOwnerRequired(async (session, stationId, newPrivacy, cb) => {
  1021. const stationModel = await db.runJob("GET_MODEL", {
  1022. modelName: "station"
  1023. });
  1024. async.waterfall(
  1025. [
  1026. next => {
  1027. stationModel.updateOne(
  1028. { _id: stationId },
  1029. { $set: { privacy: newPrivacy } },
  1030. { runValidators: true },
  1031. next
  1032. );
  1033. },
  1034. (res, next) => {
  1035. stations
  1036. .runJob("UPDATE_STATION", { stationId })
  1037. .then(station => {
  1038. next(null, station);
  1039. })
  1040. .catch(next);
  1041. }
  1042. ],
  1043. async err => {
  1044. if (err) {
  1045. err = await utils.runJob("GET_ERROR", { error: err });
  1046. console.log(
  1047. "ERROR",
  1048. "STATIONS_UPDATE_PRIVACY",
  1049. `Updating station "${stationId}" privacy to "${newPrivacy}" failed. "${err}"`
  1050. );
  1051. return cb({ status: "failure", message: err });
  1052. }
  1053. console.log(
  1054. "SUCCESS",
  1055. "STATIONS_UPDATE_PRIVACY",
  1056. `Updated station "${stationId}" privacy to "${newPrivacy}" successfully.`
  1057. );
  1058. return cb({
  1059. status: "success",
  1060. message: "Successfully updated the privacy."
  1061. });
  1062. }
  1063. );
  1064. }),
  1065. /**
  1066. * Updates a station's genres
  1067. *
  1068. * @param session
  1069. * @param stationId - the station id
  1070. * @param newGenres - the new station genres
  1071. * @param cb
  1072. */
  1073. updateGenres: isOwnerRequired(async (session, stationId, newGenres, cb) => {
  1074. const stationModel = await db.runJob("GET_MODEL", {
  1075. modelName: "station"
  1076. });
  1077. async.waterfall(
  1078. [
  1079. next => {
  1080. stationModel.updateOne(
  1081. { _id: stationId },
  1082. { $set: { genres: newGenres } },
  1083. { runValidators: true },
  1084. next
  1085. );
  1086. },
  1087. (res, next) => {
  1088. stations
  1089. .runJob("UPDATE_STATION", { stationId })
  1090. .then(station => {
  1091. next(null, station);
  1092. })
  1093. .catch(next);
  1094. }
  1095. ],
  1096. async err => {
  1097. if (err) {
  1098. err = await utils.runJob("GET_ERROR", { error: err });
  1099. console.log(
  1100. "ERROR",
  1101. "STATIONS_UPDATE_GENRES",
  1102. `Updating station "${stationId}" genres to "${newGenres}" failed. "${err}"`
  1103. );
  1104. return cb({ status: "failure", message: err });
  1105. }
  1106. console.log(
  1107. "SUCCESS",
  1108. "STATIONS_UPDATE_GENRES",
  1109. `Updated station "${stationId}" genres to "${newGenres}" successfully.`
  1110. );
  1111. return cb({
  1112. status: "success",
  1113. message: "Successfully updated the genres."
  1114. });
  1115. }
  1116. );
  1117. }),
  1118. /**
  1119. * Updates a station's blacklisted genres
  1120. *
  1121. * @param session
  1122. * @param stationId - the station id
  1123. * @param newBlacklistedGenres - the new station blacklisted genres
  1124. * @param cb
  1125. */
  1126. updateBlacklistedGenres: isOwnerRequired(async (session, stationId, newBlacklistedGenres, cb) => {
  1127. const stationModel = await db.runJob("GET_MODEL", {
  1128. modelName: "station"
  1129. });
  1130. async.waterfall(
  1131. [
  1132. next => {
  1133. stationModel.updateOne(
  1134. { _id: stationId },
  1135. {
  1136. $set: {
  1137. blacklistedGenres: newBlacklistedGenres
  1138. }
  1139. },
  1140. { runValidators: true },
  1141. next
  1142. );
  1143. },
  1144. (res, next) => {
  1145. stations
  1146. .runJob("UPDATE_STATION", { stationId })
  1147. .then(station => {
  1148. next(null, station);
  1149. })
  1150. .catch(next);
  1151. }
  1152. ],
  1153. async err => {
  1154. if (err) {
  1155. err = await utils.runJob("GET_ERROR", { error: err });
  1156. console.log(
  1157. "ERROR",
  1158. "STATIONS_UPDATE_BLACKLISTED_GENRES",
  1159. `Updating station "${stationId}" blacklisted genres to "${newBlacklistedGenres}" failed. "${err}"`
  1160. );
  1161. return cb({ status: "failure", message: err });
  1162. }
  1163. console.log(
  1164. "SUCCESS",
  1165. "STATIONS_UPDATE_BLACKLISTED_GENRES",
  1166. `Updated station "${stationId}" blacklisted genres to "${newBlacklistedGenres}" successfully.`
  1167. );
  1168. return cb({
  1169. status: "success",
  1170. message: "Successfully updated the blacklisted genres."
  1171. });
  1172. }
  1173. );
  1174. }),
  1175. /**
  1176. * Updates a station's party mode
  1177. *
  1178. * @param session
  1179. * @param stationId - the station id
  1180. * @param newPartyMode - the new station party mode
  1181. * @param cb
  1182. */
  1183. updatePartyMode: isOwnerRequired(async (session, stationId, newPartyMode, cb) => {
  1184. const stationModel = await db.runJob("GET_MODEL", {
  1185. modelName: "station"
  1186. });
  1187. async.waterfall(
  1188. [
  1189. next => {
  1190. stations
  1191. .runJob("GET_STATION", { stationId })
  1192. .then(station => {
  1193. next(null, station);
  1194. })
  1195. .catch(next);
  1196. },
  1197. (station, next) => {
  1198. if (!station) return next("Station not found.");
  1199. if (station.partyMode === newPartyMode)
  1200. return next(`The party mode was already ${newPartyMode ? "enabled." : "disabled."}`);
  1201. return stationModel.updateOne(
  1202. { _id: stationId },
  1203. { $set: { partyMode: newPartyMode } },
  1204. { runValidators: true },
  1205. next
  1206. );
  1207. },
  1208. (res, next) => {
  1209. stations
  1210. .runJob("UPDATE_STATION", { stationId })
  1211. .then(station => {
  1212. next(null, station);
  1213. })
  1214. .catch(next);
  1215. }
  1216. ],
  1217. async err => {
  1218. if (err) {
  1219. err = await utils.runJob("GET_ERROR", { error: err });
  1220. console.log(
  1221. "ERROR",
  1222. "STATIONS_UPDATE_PARTY_MODE",
  1223. `Updating station "${stationId}" party mode to "${newPartyMode}" failed. "${err}"`
  1224. );
  1225. return cb({ status: "failure", message: err });
  1226. }
  1227. console.log(
  1228. "SUCCESS",
  1229. "STATIONS_UPDATE_PARTY_MODE",
  1230. `Updated station "${stationId}" party mode to "${newPartyMode}" successfully.`
  1231. );
  1232. cache.runJob("PUB", {
  1233. channel: "station.updatePartyMode",
  1234. value: {
  1235. stationId,
  1236. partyMode: newPartyMode
  1237. }
  1238. });
  1239. stations.runJob("SKIP_STATION", { stationId });
  1240. return cb({
  1241. status: "success",
  1242. message: "Successfully updated the party mode."
  1243. });
  1244. }
  1245. );
  1246. }),
  1247. /**
  1248. * Pauses a station
  1249. *
  1250. * @param session
  1251. * @param stationId - the station id
  1252. * @param cb
  1253. */
  1254. pause: isOwnerRequired(async (session, stationId, cb) => {
  1255. const stationModel = await db.runJob("GET_MODEL", {
  1256. modelName: "station"
  1257. });
  1258. async.waterfall(
  1259. [
  1260. next => {
  1261. stations
  1262. .runJob("GET_STATION", { stationId })
  1263. .then(station => {
  1264. next(null, station);
  1265. })
  1266. .catch(next);
  1267. },
  1268. (station, next) => {
  1269. if (!station) return next("Station not found.");
  1270. if (station.paused) return next("That station was already paused.");
  1271. return stationModel.updateOne(
  1272. { _id: stationId },
  1273. { $set: { paused: true, pausedAt: Date.now() } },
  1274. next
  1275. );
  1276. },
  1277. (res, next) => {
  1278. stations
  1279. .runJob("UPDATE_STATION", { stationId })
  1280. .then(station => {
  1281. next(null, station);
  1282. })
  1283. .catch(next);
  1284. }
  1285. ],
  1286. async err => {
  1287. if (err) {
  1288. err = await utils.runJob("GET_ERROR", { error: err });
  1289. console.log("ERROR", "STATIONS_PAUSE", `Pausing station "${stationId}" failed. "${err}"`);
  1290. return cb({ status: "failure", message: err });
  1291. }
  1292. console.log("SUCCESS", "STATIONS_PAUSE", `Paused station "${stationId}" successfully.`);
  1293. cache.runJob("PUB", {
  1294. channel: "station.pause",
  1295. value: stationId
  1296. });
  1297. notifications.runJob("UNSCHEDULE", {
  1298. name: `stations.nextSong?id=${stationId}`
  1299. });
  1300. return cb({
  1301. status: "success",
  1302. message: "Successfully paused."
  1303. });
  1304. }
  1305. );
  1306. }),
  1307. /**
  1308. * Resumes a station
  1309. *
  1310. * @param session
  1311. * @param stationId - the station id
  1312. * @param cb
  1313. */
  1314. resume: isOwnerRequired(async (session, stationId, cb) => {
  1315. const stationModel = await db.runJob("GET_MODEL", {
  1316. modelName: "station"
  1317. });
  1318. async.waterfall(
  1319. [
  1320. next => {
  1321. stations
  1322. .runJob("GET_STATION", { stationId })
  1323. .then(station => {
  1324. next(null, station);
  1325. })
  1326. .catch(next);
  1327. },
  1328. (station, next) => {
  1329. if (!station) return next("Station not found.");
  1330. if (!station.paused) return next("That station is not paused.");
  1331. station.timePaused += Date.now() - station.pausedAt;
  1332. return stationModel.updateOne(
  1333. { _id: stationId },
  1334. {
  1335. $set: { paused: false },
  1336. $inc: { timePaused: Date.now() - station.pausedAt }
  1337. },
  1338. next
  1339. );
  1340. },
  1341. (res, next) => {
  1342. stations
  1343. .runJob("UPDATE_STATION", { stationId })
  1344. .then(station => {
  1345. next(null, station);
  1346. })
  1347. .catch(next);
  1348. }
  1349. ],
  1350. async err => {
  1351. if (err) {
  1352. err = await utils.runJob("GET_ERROR", { error: err });
  1353. console.log("ERROR", "STATIONS_RESUME", `Resuming station "${stationId}" failed. "${err}"`);
  1354. return cb({ status: "failure", message: err });
  1355. }
  1356. console.log("SUCCESS", "STATIONS_RESUME", `Resuming station "${stationId}" successfully.`);
  1357. cache.runJob("PUB", {
  1358. channel: "station.resume",
  1359. value: stationId
  1360. });
  1361. return cb({
  1362. status: "success",
  1363. message: "Successfully resumed."
  1364. });
  1365. }
  1366. );
  1367. }),
  1368. /**
  1369. * Removes a station
  1370. *
  1371. * @param session
  1372. * @param stationId - the station id
  1373. * @param cb
  1374. */
  1375. remove: isOwnerRequired(async (session, stationId, cb) => {
  1376. const stationModel = await db.runJob("GET_MODEL", {
  1377. modelName: "station"
  1378. });
  1379. async.waterfall(
  1380. [
  1381. next => {
  1382. stationModel.deleteOne({ _id: stationId }, err => next(err));
  1383. },
  1384. next => {
  1385. cache.runJob("HDEL", { table: "stations", key: stationId }).then(next).catch(next);
  1386. }
  1387. ],
  1388. async err => {
  1389. if (err) {
  1390. err = await utils.runJob("GET_ERROR", { error: err });
  1391. console.log("ERROR", "STATIONS_REMOVE", `Removing station "${stationId}" failed. "${err}"`);
  1392. return cb({ status: "failure", message: err });
  1393. }
  1394. console.log("SUCCESS", "STATIONS_REMOVE", `Removing station "${stationId}" successfully.`);
  1395. cache.runJob("PUB", {
  1396. channel: "station.remove",
  1397. value: stationId
  1398. });
  1399. activities.runJob("ADD_ACTIVITY", {
  1400. userId: session.userId,
  1401. activityType: "deleted_station",
  1402. payload: [stationId]
  1403. });
  1404. return cb({
  1405. status: "success",
  1406. message: "Successfully removed."
  1407. });
  1408. }
  1409. );
  1410. }),
  1411. /**
  1412. * Create a station
  1413. *
  1414. * @param session
  1415. * @param data - the station data
  1416. * @param cb
  1417. */
  1418. create: isLoginRequired(async (session, data, cb) => {
  1419. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  1420. const stationModel = await db.runJob("GET_MODEL", {
  1421. modelName: "station"
  1422. });
  1423. data.name = data.name.toLowerCase();
  1424. const blacklist = [
  1425. "country",
  1426. "edm",
  1427. "musare",
  1428. "hip-hop",
  1429. "rap",
  1430. "top-hits",
  1431. "todays-hits",
  1432. "old-school",
  1433. "christmas",
  1434. "about",
  1435. "support",
  1436. "staff",
  1437. "help",
  1438. "news",
  1439. "terms",
  1440. "privacy",
  1441. "profile",
  1442. "c",
  1443. "community",
  1444. "tos",
  1445. "login",
  1446. "register",
  1447. "p",
  1448. "official",
  1449. "o",
  1450. "trap",
  1451. "faq",
  1452. "team",
  1453. "donate",
  1454. "buy",
  1455. "shop",
  1456. "forums",
  1457. "explore",
  1458. "settings",
  1459. "admin",
  1460. "auth",
  1461. "reset_password"
  1462. ];
  1463. async.waterfall(
  1464. [
  1465. next => {
  1466. if (!data) return next("Invalid data.");
  1467. return next();
  1468. },
  1469. next => {
  1470. stationModel.findOne(
  1471. {
  1472. $or: [
  1473. { name: data.name },
  1474. {
  1475. displayName: new RegExp(`^${data.displayName}$`, "i")
  1476. }
  1477. ]
  1478. },
  1479. next
  1480. );
  1481. },
  1482. // eslint-disable-next-line consistent-return
  1483. (station, next) => {
  1484. console.log(station);
  1485. if (station) return next("A station with that name or display name already exists.");
  1486. const { name, displayName, description, genres, playlist, type, blacklistedGenres } = data;
  1487. if (type === "official") {
  1488. return userModel.findOne({ _id: session.userId }, (err, user) => {
  1489. if (err) return next(err);
  1490. if (!user) return next("User not found.");
  1491. if (user.role !== "admin") return next("Admin required.");
  1492. return stationModel.create(
  1493. {
  1494. name,
  1495. displayName,
  1496. description,
  1497. type,
  1498. privacy: "private",
  1499. playlist,
  1500. genres,
  1501. blacklistedGenres,
  1502. currentSong: stations.defaultSong
  1503. },
  1504. next
  1505. );
  1506. });
  1507. }
  1508. if (type === "community") {
  1509. if (blacklist.indexOf(name) !== -1)
  1510. return next("That name is blacklisted. Please use a different name.");
  1511. return stationModel.create(
  1512. {
  1513. name,
  1514. displayName,
  1515. description,
  1516. type,
  1517. privacy: "private",
  1518. owner: session.userId,
  1519. queue: [],
  1520. currentSong: null
  1521. },
  1522. next
  1523. );
  1524. }
  1525. }
  1526. ],
  1527. async (err, station) => {
  1528. if (err) {
  1529. err = await utils.runJob("GET_ERROR", { error: err });
  1530. console.log("ERROR", "STATIONS_CREATE", `Creating station failed. "${err}"`);
  1531. return cb({ status: "failure", message: err });
  1532. }
  1533. console.log("SUCCESS", "STATIONS_CREATE", `Created station "${station._id}" successfully.`);
  1534. cache.runJob("PUB", {
  1535. channel: "station.create",
  1536. value: station._id
  1537. });
  1538. activities.runJob("ADD_ACTIVITY", {
  1539. userId: session.userId,
  1540. activityType: "created_station",
  1541. payload: [station._id]
  1542. });
  1543. return cb({
  1544. status: "success",
  1545. message: "Successfully created station."
  1546. });
  1547. }
  1548. );
  1549. }),
  1550. /**
  1551. * Adds song to station queue
  1552. *
  1553. * @param session
  1554. * @param stationId - the station id
  1555. * @param songId - the song id
  1556. * @param cb
  1557. */
  1558. addToQueue: isLoginRequired(async (session, stationId, songId, cb) => {
  1559. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  1560. const stationModel = await db.runJob("GET_MODEL", {
  1561. modelName: "station"
  1562. });
  1563. async.waterfall(
  1564. [
  1565. next => {
  1566. stations
  1567. .runJob("GET_STATION", { stationId })
  1568. .then(station => {
  1569. next(null, station);
  1570. })
  1571. .catch(next);
  1572. },
  1573. (station, next) => {
  1574. if (!station) return next("Station not found.");
  1575. if (station.locked) {
  1576. return userModel.findOne({ _id: session.userId }, (err, user) => {
  1577. if (user.role !== "admin" && station.owner !== session.userId)
  1578. return next("Only owners and admins can add songs to a locked queue.");
  1579. return next(null, station);
  1580. });
  1581. }
  1582. return next(null, station);
  1583. },
  1584. (station, next) => {
  1585. if (station.type !== "community") return next("That station is not a community station.");
  1586. return stations
  1587. .runJob("CAN_USER_VIEW_STATION", {
  1588. station,
  1589. userId: session.userId
  1590. })
  1591. .then(canView => {
  1592. if (canView) return next(null, station);
  1593. return next("Insufficient permissions.");
  1594. })
  1595. .catch(err => next(err));
  1596. },
  1597. (station, next) => {
  1598. if (station.currentSong && station.currentSong.songId === songId)
  1599. return next("That song is currently playing.");
  1600. return async.each(
  1601. station.queue,
  1602. (queueSong, next) => {
  1603. if (queueSong.songId === songId) return next("That song is already in the queue.");
  1604. return next();
  1605. },
  1606. err => next(err, station)
  1607. );
  1608. },
  1609. (station, next) => {
  1610. // songs
  1611. // .runJob("GET_SONG", { id: songId })
  1612. // .then((song) => {
  1613. // if (song) return next(null, song, station);
  1614. // else {
  1615. utils
  1616. .runJob("GET_SONG_FROM_YOUTUBE", { songId })
  1617. .then(response => {
  1618. const { song } = response;
  1619. song.artists = [];
  1620. song.skipDuration = 0;
  1621. song.likes = -1;
  1622. song.dislikes = -1;
  1623. song.thumbnail = "empty";
  1624. song.explicit = false;
  1625. next(null, song, station);
  1626. })
  1627. .catch(err => {
  1628. next(err);
  1629. });
  1630. // }
  1631. // })
  1632. // .catch((err) => {
  1633. // next(err);
  1634. // });
  1635. },
  1636. (song, station, next) => {
  1637. const { queue } = station;
  1638. song.requestedBy = session.userId;
  1639. queue.push(song);
  1640. let totalDuration = 0;
  1641. queue.forEach(song => {
  1642. totalDuration += song.duration;
  1643. });
  1644. if (totalDuration >= 3600 * 3) return next("The max length of the queue is 3 hours.");
  1645. return next(null, song, station);
  1646. },
  1647. (song, station, next) => {
  1648. const { queue } = station;
  1649. if (queue.length === 0) return next(null, song, station);
  1650. let totalDuration = 0;
  1651. const userId = queue[queue.length - 1].requestedBy;
  1652. station.queue.forEach(song => {
  1653. if (userId === song.requestedBy) {
  1654. totalDuration += song.duration;
  1655. }
  1656. });
  1657. if (totalDuration >= 900) return next("The max length of songs per user is 15 minutes.");
  1658. return next(null, song, station);
  1659. },
  1660. (song, station, next) => {
  1661. const { queue } = station;
  1662. if (queue.length === 0) return next(null, song);
  1663. let totalSongs = 0;
  1664. const userId = queue[queue.length - 1].requestedBy;
  1665. queue.forEach(song => {
  1666. if (userId === song.requestedBy) {
  1667. totalSongs += 1;
  1668. }
  1669. });
  1670. if (totalSongs <= 2) return next(null, song);
  1671. if (totalSongs > 3)
  1672. return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1673. if (queue[queue.length - 2].requestedBy !== userId || queue[queue.length - 3] !== userId)
  1674. return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1675. return next(null, song);
  1676. },
  1677. (song, next) => {
  1678. stationModel.updateOne(
  1679. { _id: stationId },
  1680. { $push: { queue: song } },
  1681. { runValidators: true },
  1682. next
  1683. );
  1684. },
  1685. (res, next) => {
  1686. stations
  1687. .runJob("UPDATE_STATION", { stationId })
  1688. .then(station => {
  1689. next(null, station);
  1690. })
  1691. .catch(next);
  1692. }
  1693. ],
  1694. async err => {
  1695. if (err) {
  1696. err = await utils.runJob("GET_ERROR", { error: err });
  1697. console.log(
  1698. "ERROR",
  1699. "STATIONS_ADD_SONG_TO_QUEUE",
  1700. `Adding song "${songId}" to station "${stationId}" queue failed. "${err}"`
  1701. );
  1702. return cb({ status: "failure", message: err });
  1703. }
  1704. console.log(
  1705. "SUCCESS",
  1706. "STATIONS_ADD_SONG_TO_QUEUE",
  1707. `Added song "${songId}" to station "${stationId}" successfully.`
  1708. );
  1709. cache.runJob("PUB", {
  1710. channel: "station.queueUpdate",
  1711. value: stationId
  1712. });
  1713. return cb({
  1714. status: "success",
  1715. message: "Successfully added song to queue."
  1716. });
  1717. }
  1718. );
  1719. }),
  1720. /**
  1721. * Removes song from station queue
  1722. *
  1723. * @param session
  1724. * @param stationId - the station id
  1725. * @param songId - the song id
  1726. * @param cb
  1727. */
  1728. removeFromQueue: isOwnerRequired(async (session, stationId, songId, cb) => {
  1729. const stationModel = await db.runJob("GET_MODEL", {
  1730. modelName: "station"
  1731. });
  1732. async.waterfall(
  1733. [
  1734. next => {
  1735. if (!songId) return next("Invalid song id.");
  1736. return stations
  1737. .runJob("GET_STATION", { stationId })
  1738. .then(station => {
  1739. next(null, station);
  1740. })
  1741. .catch(next);
  1742. },
  1743. (station, next) => {
  1744. if (!station) return next("Station not found.");
  1745. if (station.type !== "community") return next("Station is not a community station.");
  1746. return async.each(
  1747. station.queue,
  1748. (queueSong, next) => {
  1749. if (queueSong.songId === songId) return next(true);
  1750. return next();
  1751. },
  1752. err => {
  1753. if (err === true) return next();
  1754. return next("Song is not currently in the queue.");
  1755. }
  1756. );
  1757. },
  1758. next => {
  1759. stationModel.updateOne({ _id: stationId }, { $pull: { queue: { songId } } }, next);
  1760. },
  1761. (res, next) => {
  1762. stations
  1763. .runJob("UPDATE_STATION", { stationId })
  1764. .then(station => {
  1765. next(null, station);
  1766. })
  1767. .catch(next);
  1768. }
  1769. ],
  1770. async err => {
  1771. if (err) {
  1772. err = await utils.runJob("GET_ERROR", { error: err });
  1773. console.log(
  1774. "ERROR",
  1775. "STATIONS_REMOVE_SONG_TO_QUEUE",
  1776. `Removing song "${songId}" from station "${stationId}" queue failed. "${err}"`
  1777. );
  1778. return cb({ status: "failure", message: err });
  1779. }
  1780. console.log(
  1781. "SUCCESS",
  1782. "STATIONS_REMOVE_SONG_TO_QUEUE",
  1783. `Removed song "${songId}" from station "${stationId}" successfully.`
  1784. );
  1785. cache.runJob("PUB", {
  1786. channel: "station.queueUpdate",
  1787. value: stationId
  1788. });
  1789. return cb({
  1790. status: "success",
  1791. message: "Successfully removed song from queue."
  1792. });
  1793. }
  1794. );
  1795. }),
  1796. /**
  1797. * Gets the queue from a station
  1798. *
  1799. * @param {object} session - user session
  1800. * @param {string} stationId - the station id
  1801. * @param {Function} cb - callback
  1802. */
  1803. getQueue: (session, stationId, cb) => {
  1804. async.waterfall(
  1805. [
  1806. next => {
  1807. stations
  1808. .runJob("GET_STATION", { stationId })
  1809. .then(station => {
  1810. next(null, station);
  1811. })
  1812. .catch(next);
  1813. },
  1814. (station, next) => {
  1815. if (!station) return next("Station not found.");
  1816. if (station.type !== "community") return next("Station is not a community station.");
  1817. return next(null, station);
  1818. },
  1819. (station, next) => {
  1820. stations
  1821. .runJob("CAN_USER_VIEW_STATION", {
  1822. station,
  1823. userId: session.userId
  1824. })
  1825. .then(canView => {
  1826. if (canView) return next(null, station);
  1827. return next("Insufficient permissions.");
  1828. })
  1829. .catch(err => next(err));
  1830. }
  1831. ],
  1832. async (err, station) => {
  1833. if (err) {
  1834. err = await utils.runJob("GET_ERROR", { error: err });
  1835. console.log(
  1836. "ERROR",
  1837. "STATIONS_GET_QUEUE",
  1838. `Getting queue for station "${stationId}" failed. "${err}"`
  1839. );
  1840. return cb({ status: "failure", message: err });
  1841. }
  1842. console.log("SUCCESS", "STATIONS_GET_QUEUE", `Got queue for station "${stationId}" successfully.`);
  1843. return cb({
  1844. status: "success",
  1845. message: "Successfully got queue.",
  1846. queue: station.queue
  1847. });
  1848. }
  1849. );
  1850. },
  1851. /**
  1852. * Selects a private playlist for a station
  1853. *
  1854. * @param session
  1855. * @param stationId - the station id
  1856. * @param playlistId - the private playlist id
  1857. * @param cb
  1858. */
  1859. selectPrivatePlaylist: isOwnerRequired(async (session, stationId, playlistId, cb) => {
  1860. const stationModel = await db.runJob("GET_MODEL", {
  1861. modelName: "station"
  1862. });
  1863. const playlistModel = await db.runJob("GET_MODEL", {
  1864. modelName: "playlist"
  1865. });
  1866. async.waterfall(
  1867. [
  1868. next => {
  1869. stations
  1870. .runJob("GET_STATION", { stationId })
  1871. .then(station => {
  1872. next(null, station);
  1873. })
  1874. .catch(next);
  1875. },
  1876. (station, next) => {
  1877. if (!station) return next("Station not found.");
  1878. if (station.type !== "community") return next("Station is not a community station.");
  1879. if (station.privatePlaylist === playlistId)
  1880. return next("That private playlist is already selected.");
  1881. return playlistModel.findOne({ _id: playlistId }, next);
  1882. },
  1883. (playlist, next) => {
  1884. if (!playlist) return next("Playlist not found.");
  1885. const currentSongIndex = playlist.songs.length > 0 ? playlist.songs.length - 1 : 0;
  1886. return stationModel.updateOne(
  1887. { _id: stationId },
  1888. {
  1889. $set: {
  1890. privatePlaylist: playlistId,
  1891. currentSongIndex
  1892. }
  1893. },
  1894. { runValidators: true },
  1895. next
  1896. );
  1897. },
  1898. (res, next) => {
  1899. stations
  1900. .runJob("UPDATE_STATION", { stationId })
  1901. .then(station => {
  1902. next(null, station);
  1903. })
  1904. .catch(next);
  1905. }
  1906. ],
  1907. async (err, station) => {
  1908. if (err) {
  1909. err = await utils.runJob("GET_ERROR", { error: err });
  1910. console.log(
  1911. "ERROR",
  1912. "STATIONS_SELECT_PRIVATE_PLAYLIST",
  1913. `Selecting private playlist "${playlistId}" for station "${stationId}" failed. "${err}"`
  1914. );
  1915. return cb({ status: "failure", message: err });
  1916. }
  1917. console.log(
  1918. "SUCCESS",
  1919. "STATIONS_SELECT_PRIVATE_PLAYLIST",
  1920. `Selected private playlist "${playlistId}" for station "${stationId}" successfully.`
  1921. );
  1922. notifications.runJob("UNSCHEDULE", {
  1923. name: `stations.nextSong?id${stationId}`
  1924. });
  1925. if (!station.partyMode) stations.runJob("SKIP_STATION", { stationId });
  1926. cache.runJob("PUB", {
  1927. channel: "privatePlaylist.selected",
  1928. value: {
  1929. playlistId,
  1930. stationId
  1931. }
  1932. });
  1933. return cb({
  1934. status: "success",
  1935. message: "Successfully selected playlist."
  1936. });
  1937. }
  1938. );
  1939. }),
  1940. favoriteStation: isLoginRequired(async (session, stationId, cb) => {
  1941. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  1942. async.waterfall(
  1943. [
  1944. next => {
  1945. stations
  1946. .runJob("GET_STATION", { stationId })
  1947. .then(station => {
  1948. next(null, station);
  1949. })
  1950. .catch(next);
  1951. },
  1952. (station, next) => {
  1953. if (!station) return next("Station not found.");
  1954. return stations
  1955. .runJob("CAN_USER_VIEW_STATION", {
  1956. station,
  1957. userId: session.userId
  1958. })
  1959. .then(canView => {
  1960. if (canView) return next();
  1961. return next("Insufficient permissions.");
  1962. })
  1963. .catch(err => next(err));
  1964. },
  1965. next => {
  1966. userModel.updateOne({ _id: session.userId }, { $addToSet: { favoriteStations: stationId } }, next);
  1967. },
  1968. (res, next) => {
  1969. if (res.nModified === 0) return next("The station was already favorited.");
  1970. return next();
  1971. }
  1972. ],
  1973. async err => {
  1974. if (err) {
  1975. err = await utils.runJob("GET_ERROR", { error: err });
  1976. console.log("ERROR", "FAVORITE_STATION", `Favoriting station "${stationId}" failed. "${err}"`);
  1977. return cb({ status: "failure", message: err });
  1978. }
  1979. console.log("SUCCESS", "FAVORITE_STATION", `Favorited station "${stationId}" successfully.`);
  1980. cache.runJob("PUB", {
  1981. channel: "user.favoritedStation",
  1982. value: {
  1983. userId: session.userId,
  1984. stationId
  1985. }
  1986. });
  1987. return cb({
  1988. status: "success",
  1989. message: "Succesfully favorited station."
  1990. });
  1991. }
  1992. );
  1993. }),
  1994. unfavoriteStation: isLoginRequired(async (session, stationId, cb) => {
  1995. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  1996. async.waterfall(
  1997. [
  1998. next => {
  1999. userModel.updateOne({ _id: session.userId }, { $pull: { favoriteStations: stationId } }, next);
  2000. },
  2001. (res, next) => {
  2002. if (res.nModified === 0) return next("The station wasn't favorited.");
  2003. return next();
  2004. }
  2005. ],
  2006. async err => {
  2007. if (err) {
  2008. err = await utils.runJob("GET_ERROR", { error: err });
  2009. console.log("ERROR", "UNFAVORITE_STATION", `Unfavoriting station "${stationId}" failed. "${err}"`);
  2010. return cb({ status: "failure", message: err });
  2011. }
  2012. console.log("SUCCESS", "UNFAVORITE_STATION", `Unfavorited station "${stationId}" successfully.`);
  2013. cache.runJob("PUB", {
  2014. channel: "user.unfavoritedStation",
  2015. value: {
  2016. userId: session.userId,
  2017. stationId
  2018. }
  2019. });
  2020. return cb({
  2021. status: "success",
  2022. message: "Succesfully unfavorited station."
  2023. });
  2024. }
  2025. );
  2026. })
  2027. };