stations.js 53 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151
  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_FROM_ID", { songId })
  1612. .then(res => {
  1613. if (res.song) return next(null, res.song, station);
  1614. return utils
  1615. .runJob("GET_SONG_FROM_YOUTUBE", { songId })
  1616. .then(response => {
  1617. const { song } = response;
  1618. song.artists = [];
  1619. song.skipDuration = 0;
  1620. song.likes = -1;
  1621. song.dislikes = -1;
  1622. song.thumbnail = "empty";
  1623. song.explicit = false;
  1624. return next(null, song, station);
  1625. })
  1626. .catch(err => next(err));
  1627. })
  1628. .catch(err => next(err));
  1629. },
  1630. (song, station, next) => {
  1631. const { queue } = station;
  1632. song.requestedBy = session.userId;
  1633. queue.push(song);
  1634. let totalDuration = 0;
  1635. queue.forEach(song => {
  1636. totalDuration += song.duration;
  1637. });
  1638. if (totalDuration >= 3600 * 3) return next("The max length of the queue is 3 hours.");
  1639. return next(null, song, station);
  1640. },
  1641. (song, station, next) => {
  1642. const { queue } = station;
  1643. if (queue.length === 0) return next(null, song, station);
  1644. let totalDuration = 0;
  1645. const userId = queue[queue.length - 1].requestedBy;
  1646. station.queue.forEach(song => {
  1647. if (userId === song.requestedBy) {
  1648. totalDuration += song.duration;
  1649. }
  1650. });
  1651. if (totalDuration >= 900) return next("The max length of songs per user is 15 minutes.");
  1652. return next(null, song, station);
  1653. },
  1654. (song, station, next) => {
  1655. const { queue } = station;
  1656. if (queue.length === 0) return next(null, song);
  1657. let totalSongs = 0;
  1658. const userId = queue[queue.length - 1].requestedBy;
  1659. queue.forEach(song => {
  1660. if (userId === song.requestedBy) {
  1661. totalSongs += 1;
  1662. }
  1663. });
  1664. if (totalSongs <= 2) return next(null, song);
  1665. if (totalSongs > 3)
  1666. return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1667. if (queue[queue.length - 2].requestedBy !== userId || queue[queue.length - 3] !== userId)
  1668. return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1669. return next(null, song);
  1670. },
  1671. (song, next) => {
  1672. stationModel.updateOne(
  1673. { _id: stationId },
  1674. { $push: { queue: song } },
  1675. { runValidators: true },
  1676. next
  1677. );
  1678. },
  1679. (res, next) => {
  1680. stations
  1681. .runJob("UPDATE_STATION", { stationId })
  1682. .then(station => {
  1683. next(null, station);
  1684. })
  1685. .catch(next);
  1686. }
  1687. ],
  1688. async err => {
  1689. if (err) {
  1690. err = await utils.runJob("GET_ERROR", { error: err });
  1691. console.log(
  1692. "ERROR",
  1693. "STATIONS_ADD_SONG_TO_QUEUE",
  1694. `Adding song "${songId}" to station "${stationId}" queue failed. "${err}"`
  1695. );
  1696. return cb({ status: "failure", message: err });
  1697. }
  1698. console.log(
  1699. "SUCCESS",
  1700. "STATIONS_ADD_SONG_TO_QUEUE",
  1701. `Added song "${songId}" to station "${stationId}" successfully.`
  1702. );
  1703. cache.runJob("PUB", {
  1704. channel: "station.queueUpdate",
  1705. value: stationId
  1706. });
  1707. return cb({
  1708. status: "success",
  1709. message: "Successfully added song to queue."
  1710. });
  1711. }
  1712. );
  1713. }),
  1714. /**
  1715. * Removes song from station queue
  1716. *
  1717. * @param session
  1718. * @param stationId - the station id
  1719. * @param songId - the song id
  1720. * @param cb
  1721. */
  1722. removeFromQueue: isOwnerRequired(async (session, stationId, songId, cb) => {
  1723. const stationModel = await db.runJob("GET_MODEL", {
  1724. modelName: "station"
  1725. });
  1726. async.waterfall(
  1727. [
  1728. next => {
  1729. if (!songId) return next("Invalid song id.");
  1730. return stations
  1731. .runJob("GET_STATION", { stationId })
  1732. .then(station => {
  1733. next(null, station);
  1734. })
  1735. .catch(next);
  1736. },
  1737. (station, next) => {
  1738. if (!station) return next("Station not found.");
  1739. if (station.type !== "community") return next("Station is not a community station.");
  1740. return async.each(
  1741. station.queue,
  1742. (queueSong, next) => {
  1743. if (queueSong.songId === songId) return next(true);
  1744. return next();
  1745. },
  1746. err => {
  1747. if (err === true) return next();
  1748. return next("Song is not currently in the queue.");
  1749. }
  1750. );
  1751. },
  1752. next => {
  1753. stationModel.updateOne({ _id: stationId }, { $pull: { queue: { songId } } }, next);
  1754. },
  1755. (res, next) => {
  1756. stations
  1757. .runJob("UPDATE_STATION", { stationId })
  1758. .then(station => {
  1759. next(null, station);
  1760. })
  1761. .catch(next);
  1762. }
  1763. ],
  1764. async err => {
  1765. if (err) {
  1766. err = await utils.runJob("GET_ERROR", { error: err });
  1767. console.log(
  1768. "ERROR",
  1769. "STATIONS_REMOVE_SONG_TO_QUEUE",
  1770. `Removing song "${songId}" from station "${stationId}" queue failed. "${err}"`
  1771. );
  1772. return cb({ status: "failure", message: err });
  1773. }
  1774. console.log(
  1775. "SUCCESS",
  1776. "STATIONS_REMOVE_SONG_TO_QUEUE",
  1777. `Removed song "${songId}" from station "${stationId}" successfully.`
  1778. );
  1779. cache.runJob("PUB", {
  1780. channel: "station.queueUpdate",
  1781. value: stationId
  1782. });
  1783. return cb({
  1784. status: "success",
  1785. message: "Successfully removed song from queue."
  1786. });
  1787. }
  1788. );
  1789. }),
  1790. /**
  1791. * Gets the queue from a station
  1792. *
  1793. * @param {object} session - user session
  1794. * @param {string} stationId - the station id
  1795. * @param {Function} cb - callback
  1796. */
  1797. getQueue: (session, stationId, cb) => {
  1798. async.waterfall(
  1799. [
  1800. next => {
  1801. stations
  1802. .runJob("GET_STATION", { stationId })
  1803. .then(station => {
  1804. next(null, station);
  1805. })
  1806. .catch(next);
  1807. },
  1808. (station, next) => {
  1809. if (!station) return next("Station not found.");
  1810. if (station.type !== "community") return next("Station is not a community station.");
  1811. return next(null, station);
  1812. },
  1813. (station, next) => {
  1814. stations
  1815. .runJob("CAN_USER_VIEW_STATION", {
  1816. station,
  1817. userId: session.userId
  1818. })
  1819. .then(canView => {
  1820. if (canView) return next(null, station);
  1821. return next("Insufficient permissions.");
  1822. })
  1823. .catch(err => next(err));
  1824. }
  1825. ],
  1826. async (err, station) => {
  1827. if (err) {
  1828. err = await utils.runJob("GET_ERROR", { error: err });
  1829. console.log(
  1830. "ERROR",
  1831. "STATIONS_GET_QUEUE",
  1832. `Getting queue for station "${stationId}" failed. "${err}"`
  1833. );
  1834. return cb({ status: "failure", message: err });
  1835. }
  1836. console.log("SUCCESS", "STATIONS_GET_QUEUE", `Got queue for station "${stationId}" successfully.`);
  1837. return cb({
  1838. status: "success",
  1839. message: "Successfully got queue.",
  1840. queue: station.queue
  1841. });
  1842. }
  1843. );
  1844. },
  1845. /**
  1846. * Selects a private playlist for a station
  1847. *
  1848. * @param session
  1849. * @param stationId - the station id
  1850. * @param playlistId - the private playlist id
  1851. * @param cb
  1852. */
  1853. selectPrivatePlaylist: isOwnerRequired(async (session, stationId, playlistId, cb) => {
  1854. const stationModel = await db.runJob("GET_MODEL", {
  1855. modelName: "station"
  1856. });
  1857. const playlistModel = await db.runJob("GET_MODEL", {
  1858. modelName: "playlist"
  1859. });
  1860. async.waterfall(
  1861. [
  1862. next => {
  1863. stations
  1864. .runJob("GET_STATION", { stationId })
  1865. .then(station => {
  1866. next(null, station);
  1867. })
  1868. .catch(next);
  1869. },
  1870. (station, next) => {
  1871. if (!station) return next("Station not found.");
  1872. if (station.type !== "community") return next("Station is not a community station.");
  1873. if (station.privatePlaylist === playlistId)
  1874. return next("That private playlist is already selected.");
  1875. return playlistModel.findOne({ _id: playlistId }, next);
  1876. },
  1877. (playlist, next) => {
  1878. if (!playlist) return next("Playlist not found.");
  1879. const currentSongIndex = playlist.songs.length > 0 ? playlist.songs.length - 1 : 0;
  1880. return stationModel.updateOne(
  1881. { _id: stationId },
  1882. {
  1883. $set: {
  1884. privatePlaylist: playlistId,
  1885. currentSongIndex
  1886. }
  1887. },
  1888. { runValidators: true },
  1889. next
  1890. );
  1891. },
  1892. (res, next) => {
  1893. stations
  1894. .runJob("UPDATE_STATION", { stationId })
  1895. .then(station => {
  1896. next(null, station);
  1897. })
  1898. .catch(next);
  1899. }
  1900. ],
  1901. async (err, station) => {
  1902. if (err) {
  1903. err = await utils.runJob("GET_ERROR", { error: err });
  1904. console.log(
  1905. "ERROR",
  1906. "STATIONS_SELECT_PRIVATE_PLAYLIST",
  1907. `Selecting private playlist "${playlistId}" for station "${stationId}" failed. "${err}"`
  1908. );
  1909. return cb({ status: "failure", message: err });
  1910. }
  1911. console.log(
  1912. "SUCCESS",
  1913. "STATIONS_SELECT_PRIVATE_PLAYLIST",
  1914. `Selected private playlist "${playlistId}" for station "${stationId}" successfully.`
  1915. );
  1916. notifications.runJob("UNSCHEDULE", {
  1917. name: `stations.nextSong?id${stationId}`
  1918. });
  1919. if (!station.partyMode) stations.runJob("SKIP_STATION", { stationId });
  1920. cache.runJob("PUB", {
  1921. channel: "privatePlaylist.selected",
  1922. value: {
  1923. playlistId,
  1924. stationId
  1925. }
  1926. });
  1927. return cb({
  1928. status: "success",
  1929. message: "Successfully selected playlist."
  1930. });
  1931. }
  1932. );
  1933. }),
  1934. favoriteStation: isLoginRequired(async (session, stationId, cb) => {
  1935. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  1936. async.waterfall(
  1937. [
  1938. next => {
  1939. stations
  1940. .runJob("GET_STATION", { stationId })
  1941. .then(station => {
  1942. next(null, station);
  1943. })
  1944. .catch(next);
  1945. },
  1946. (station, next) => {
  1947. if (!station) return next("Station not found.");
  1948. return stations
  1949. .runJob("CAN_USER_VIEW_STATION", {
  1950. station,
  1951. userId: session.userId
  1952. })
  1953. .then(canView => {
  1954. if (canView) return next();
  1955. return next("Insufficient permissions.");
  1956. })
  1957. .catch(err => next(err));
  1958. },
  1959. next => {
  1960. userModel.updateOne({ _id: session.userId }, { $addToSet: { favoriteStations: stationId } }, next);
  1961. },
  1962. (res, next) => {
  1963. if (res.nModified === 0) return next("The station was already favorited.");
  1964. return next();
  1965. }
  1966. ],
  1967. async err => {
  1968. if (err) {
  1969. err = await utils.runJob("GET_ERROR", { error: err });
  1970. console.log("ERROR", "FAVORITE_STATION", `Favoriting station "${stationId}" failed. "${err}"`);
  1971. return cb({ status: "failure", message: err });
  1972. }
  1973. console.log("SUCCESS", "FAVORITE_STATION", `Favorited station "${stationId}" successfully.`);
  1974. cache.runJob("PUB", {
  1975. channel: "user.favoritedStation",
  1976. value: {
  1977. userId: session.userId,
  1978. stationId
  1979. }
  1980. });
  1981. return cb({
  1982. status: "success",
  1983. message: "Succesfully favorited station."
  1984. });
  1985. }
  1986. );
  1987. }),
  1988. unfavoriteStation: isLoginRequired(async (session, stationId, cb) => {
  1989. const userModel = await db.runJob("GET_MODEL", { modelName: "user" });
  1990. async.waterfall(
  1991. [
  1992. next => {
  1993. userModel.updateOne({ _id: session.userId }, { $pull: { favoriteStations: stationId } }, next);
  1994. },
  1995. (res, next) => {
  1996. if (res.nModified === 0) return next("The station wasn't favorited.");
  1997. return next();
  1998. }
  1999. ],
  2000. async err => {
  2001. if (err) {
  2002. err = await utils.runJob("GET_ERROR", { error: err });
  2003. console.log("ERROR", "UNFAVORITE_STATION", `Unfavoriting station "${stationId}" failed. "${err}"`);
  2004. return cb({ status: "failure", message: err });
  2005. }
  2006. console.log("SUCCESS", "UNFAVORITE_STATION", `Unfavorited station "${stationId}" successfully.`);
  2007. cache.runJob("PUB", {
  2008. channel: "user.unfavoritedStation",
  2009. value: {
  2010. userId: session.userId,
  2011. stationId
  2012. }
  2013. });
  2014. return cb({
  2015. status: "success",
  2016. message: "Succesfully unfavorited station."
  2017. });
  2018. }
  2019. );
  2020. })
  2021. };