stations.js 54 KB

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