stations.js 62 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353
  1. import async from "async";
  2. import config from "config";
  3. import CoreClass from "../core";
  4. import { hasPermission } from "./hooks/hasPermission";
  5. let StationsModule;
  6. let CacheModule;
  7. let DBModule;
  8. let UtilsModule;
  9. let WSModule;
  10. let PlaylistsModule;
  11. let NotificationsModule;
  12. let MediaModule;
  13. let SongsModule;
  14. class _StationsModule extends CoreClass {
  15. // eslint-disable-next-line require-jsdoc
  16. constructor() {
  17. super("stations");
  18. StationsModule = this;
  19. }
  20. /**
  21. * Initialises the stations module
  22. *
  23. * @returns {Promise} - returns promise (reject, resolve)
  24. */
  25. async initialize() {
  26. CacheModule = this.moduleManager.modules.cache;
  27. DBModule = this.moduleManager.modules.db;
  28. UtilsModule = this.moduleManager.modules.utils;
  29. WSModule = this.moduleManager.modules.ws;
  30. PlaylistsModule = this.moduleManager.modules.playlists;
  31. NotificationsModule = this.moduleManager.modules.notifications;
  32. MediaModule = this.moduleManager.modules.media;
  33. SongsModule = this.moduleManager.modules.songs;
  34. this.userList = {};
  35. this.usersPerStation = {};
  36. this.usersPerStationCount = {};
  37. // TEMP
  38. CacheModule.runJob("SUB", {
  39. channel: "station.pause",
  40. cb: async stationId => {
  41. NotificationsModule.runJob("REMOVE", {
  42. subscription: `stations.nextSong?id=${stationId}`
  43. }).then();
  44. }
  45. });
  46. CacheModule.runJob("SUB", {
  47. channel: "station.resume",
  48. cb: async stationId => {
  49. StationsModule.runJob("INITIALIZE_STATION", { stationId }).then();
  50. }
  51. });
  52. CacheModule.runJob("SUB", {
  53. channel: "station.queueUpdate",
  54. cb: async stationId => {
  55. StationsModule.runJob("GET_STATION", { stationId }).then(station => {
  56. if (!station.currentSong && station.queue.length > 0) {
  57. StationsModule.runJob("INITIALIZE_STATION", {
  58. stationId
  59. }).then();
  60. }
  61. WSModule.runJob("EMIT_TO_ROOM", {
  62. room: `station.${stationId}`,
  63. args: ["event:station.queue.updated", { data: { queue: station.queue } }]
  64. });
  65. WSModule.runJob("EMIT_TO_ROOM", {
  66. room: `manage-station.${stationId}`,
  67. args: ["event:manageStation.queue.updated", { data: { stationId, queue: station.queue } }]
  68. });
  69. });
  70. }
  71. });
  72. const userModel = (this.userModel = await DBModule.runJob("GET_MODEL", { modelName: "user" }));
  73. CacheModule.runJob("SUB", {
  74. channel: "station.djs.added",
  75. cb: async ({ stationId, userId }) => {
  76. userModel.findOne({ _id: userId }, (err, user) => {
  77. if (!err && user) {
  78. const { _id, name, username, avatar } = user;
  79. const data = { data: { user: { _id, name, username, avatar }, stationId } };
  80. WSModule.runJob("EMIT_TO_ROOMS", {
  81. rooms: [`station.${stationId}`, "home"],
  82. args: ["event:station.djs.added", data]
  83. });
  84. WSModule.runJob("EMIT_TO_ROOM", {
  85. room: `manage-station.${stationId}`,
  86. args: ["event:manageStation.djs.added", data]
  87. });
  88. }
  89. });
  90. }
  91. });
  92. CacheModule.runJob("SUB", {
  93. channel: "station.djs.removed",
  94. cb: async ({ stationId, userId }) => {
  95. userModel.findOne({ _id: userId }, (err, user) => {
  96. if (!err && user) {
  97. const { _id, name, username, avatar } = user;
  98. const data = { data: { user: { _id, name, username, avatar }, stationId } };
  99. WSModule.runJob("EMIT_TO_ROOMS", {
  100. rooms: [`station.${stationId}`, "home"],
  101. args: ["event:station.djs.removed", data]
  102. });
  103. WSModule.runJob("EMIT_TO_ROOM", {
  104. room: `manage-station.${stationId}`,
  105. args: ["event:manageStation.djs.removed", data]
  106. });
  107. }
  108. });
  109. }
  110. });
  111. const stationModel = (this.stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }));
  112. const stationSchema = (this.stationSchema = await CacheModule.runJob("GET_SCHEMA", { schemaName: "station" }));
  113. this.stationHistoryModel = await DBModule.runJob("GET_MODEL", {
  114. modelName: "stationHistory"
  115. });
  116. return new Promise((resolve, reject) => {
  117. async.waterfall(
  118. [
  119. next => {
  120. this.setStage(2);
  121. CacheModule.runJob("HGETALL", { table: "stations" })
  122. .then(stations => {
  123. next(null, stations);
  124. })
  125. .catch(next);
  126. },
  127. (stations, next) => {
  128. this.setStage(3);
  129. if (!stations) return next();
  130. const stationIds = Object.keys(stations);
  131. return async.each(
  132. stationIds,
  133. (stationId, next) => {
  134. stationModel.findOne({ _id: stationId }, (err, station) => {
  135. if (err) next(err);
  136. else if (!station) {
  137. CacheModule.runJob("HDEL", {
  138. table: "stations",
  139. key: stationId
  140. })
  141. .then(() => {
  142. next();
  143. })
  144. .catch(next);
  145. } else next();
  146. });
  147. },
  148. next
  149. );
  150. },
  151. next => {
  152. this.setStage(4);
  153. const mediaSources = [];
  154. if (!config.get("experimental.soundcloud")) {
  155. mediaSources.push(/^soundcloud:/);
  156. }
  157. if (!config.get("experimental.spotify")) {
  158. mediaSources.push(/^spotify:/);
  159. }
  160. if (mediaSources.length > 0)
  161. stationModel.updateMany(
  162. {},
  163. { $pull: { queue: { mediaSource: { $in: mediaSources } } } },
  164. err => {
  165. if (err) next(err);
  166. else next();
  167. }
  168. );
  169. else next();
  170. },
  171. next => {
  172. this.setStage(5);
  173. stationModel.find({}, next);
  174. },
  175. (stations, next) => {
  176. this.setStage(6);
  177. async.each(
  178. stations,
  179. (station, next2) => {
  180. async.waterfall(
  181. [
  182. next => {
  183. CacheModule.runJob("HSET", {
  184. table: "stations",
  185. key: station._id,
  186. value: stationSchema(station)
  187. })
  188. .then(station => next(null, station))
  189. .catch(next);
  190. },
  191. (station, next) => {
  192. StationsModule.runJob(
  193. "INITIALIZE_STATION",
  194. {
  195. stationId: station._id
  196. },
  197. null,
  198. -1
  199. )
  200. .then(() => {
  201. next();
  202. })
  203. .catch(next);
  204. }
  205. ],
  206. err => {
  207. next2(err);
  208. }
  209. );
  210. },
  211. next
  212. );
  213. }
  214. ],
  215. async err => {
  216. if (err) {
  217. err = await UtilsModule.runJob("GET_ERROR", {
  218. error: err
  219. });
  220. reject(new Error(err));
  221. } else {
  222. resolve();
  223. }
  224. }
  225. );
  226. });
  227. }
  228. /**
  229. * Initialises a station
  230. *
  231. * @param {object} payload - object that contains the payload
  232. * @param {string} payload.stationId - id of the station to initialise
  233. * @returns {Promise} - returns a promise (resolve, reject)
  234. */
  235. INITIALIZE_STATION(payload) {
  236. return new Promise((resolve, reject) => {
  237. async.waterfall(
  238. [
  239. next => {
  240. StationsModule.runJob(
  241. "GET_STATION",
  242. {
  243. stationId: payload.stationId
  244. },
  245. this
  246. )
  247. .then(station => {
  248. next(null, station);
  249. })
  250. .catch(next);
  251. },
  252. (station, next) => {
  253. if (!station) return next("Station not found.");
  254. return NotificationsModule.runJob(
  255. "UNSCHEDULE",
  256. {
  257. name: `stations.nextSong?id=${station._id}`
  258. },
  259. this
  260. )
  261. .then()
  262. .catch()
  263. .finally(() => {
  264. NotificationsModule.runJob("SUBSCRIBE", {
  265. name: `stations.nextSong?id=${station._id}`,
  266. cb: () =>
  267. StationsModule.runJob("SKIP_STATION", {
  268. stationId: station._id,
  269. natural: true,
  270. skipReason: "natural"
  271. }),
  272. unique: true,
  273. station
  274. })
  275. .then()
  276. .catch();
  277. return next(null, station);
  278. });
  279. },
  280. (station, next) => {
  281. // A current song is invalid if it isn't allowed to be played. Spotify songs can never be played, and SoundCloud songs can't be played if SoundCloud isn't enabled
  282. let currentSongIsInvalid = false;
  283. if (station.currentSong && station.currentSong.mediaSource) {
  284. if (station.currentSong.mediaSource.startsWith("spotify:")) currentSongIsInvalid = true;
  285. if (
  286. station.currentSong.mediaSource.startsWith("soundcloud:") &&
  287. !config.get("experimental.soundcloud")
  288. )
  289. currentSongIsInvalid = true;
  290. }
  291. if (
  292. (!station.paused && !station.currentSong) ||
  293. (station.currentSong && currentSongIsInvalid)
  294. ) {
  295. return StationsModule.runJob(
  296. "SKIP_STATION",
  297. {
  298. stationId: station._id,
  299. natural: false,
  300. skipReason: "other"
  301. },
  302. this
  303. )
  304. .then(station => {
  305. next(null, station);
  306. })
  307. .catch(next)
  308. .finally(() => {});
  309. }
  310. if (station.paused) return next(null, station);
  311. let timeLeft =
  312. station.currentSong.duration * 1000 - (Date.now() - station.startedAt - station.timePaused);
  313. if (Number.isNaN(timeLeft)) timeLeft = -1;
  314. if (station.currentSong.duration * 1000 < timeLeft || timeLeft < 0) {
  315. return StationsModule.runJob(
  316. "SKIP_STATION",
  317. {
  318. stationId: station._id,
  319. natural: false,
  320. skipReason: "other"
  321. },
  322. this
  323. )
  324. .then(station => {
  325. next(null, station);
  326. })
  327. .catch(next);
  328. }
  329. // name, time, cb, station
  330. NotificationsModule.runJob("SCHEDULE", {
  331. name: `stations.nextSong?id=${station._id}`,
  332. time: timeLeft,
  333. station
  334. });
  335. return next(null, station);
  336. }
  337. ],
  338. async (err, station) => {
  339. if (err && err !== true) {
  340. err = await UtilsModule.runJob(
  341. "GET_ERROR",
  342. {
  343. error: err
  344. },
  345. this
  346. );
  347. reject(new Error(err));
  348. } else resolve(station);
  349. }
  350. );
  351. });
  352. }
  353. /**
  354. * Attempts to get the station from Redis. If it's not in Redis, get it from Mongo and add it to Redis.
  355. *
  356. * @param {object} payload - object that contains the payload
  357. * @param {string} payload.stationId - id of the station
  358. * @returns {Promise} - returns a promise (resolve, reject)
  359. */
  360. GET_STATION(payload) {
  361. return new Promise((resolve, reject) => {
  362. async.waterfall(
  363. [
  364. next => {
  365. CacheModule.runJob("HGET", { table: "stations", key: payload.stationId }, this)
  366. .then(station => next(null, station))
  367. .catch(next);
  368. },
  369. (station, next) => {
  370. if (station) return next(true, station);
  371. return StationsModule.stationModel.findOne({ _id: payload.stationId }, next);
  372. },
  373. (station, next) => {
  374. if (station) {
  375. station = StationsModule.stationSchema(station);
  376. CacheModule.runJob("HSET", {
  377. table: "stations",
  378. key: payload.stationId,
  379. value: station
  380. })
  381. .then()
  382. .catch();
  383. next(true, station);
  384. } else next("Station not found");
  385. }
  386. ],
  387. async (err, station) => {
  388. if (err && err !== true) {
  389. err = await UtilsModule.runJob(
  390. "GET_ERROR",
  391. {
  392. error: err
  393. },
  394. this
  395. );
  396. reject(new Error(err));
  397. } else resolve(station);
  398. }
  399. );
  400. });
  401. }
  402. /**
  403. * Attempts to get a station by name, firstly from Redis. If it's not in Redis, get it from Mongo and add it to Redis.
  404. *
  405. * @param {object} payload - object that contains the payload
  406. * @param {string} payload.stationName - the unique name of the station
  407. * @returns {Promise} - returns a promise (resolve, reject)
  408. */
  409. async GET_STATION_BY_NAME(payload) {
  410. return new Promise((resolve, reject) => {
  411. async.waterfall(
  412. [
  413. next => {
  414. StationsModule.stationModel.findOne({ name: payload.stationName }, next);
  415. },
  416. (station, next) => {
  417. if (station) {
  418. station = StationsModule.stationSchema(station);
  419. CacheModule.runJob("HSET", {
  420. table: "stations",
  421. key: station._id,
  422. value: station
  423. });
  424. next(true, station);
  425. } else next("Station not found");
  426. }
  427. ],
  428. (err, station) => {
  429. if (err && err !== true) return reject(new Error(err));
  430. return resolve(station);
  431. }
  432. );
  433. });
  434. }
  435. /**
  436. * Updates the station in cache from mongo or deletes station in cache if no longer in mongo.
  437. *
  438. * @param {object} payload - object that contains the payload
  439. * @param {string} payload.stationId - the id of the station to update
  440. * @returns {Promise} - returns a promise (resolve, reject)
  441. */
  442. UPDATE_STATION(payload) {
  443. return new Promise((resolve, reject) => {
  444. async.waterfall(
  445. [
  446. next => {
  447. StationsModule.stationModel.findOne({ _id: payload.stationId }, next);
  448. },
  449. (station, next) => {
  450. if (!station) {
  451. CacheModule.runJob("HDEL", {
  452. table: "stations",
  453. key: payload.stationId
  454. })
  455. .then()
  456. .catch();
  457. return next("Station not found");
  458. }
  459. return CacheModule.runJob(
  460. "HSET",
  461. {
  462. table: "stations",
  463. key: payload.stationId,
  464. value: station
  465. },
  466. this
  467. )
  468. .then(station => {
  469. next(null, station);
  470. })
  471. .catch(next);
  472. }
  473. ],
  474. async (err, station) => {
  475. if (err && err !== true) {
  476. err = await UtilsModule.runJob(
  477. "GET_ERROR",
  478. {
  479. error: err
  480. },
  481. this
  482. );
  483. reject(new Error(err));
  484. } else resolve(station);
  485. }
  486. );
  487. });
  488. }
  489. /**
  490. * Autofill station queue from station playlist
  491. *
  492. * @param {object} payload - object that contains the payload
  493. * @param {string} payload.stationId - the id of the station
  494. * @param {string} payload.ignoreExistingQueue - ignore the existing queue songs, replacing the old queue with a completely fresh one
  495. * @returns {Promise} - returns a promise (resolve, reject)
  496. */
  497. AUTOFILL_STATION(payload) {
  498. return new Promise((resolve, reject) => {
  499. const { stationId, ignoreExistingQueue } = payload;
  500. async.waterfall(
  501. [
  502. next => {
  503. PlaylistsModule.runJob("GET_STATION_PLAYLIST", { stationId, includeSongs: true }, this)
  504. .then(response => {
  505. next(null, response.playlist);
  506. })
  507. .catch(next);
  508. },
  509. (playlist, next) => {
  510. StationsModule.runJob("GET_STATION", { stationId }, this)
  511. .then(station => {
  512. if (!station.autofill.enabled) return next("Autofill is disabled in this station");
  513. if (
  514. !ignoreExistingQueue &&
  515. station.autofill.limit <= station.queue.filter(song => !song.requestedBy).length
  516. )
  517. return next("Autofill limit reached");
  518. if (ignoreExistingQueue) station.queue = [];
  519. return next(null, playlist, station);
  520. })
  521. .catch(next);
  522. },
  523. (playlist, station, next) => {
  524. if (station.autofill.mode === "random") {
  525. UtilsModule.runJob("SHUFFLE", { array: playlist.songs }, this)
  526. .then(response => {
  527. next(null, response.array, station);
  528. })
  529. .catch(next);
  530. } else next(null, playlist.songs, station);
  531. },
  532. async (_playlistSongs, station) => {
  533. let playlistSongs = JSON.parse(JSON.stringify(_playlistSongs));
  534. if (station.autofill.mode === "sequential") {
  535. if (station.currentSongIndex <= playlistSongs.length) {
  536. const songsToAddToEnd = playlistSongs.splice(0, station.currentSongIndex);
  537. playlistSongs = [...playlistSongs, ...songsToAddToEnd];
  538. }
  539. }
  540. const currentRequests = station.queue.filter(song => !song.requestedBy).length;
  541. const songsStillNeeded = station.autofill.limit - currentRequests;
  542. const currentSongs = station.queue;
  543. let currentMediaSources = station.queue.map(song => song.mediaSource);
  544. const songsToAdd = [];
  545. let lastSongAdded = null;
  546. if (station.currentSong && station.currentSong.mediaSource)
  547. currentMediaSources.push(station.currentSong.mediaSource);
  548. // Block for experiment: queue_autofill_skip_last_x_played
  549. if (config.has(`experimental.queue_autofill_skip_last_x_played.${stationId}`)) {
  550. const redisList = `experimental:queue_autofill_skip_last_x_played:${stationId}`;
  551. // Get list of last x youtube video's played, to make sure they can't be autofilled
  552. const listOfYoutubeIds = await CacheModule.runJob("LRANGE", { key: redisList }, this);
  553. currentMediaSources = [...currentMediaSources, ...listOfYoutubeIds];
  554. }
  555. // Block for experiment: weight_stations
  556. if (
  557. config.has(`experimental.weight_stations.${stationId}`) &&
  558. !!config.get(`experimental.weight_stations.${stationId}`)
  559. ) {
  560. const weightTagName =
  561. config.get(`experimental.weight_stations.${stationId}`) === true
  562. ? "weight"
  563. : config.get(`experimental.weight_stations.${stationId}`);
  564. const weightMap = {};
  565. const getYoutubeIds = playlistSongs
  566. .map(playlistSong => playlistSong.mediaSource)
  567. .filter(mediaSource => currentMediaSources.indexOf(mediaSource) === -1);
  568. const { songs } = await SongsModule.runJob("GET_SONGS", { mediaSources: getYoutubeIds });
  569. const weightRegex = new RegExp(`${weightTagName}\\[(\\d+)\\]`);
  570. songs.forEach(song => {
  571. let weight = 1;
  572. song.tags.forEach(tag => {
  573. const regexResponse = weightRegex.exec(tag);
  574. if (regexResponse) weight = Number(regexResponse[1]);
  575. });
  576. if (Number.isNaN(weight)) weight = 1;
  577. weight = Math.round(weight);
  578. weight = Math.max(1, weight);
  579. weight = Math.min(10000, weight);
  580. weightMap[song.mediaSource] = weight;
  581. });
  582. const adjustedPlaylistSongs = [];
  583. playlistSongs.forEach(playlistSong => {
  584. Array.from({ length: weightMap[playlistSong.mediaSource] }).forEach(() => {
  585. adjustedPlaylistSongs.push(playlistSong);
  586. });
  587. });
  588. const { array } = await UtilsModule.runJob(
  589. "SHUFFLE",
  590. { array: adjustedPlaylistSongs },
  591. this
  592. );
  593. playlistSongs = array;
  594. }
  595. playlistSongs.every(song => {
  596. if (songsToAdd.length >= songsStillNeeded) return false;
  597. // Skip Spotify songs
  598. if (song.mediaSource.startsWith("spotify:")) return true;
  599. // Skip SoundCloud songs if Soundcloud isn't enabled
  600. if (song.mediaSource.startsWith("soundcloud:") && !config.get("experimental.soundcloud"))
  601. return true;
  602. // Skip songs already in songsToAdd
  603. if (songsToAdd.find(songToAdd => songToAdd.mediaSource === song.mediaSource)) return true;
  604. // Skip songs already in the queue
  605. if (currentMediaSources.indexOf(song.mediaSource) !== -1) return true;
  606. lastSongAdded = song;
  607. songsToAdd.push(song);
  608. return true;
  609. });
  610. let { currentSongIndex } = station;
  611. if (station.autofill.mode === "sequential" && lastSongAdded) {
  612. const indexOfLastSong = _playlistSongs
  613. .map(song => song.mediaSource)
  614. .indexOf(lastSongAdded.mediaSource);
  615. if (indexOfLastSong !== -1) currentSongIndex = indexOfLastSong;
  616. }
  617. return { currentSongs, songsToAdd, currentSongIndex };
  618. },
  619. ({ currentSongs, songsToAdd, currentSongIndex }, next) => {
  620. const songs = [];
  621. async.eachLimit(
  622. songsToAdd.map(song => song.mediaSource),
  623. 2,
  624. (mediaSource, next) => {
  625. MediaModule.runJob("GET_MEDIA", { mediaSource }, this)
  626. .then(response => {
  627. const { song } = response;
  628. const { _id, title, artists, thumbnail, duration, skipDuration, verified } =
  629. song;
  630. songs.push({
  631. _id,
  632. mediaSource,
  633. title,
  634. artists,
  635. thumbnail,
  636. duration,
  637. skipDuration,
  638. verified
  639. });
  640. next();
  641. })
  642. .catch(next);
  643. },
  644. err => {
  645. if (err) next(err);
  646. else {
  647. const newSongsToAdd = songsToAdd.map(song =>
  648. songs.find(newSong => newSong.mediaSource === song.mediaSource)
  649. );
  650. next(null, currentSongs, newSongsToAdd, currentSongIndex);
  651. }
  652. }
  653. );
  654. },
  655. (currentSongs, songsToAdd, currentSongIndex, next) => {
  656. const newPlaylist = [...currentSongs, ...songsToAdd].map(song => {
  657. if (!song._id) song._id = null;
  658. if (!song.requestedAt) song.requestedAt = Date.now();
  659. if (!song.requestedType) song.requestedType = "autofill";
  660. return song;
  661. });
  662. next(null, newPlaylist, currentSongIndex);
  663. },
  664. (newPlaylist, currentSongIndex, next) => {
  665. StationsModule.stationModel.updateOne(
  666. { _id: stationId },
  667. { $set: { queue: newPlaylist, currentSongIndex } },
  668. { runValidators: true },
  669. err => {
  670. if (err) next(err);
  671. else
  672. StationsModule.runJob(
  673. "UPDATE_STATION",
  674. {
  675. stationId
  676. },
  677. this
  678. )
  679. .then(() => {
  680. next(null);
  681. })
  682. .catch(next);
  683. }
  684. );
  685. }
  686. ],
  687. err => {
  688. if (err) reject(err);
  689. else resolve();
  690. }
  691. );
  692. });
  693. }
  694. /**
  695. * Gets next station song
  696. *
  697. * @param {object} payload - object that contains the payload
  698. * @param {string} payload.stationId - the id of the station
  699. * @returns {Promise} - returns a promise (resolve, reject)
  700. */
  701. GET_NEXT_STATION_SONG(payload) {
  702. return new Promise((resolve, reject) => {
  703. const { stationId } = payload;
  704. async.waterfall(
  705. [
  706. next => {
  707. StationsModule.runJob("GET_STATION", { stationId }, this)
  708. .then(station => {
  709. next(null, station);
  710. })
  711. .catch(next);
  712. },
  713. (station, next) => {
  714. if (station.queue.length === 0) next("No songs available.");
  715. else {
  716. next(null, station.queue[0]);
  717. }
  718. },
  719. (queueSong, next) => {
  720. MediaModule.runJob(
  721. "GET_MEDIA",
  722. {
  723. mediaSource: queueSong.mediaSource
  724. },
  725. this
  726. )
  727. .then(response => {
  728. const { song } = response;
  729. const {
  730. _id,
  731. mediaSource,
  732. title,
  733. skipDuration,
  734. artists,
  735. thumbnail,
  736. duration,
  737. verified
  738. } = song;
  739. next(null, {
  740. _id,
  741. mediaSource,
  742. title,
  743. skipDuration,
  744. artists,
  745. thumbnail,
  746. duration,
  747. verified,
  748. requestedAt: queueSong.requestedAt,
  749. requestedBy: queueSong.requestedBy,
  750. requestedType: queueSong.requestedType,
  751. likes: song.likes || 0,
  752. dislikes: song.dislikes || 0
  753. });
  754. })
  755. .catch(next);
  756. }
  757. ],
  758. (err, song) => {
  759. if (err) reject(err);
  760. else resolve({ song });
  761. }
  762. );
  763. });
  764. }
  765. /**
  766. * Removes first station queue song
  767. *
  768. * @param {object} payload - object that contains the payload
  769. * @param {string} payload.stationId - the id of the station
  770. * @returns {Promise} - returns a promise (resolve, reject)
  771. */
  772. REMOVE_FIRST_QUEUE_SONG(payload) {
  773. return new Promise((resolve, reject) => {
  774. const { stationId } = payload;
  775. async.waterfall(
  776. [
  777. next => {
  778. StationsModule.stationModel.updateOne(
  779. { _id: stationId },
  780. { $pop: { queue: -1 } },
  781. { runValidators: true },
  782. err => {
  783. if (err) next(err);
  784. else
  785. StationsModule.runJob(
  786. "UPDATE_STATION",
  787. {
  788. stationId
  789. },
  790. this
  791. )
  792. .then(() => {
  793. next(null);
  794. })
  795. .catch(next);
  796. }
  797. );
  798. }
  799. ],
  800. err => {
  801. if (err) reject(err);
  802. else resolve();
  803. }
  804. );
  805. });
  806. }
  807. /**
  808. * Process vote to skips for a station
  809. *
  810. * @param {object} payload - object that contains the payload
  811. * @param {string} payload.stationId - the id of the station to process
  812. * @returns {Promise} - returns a promise (resolve, reject)
  813. */
  814. PROCESS_SKIP_VOTES(payload) {
  815. return new Promise((resolve, reject) => {
  816. StationsModule.log("INFO", `Processing skip votes for station ${payload.stationId}.`);
  817. async.waterfall(
  818. [
  819. next => {
  820. StationsModule.runJob(
  821. "GET_STATION",
  822. {
  823. stationId: payload.stationId
  824. },
  825. this
  826. )
  827. .then(station => next(null, station))
  828. .catch(next);
  829. },
  830. (station, next) => {
  831. if (!station) return next("Station not found.");
  832. return next(null, station);
  833. },
  834. (station, next) => {
  835. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: `station.${station._id}` }, this)
  836. .then(sockets => next(null, station, sockets))
  837. .catch(next);
  838. },
  839. (station, sockets, next) => {
  840. const skipVotes =
  841. station.currentSong && station.currentSong.skipVotes
  842. ? station.currentSong.skipVotes.length
  843. : 0;
  844. let shouldSkip = false;
  845. if (skipVotes === 0) {
  846. if (!station.paused && !station.currentSong && station.queue.length > 0) shouldSkip = true;
  847. return next(null, shouldSkip);
  848. }
  849. const users = [];
  850. return async.each(
  851. sockets,
  852. (socketId, next) => {
  853. WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this)
  854. .then(socket => {
  855. if (socket && socket.session && socket.session.userId) {
  856. if (
  857. !users.includes(socket.session.userId) &&
  858. (socket.session.stationState !== "participate" ||
  859. station.currentSong.skipVotes.includes(socket.session.userId))
  860. )
  861. users.push(socket.session.userId);
  862. }
  863. return next();
  864. })
  865. .catch(next);
  866. },
  867. err => {
  868. if (err) return next(err);
  869. if (
  870. !station.paused &&
  871. Math.min(skipVotes, users.length) / users.length >= station.skipVoteThreshold / 100
  872. )
  873. shouldSkip = true;
  874. return next(null, shouldSkip);
  875. }
  876. );
  877. },
  878. (shouldSkip, next) => {
  879. if (shouldSkip)
  880. StationsModule.runJob(
  881. "SKIP_STATION",
  882. {
  883. stationId: payload.stationId,
  884. natural: false,
  885. skipReason: "vote_skip"
  886. },
  887. this
  888. )
  889. .then(() => next())
  890. .catch(next);
  891. else next();
  892. }
  893. ],
  894. err => {
  895. if (err) reject(err);
  896. else resolve();
  897. }
  898. );
  899. });
  900. }
  901. /**
  902. * Creates a station history item
  903. *
  904. * @param {object} payload - object containing the payload
  905. * @param {string} payload.stationId - the station id to create the history item for
  906. * @param {object} payload.currentSong - the song to create the history item for
  907. * @param {string} payload.skipReason - the reason the song was skipped
  908. * @param {string} payload.skippedAt - the date/time the song was skipped
  909. * @returns {Promise} - returns a promise (resolve, reject)
  910. */
  911. async ADD_STATION_HISTORY_ITEM(payload) {
  912. if (!config.get("experimental.station_history")) throw new Error("Station history is not enabled");
  913. const { stationId, currentSong, skipReason, skippedAt } = payload;
  914. let document = await StationsModule.stationHistoryModel.create({
  915. stationId,
  916. type: "song_played",
  917. payload: {
  918. song: currentSong,
  919. skippedAt,
  920. skipReason
  921. }
  922. });
  923. if (!document) return;
  924. document = document._doc;
  925. delete document.__v;
  926. delete document.documentVersion;
  927. WSModule.runJob("EMIT_TO_ROOM", {
  928. room: `station.${stationId}`,
  929. args: ["event:station.history.new", { data: { historyItem: document } }]
  930. });
  931. WSModule.runJob("EMIT_TO_ROOM", {
  932. room: `manage-station.${stationId}`,
  933. args: ["event:manageStation.history.new", { data: { stationId, historyItem: document } }]
  934. });
  935. }
  936. /**
  937. * Skips a station
  938. *
  939. * @param {object} payload - object that contains the payload
  940. * @param {string} payload.stationId - the id of the station to skip
  941. * @param {string} payload.natural - whether to skip naturally or forcefully
  942. * @param {string} payload.skipReason - if it was skipped via force skip or via vote skipping or if it was natural
  943. * @returns {Promise} - returns a promise (resolve, reject)
  944. */
  945. SKIP_STATION(payload) {
  946. return new Promise((resolve, reject) => {
  947. StationsModule.log("INFO", `Skipping station ${payload.stationId}.`);
  948. StationsModule.log("STATION_ISSUE", `SKIP_STATION_CB - Station ID: ${payload.stationId}.`);
  949. async.waterfall(
  950. [
  951. // Clears up any existing timers that would skip the station if the song ends
  952. next => {
  953. NotificationsModule.runJob("UNSCHEDULE", {
  954. name: `stations.nextSong?id=${payload.stationId}`
  955. })
  956. .then(() => {
  957. next();
  958. })
  959. .catch(next);
  960. },
  961. // Gets the station object
  962. next => {
  963. StationsModule.runJob(
  964. "GET_STATION",
  965. {
  966. stationId: payload.stationId
  967. },
  968. this
  969. )
  970. .then(station => next(null, station))
  971. .catch(next);
  972. },
  973. (station, next) => {
  974. if (!station) return next("Station not found.");
  975. if (!config.get("experimental.station_history")) return next(null, station);
  976. const { currentSong } = station;
  977. if (!currentSong || !currentSong.mediaSource) return next(null, station);
  978. const stationId = station._id;
  979. const skippedAt = new Date();
  980. const { skipReason } = payload;
  981. return StationsModule.runJob(
  982. "ADD_STATION_HISTORY_ITEM",
  983. {
  984. stationId,
  985. currentSong,
  986. skippedAt,
  987. skipReason
  988. },
  989. this
  990. ).finally(() => {
  991. next(null, station);
  992. });
  993. },
  994. (station, next) => {
  995. if (station.autofill.enabled)
  996. return StationsModule.runJob("AUTOFILL_STATION", { stationId: station._id }, this)
  997. .then(() => next(null, station))
  998. .catch(err => {
  999. if (
  1000. err === "Autofill is disabled in this station" ||
  1001. err === "Autofill limit reached"
  1002. )
  1003. return next(null, station);
  1004. return next(err);
  1005. });
  1006. return next(null, station);
  1007. },
  1008. (station, next) => {
  1009. StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)
  1010. .then(response => {
  1011. StationsModule.runJob("REMOVE_FIRST_QUEUE_SONG", { stationId: station._id }, this)
  1012. .then(() => {
  1013. next(null, response.song, station);
  1014. })
  1015. .catch(next);
  1016. })
  1017. .catch(err => {
  1018. if (err === "No songs available.") next(null, null, station);
  1019. else next(err);
  1020. });
  1021. },
  1022. async (song, station) => {
  1023. const $set = {};
  1024. if (song === null) $set.currentSong = null;
  1025. else {
  1026. // Block for experiment: queue_autofill_skip_last_x_played
  1027. if (config.has(`experimental.queue_autofill_skip_last_x_played.${payload.stationId}`)) {
  1028. const redisList = `experimental:queue_autofill_skip_last_x_played:${payload.stationId}`;
  1029. const maxListLength = Number(
  1030. config.get(`experimental.queue_autofill_skip_last_x_played.${payload.stationId}`)
  1031. );
  1032. // Add mediaSource to list for this station in Redis list
  1033. await CacheModule.runJob(
  1034. "LPUSH",
  1035. {
  1036. key: `experimental:queue_autofill_skip_last_x_played:${payload.stationId}`,
  1037. value: song.mediaSource
  1038. },
  1039. this
  1040. );
  1041. const currentListLength = await CacheModule.runJob("LLEN", { key: redisList }, this);
  1042. // Removes oldest mediaSource from list for this station in Redis list
  1043. if (currentListLength > maxListLength) {
  1044. const amount = currentListLength - maxListLength;
  1045. const promises = Array.from({ length: amount }).map(() =>
  1046. CacheModule.runJob(
  1047. "RPOP",
  1048. {
  1049. key: `experimental:queue_autofill_skip_last_x_played:${payload.stationId}`
  1050. },
  1051. this
  1052. )
  1053. );
  1054. await Promise.all(promises);
  1055. }
  1056. }
  1057. $set.currentSong = {
  1058. _id: song._id,
  1059. mediaSource: song.mediaSource,
  1060. title: song.title,
  1061. artists: song.artists,
  1062. duration: song.duration,
  1063. skipDuration: song.skipDuration,
  1064. thumbnail: song.thumbnail,
  1065. requestedAt: song.requestedAt,
  1066. requestedBy: song.requestedBy,
  1067. requestedType: song.requestedType,
  1068. verified: song.verified
  1069. };
  1070. }
  1071. $set.startedAt = Date.now();
  1072. $set.timePaused = 0;
  1073. if (station.paused) $set.pausedAt = Date.now();
  1074. return { $set, station };
  1075. },
  1076. ({ $set, station }, next) => {
  1077. StationsModule.stationModel.updateOne({ _id: station._id }, { $set }, err => {
  1078. if (err) return next(err);
  1079. return StationsModule.runJob("UPDATE_STATION", { stationId: station._id }, this)
  1080. .then(station => {
  1081. next(null, station);
  1082. })
  1083. .catch(next);
  1084. });
  1085. },
  1086. (station, next) => {
  1087. if (station.currentSong !== null && station.currentSong.mediaSource !== undefined) {
  1088. station.currentSong.skipVotes = 0;
  1089. }
  1090. next(null, station);
  1091. },
  1092. (station, next) => {
  1093. if (station.autofill.enabled)
  1094. return StationsModule.runJob("AUTOFILL_STATION", { stationId: station._id }, this)
  1095. .then(() => next(null, station))
  1096. .catch(err => {
  1097. if (
  1098. err === "Autofill is disabled in this station" ||
  1099. err === "Autofill limit reached"
  1100. )
  1101. return next(null, station);
  1102. return next(err);
  1103. });
  1104. return next(null, station);
  1105. },
  1106. (station, next) =>
  1107. StationsModule.runJob("UPDATE_STATION", { stationId: station._id }, this)
  1108. .then(station => {
  1109. CacheModule.runJob("PUB", {
  1110. channel: "station.queueUpdate",
  1111. value: payload.stationId
  1112. })
  1113. .then()
  1114. .catch();
  1115. next(null, station);
  1116. })
  1117. .catch(next)
  1118. ],
  1119. async (err, station) => {
  1120. if (err === "Autofill limit reached") return resolve({ station });
  1121. if (err) {
  1122. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1123. StationsModule.log("ERROR", `Skipping station "${payload.stationId}" failed. "${err}"`);
  1124. return reject(new Error(err));
  1125. }
  1126. // TODO Pub/Sub this
  1127. const { currentSong } = station;
  1128. WSModule.runJob("EMIT_TO_ROOM", {
  1129. room: `station.${station._id}`,
  1130. args: [
  1131. "event:station.nextSong",
  1132. {
  1133. data: {
  1134. currentSong,
  1135. startedAt: station.startedAt,
  1136. paused: station.paused,
  1137. timePaused: 0,
  1138. natural: payload.natural
  1139. }
  1140. }
  1141. ]
  1142. });
  1143. WSModule.runJob("EMIT_TO_ROOM", {
  1144. room: `manage-station.${station._id}`,
  1145. args: ["event:station.nextSong", { data: { stationId: station._id, currentSong } }]
  1146. });
  1147. if (station.privacy === "public")
  1148. WSModule.runJob("EMIT_TO_ROOM", {
  1149. room: "home",
  1150. args: ["event:station.nextSong", { data: { stationId: station._id, currentSong } }]
  1151. });
  1152. else {
  1153. const sockets = await WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: "home" }, this);
  1154. sockets.forEach(async socketId => {
  1155. const socket = await WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId });
  1156. if (!socket) return;
  1157. const { session } = socket;
  1158. if (session.sessionId) {
  1159. CacheModule.runJob("HGET", { table: "sessions", key: session.sessionId }).then(
  1160. session => {
  1161. const dispatchNextSong = () => {
  1162. socket.dispatch("event:station.nextSong", {
  1163. data: {
  1164. stationId: station._id,
  1165. currentSong
  1166. }
  1167. });
  1168. };
  1169. hasPermission("stations.index", session, station._id)
  1170. .then(() => dispatchNextSong())
  1171. .catch(() => {
  1172. hasPermission("stations.index.other", session)
  1173. .then(() => dispatchNextSong())
  1174. .catch(() => {});
  1175. });
  1176. }
  1177. );
  1178. }
  1179. });
  1180. }
  1181. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: `station.${station._id}` }).then(sockets => {
  1182. if (station.currentSong !== null && station.currentSong.mediaSource !== undefined) {
  1183. WSModule.runJob("SOCKETS_JOIN_SONG_ROOM", {
  1184. sockets,
  1185. room: `song.${station.currentSong.mediaSource}`
  1186. });
  1187. if (!station.paused) {
  1188. NotificationsModule.runJob("SCHEDULE", {
  1189. name: `stations.nextSong?id=${station._id}`,
  1190. time: station.currentSong.duration * 1000,
  1191. station
  1192. });
  1193. }
  1194. } else WSModule.runJob("SOCKETS_LEAVE_SONG_ROOMS", { sockets });
  1195. });
  1196. return resolve({ station });
  1197. }
  1198. );
  1199. });
  1200. }
  1201. /**
  1202. * Checks if a user can view/access a station
  1203. *
  1204. * @param {object} payload - object that contains the payload
  1205. * @param {object} payload.station - the station object of the station in question
  1206. * @param {string} payload.userId - the id of the user in question
  1207. * @returns {Promise} - returns a promise (resolve, reject)
  1208. */
  1209. CAN_USER_VIEW_STATION(payload) {
  1210. return new Promise((resolve, reject) => {
  1211. async.waterfall(
  1212. [
  1213. next => {
  1214. if (payload.station.privacy === "public" || payload.station.privacy === "unlisted")
  1215. return next(true);
  1216. if (!payload.userId) return next("Not allowed");
  1217. return next();
  1218. },
  1219. next => {
  1220. hasPermission("stations.view", payload.userId, payload.station._id)
  1221. .then(() => next(true))
  1222. .catch(() => next("Not allowed"));
  1223. }
  1224. ],
  1225. async errOrResult => {
  1226. if (errOrResult !== true && errOrResult !== "Not allowed") {
  1227. errOrResult = await UtilsModule.runJob(
  1228. "GET_ERROR",
  1229. {
  1230. error: errOrResult
  1231. },
  1232. this
  1233. );
  1234. reject(new Error(errOrResult));
  1235. } else {
  1236. resolve(errOrResult === true);
  1237. }
  1238. }
  1239. );
  1240. });
  1241. }
  1242. /**
  1243. * Checks if a user has favorited a station or not
  1244. *
  1245. * @param {object} payload - object that contains the payload
  1246. * @param {object} payload.stationId - the id of the station in question
  1247. * @param {string} payload.userId - the id of the user in question
  1248. * @returns {Promise} - returns a promise (resolve, reject)
  1249. */
  1250. HAS_USER_FAVORITED_STATION(payload) {
  1251. return new Promise((resolve, reject) => {
  1252. async.waterfall(
  1253. [
  1254. next => {
  1255. DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(userModel => {
  1256. userModel.findOne({ _id: payload.userId }, next);
  1257. });
  1258. },
  1259. (user, next) => {
  1260. if (!user) return next("User not found.");
  1261. if (user.favoriteStations.indexOf(payload.stationId) !== -1) return next(null, true);
  1262. return next(null, false);
  1263. }
  1264. ],
  1265. async (err, isStationFavorited) => {
  1266. if (err && err !== true) {
  1267. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1268. return reject(new Error(err));
  1269. }
  1270. return resolve(isStationFavorited);
  1271. }
  1272. );
  1273. });
  1274. }
  1275. /**
  1276. * Returns a list of sockets in a room that can and can't know about a station
  1277. *
  1278. * @param {object} payload - the payload object
  1279. * @param {object} payload.station - the station object
  1280. * @param {string} payload.room - the websockets room to get the sockets from
  1281. * @returns {Promise} - returns a promise (resolve, reject)
  1282. */
  1283. GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION(payload) {
  1284. return new Promise((resolve, reject) => {
  1285. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: payload.room }, this)
  1286. .then(socketIds => {
  1287. const sockets = [];
  1288. async.eachLimit(
  1289. socketIds,
  1290. 1,
  1291. (socketId, next) => {
  1292. WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this)
  1293. .then(socket => {
  1294. if (socket) sockets.push(socket);
  1295. next();
  1296. })
  1297. .catch(err => {
  1298. reject(err);
  1299. });
  1300. },
  1301. err => {
  1302. if (err) reject(err);
  1303. else {
  1304. let socketsThatCan = [];
  1305. const socketsThatCannot = [];
  1306. if (payload.station.privacy === "public") {
  1307. socketsThatCan = sockets;
  1308. resolve({ socketsThatCan, socketsThatCannot });
  1309. } else {
  1310. async.eachLimit(
  1311. sockets,
  1312. 1,
  1313. (socket, next) => {
  1314. if (!(socket.session && socket.session.sessionId)) {
  1315. socketsThatCannot.push(socket);
  1316. next();
  1317. } else
  1318. hasPermission("stations.view", socket.session, payload.station._id)
  1319. .then(() => {
  1320. socketsThatCan.push(socket);
  1321. next();
  1322. })
  1323. .catch(() => {
  1324. socketsThatCannot.push(socket);
  1325. next();
  1326. });
  1327. },
  1328. err => {
  1329. if (err) reject(err);
  1330. else resolve({ socketsThatCan, socketsThatCannot });
  1331. }
  1332. );
  1333. }
  1334. }
  1335. }
  1336. );
  1337. })
  1338. .catch(reject);
  1339. });
  1340. }
  1341. /**
  1342. * Adds a playlist to autofill a station
  1343. *
  1344. * @param {object} payload - object that contains the payload
  1345. * @param {object} payload.stationId - the id of the station
  1346. * @param {object} payload.playlistId - the id of the playlist
  1347. * @returns {Promise} - returns a promise (resolve, reject)
  1348. */
  1349. AUTOFILL_PLAYLIST(payload) {
  1350. return new Promise((resolve, reject) => {
  1351. async.waterfall(
  1352. [
  1353. next => {
  1354. if (!payload.stationId) next("Please specify a station id");
  1355. else if (!payload.playlistId) next("Please specify a playlist id");
  1356. else next();
  1357. },
  1358. next => {
  1359. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1360. .then(station => {
  1361. next(null, station);
  1362. })
  1363. .catch(next);
  1364. },
  1365. (station, next) => {
  1366. if (station.playlist === payload.playlistId) next("You cannot autofill the station playlist");
  1367. else if (station.autofill.playlists.indexOf(payload.playlistId) !== -1)
  1368. next("This playlist is already autofilling");
  1369. else if (station.blacklist.indexOf(payload.playlistId) !== -1)
  1370. next("This playlist is currently blacklisted");
  1371. else
  1372. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)
  1373. .then(() => {
  1374. next(null);
  1375. })
  1376. .catch(next);
  1377. },
  1378. next => {
  1379. StationsModule.stationModel.updateOne(
  1380. { _id: payload.stationId },
  1381. { $push: { "autofill.playlists": payload.playlistId } },
  1382. next
  1383. );
  1384. },
  1385. (res, next) => {
  1386. StationsModule.runJob(
  1387. "UPDATE_STATION",
  1388. {
  1389. stationId: payload.stationId
  1390. },
  1391. this
  1392. )
  1393. .then(() => {
  1394. next();
  1395. })
  1396. .catch(next);
  1397. }
  1398. ],
  1399. async err => {
  1400. if (err && err !== true) {
  1401. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1402. return reject(new Error(err));
  1403. }
  1404. return resolve();
  1405. }
  1406. );
  1407. });
  1408. }
  1409. /**
  1410. * Removes a playlist from autofill
  1411. *
  1412. * @param {object} payload - object that contains the payload
  1413. * @param {object} payload.stationId - the id of the station
  1414. * @param {object} payload.playlistId - the id of the playlist
  1415. * @returns {Promise} - returns a promise (resolve, reject)
  1416. */
  1417. REMOVE_AUTOFILL_PLAYLIST(payload) {
  1418. return new Promise((resolve, reject) => {
  1419. async.waterfall(
  1420. [
  1421. next => {
  1422. if (!payload.stationId) next("Please specify a station id");
  1423. else if (!payload.playlistId) next("Please specify a playlist id");
  1424. else next();
  1425. },
  1426. next => {
  1427. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1428. .then(station => {
  1429. next(null, station);
  1430. })
  1431. .catch(next);
  1432. },
  1433. (station, next) => {
  1434. if (station.autofill.playlists.indexOf(payload.playlistId) === -1)
  1435. next("This playlist isn't autofilling");
  1436. else next();
  1437. },
  1438. next => {
  1439. StationsModule.stationModel.updateOne(
  1440. { _id: payload.stationId },
  1441. { $pull: { "autofill.playlists": payload.playlistId } },
  1442. next
  1443. );
  1444. },
  1445. (res, next) => {
  1446. StationsModule.runJob(
  1447. "UPDATE_STATION",
  1448. {
  1449. stationId: payload.stationId
  1450. },
  1451. this
  1452. )
  1453. .then(() => {
  1454. next();
  1455. })
  1456. .catch(next);
  1457. }
  1458. ],
  1459. async err => {
  1460. if (err && err !== true) {
  1461. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1462. return reject(new Error(err));
  1463. }
  1464. return resolve();
  1465. }
  1466. );
  1467. });
  1468. }
  1469. /**
  1470. * Add a playlist to station blacklist
  1471. *
  1472. * @param {object} payload - object that contains the payload
  1473. * @param {object} payload.stationId - the id of the station
  1474. * @param {object} payload.playlistId - the id of the playlist
  1475. * @returns {Promise} - returns a promise (resolve, reject)
  1476. */
  1477. BLACKLIST_PLAYLIST(payload) {
  1478. return new Promise((resolve, reject) => {
  1479. async.waterfall(
  1480. [
  1481. next => {
  1482. if (!payload.stationId) next("Please specify a station id");
  1483. else if (!payload.playlistId) next("Please specify a playlist id");
  1484. else next();
  1485. },
  1486. next => {
  1487. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1488. .then(station => {
  1489. next(null, station);
  1490. })
  1491. .catch(next);
  1492. },
  1493. (station, next) => {
  1494. if (station.playlist === payload.playlistId) next("You cannot blacklist the station playlist");
  1495. else if (station.blacklist.indexOf(payload.playlistId) !== -1)
  1496. next("This playlist is already blacklisted");
  1497. else if (station.autofill.playlists.indexOf(payload.playlistId) !== -1)
  1498. next(
  1499. "This playlist is currently autofilling, please remove it from there before blacklisting it"
  1500. );
  1501. else
  1502. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)
  1503. .then(() => {
  1504. next(null);
  1505. })
  1506. .catch(next);
  1507. },
  1508. next => {
  1509. StationsModule.stationModel.updateOne(
  1510. { _id: payload.stationId },
  1511. { $push: { blacklist: payload.playlistId } },
  1512. next
  1513. );
  1514. },
  1515. (res, next) => {
  1516. StationsModule.runJob(
  1517. "UPDATE_STATION",
  1518. {
  1519. stationId: payload.stationId
  1520. },
  1521. this
  1522. )
  1523. .then(() => {
  1524. next();
  1525. })
  1526. .catch(next);
  1527. }
  1528. ],
  1529. async err => {
  1530. if (err && err !== true) {
  1531. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1532. return reject(new Error(err));
  1533. }
  1534. return resolve();
  1535. }
  1536. );
  1537. });
  1538. }
  1539. /**
  1540. * Remove a playlist from station blacklist
  1541. *
  1542. * @param {object} payload - object that contains the payload
  1543. * @param {object} payload.stationId - the id of the station
  1544. * @param {object} payload.playlistId - the id of the playlist
  1545. * @returns {Promise} - returns a promise (resolve, reject)
  1546. */
  1547. REMOVE_BLACKLISTED_PLAYLIST(payload) {
  1548. return new Promise((resolve, reject) => {
  1549. async.waterfall(
  1550. [
  1551. next => {
  1552. if (!payload.stationId) next("Please specify a station id");
  1553. else if (!payload.playlistId) next("Please specify a playlist id");
  1554. else next();
  1555. },
  1556. next => {
  1557. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1558. .then(station => {
  1559. next(null, station);
  1560. })
  1561. .catch(next);
  1562. },
  1563. (station, next) => {
  1564. if (station.blacklist.indexOf(payload.playlistId) === -1)
  1565. next("This playlist isn't blacklisted");
  1566. else next();
  1567. },
  1568. next => {
  1569. StationsModule.stationModel.updateOne(
  1570. { _id: payload.stationId },
  1571. { $pull: { blacklist: payload.playlistId } },
  1572. next
  1573. );
  1574. },
  1575. (res, next) => {
  1576. StationsModule.runJob(
  1577. "UPDATE_STATION",
  1578. {
  1579. stationId: payload.stationId
  1580. },
  1581. this
  1582. )
  1583. .then(() => {
  1584. next();
  1585. })
  1586. .catch(next);
  1587. }
  1588. ],
  1589. async err => {
  1590. if (err && err !== true) {
  1591. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1592. return reject(new Error(err));
  1593. }
  1594. return resolve();
  1595. }
  1596. );
  1597. });
  1598. }
  1599. /**
  1600. * Removes autofilled or blacklisted playlist from a station
  1601. *
  1602. * @param {object} payload - object that contains the payload
  1603. * @param {string} payload.playlistId - the playlist id
  1604. * @returns {Promise} - returns promise (reject, resolve)
  1605. */
  1606. REMOVE_AUTOFILLED_OR_BLACKLISTED_PLAYLIST_FROM_STATIONS(payload) {
  1607. return new Promise((resolve, reject) => {
  1608. async.waterfall(
  1609. [
  1610. next => {
  1611. if (!payload.playlistId) next("Please specify a playlist id");
  1612. else next();
  1613. },
  1614. next => {
  1615. StationsModule.stationModel.updateMany(
  1616. {
  1617. $or: [{ "autofill.playlists": payload.playlistId }, { blacklist: payload.playlistId }]
  1618. },
  1619. {
  1620. $pull: {
  1621. "autofill.playlists": payload.playlistId,
  1622. blacklist: payload.playlistId
  1623. }
  1624. },
  1625. err => {
  1626. if (err) next(err);
  1627. else next();
  1628. }
  1629. );
  1630. }
  1631. ],
  1632. async err => {
  1633. if (err && err !== true) {
  1634. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1635. return reject(new Error(err));
  1636. }
  1637. return resolve();
  1638. }
  1639. );
  1640. });
  1641. }
  1642. /**
  1643. * Gets stations that autofill or blacklist a specific playlist
  1644. *
  1645. * @param {object} payload - object that contains the payload
  1646. * @param {string} payload.playlistId - the playlist id
  1647. * @returns {Promise} - returns promise (reject, resolve)
  1648. */
  1649. GET_STATIONS_THAT_AUTOFILL_OR_BLACKLIST_PLAYLIST(payload) {
  1650. return new Promise((resolve, reject) => {
  1651. DBModule.runJob(
  1652. "GET_MODEL",
  1653. {
  1654. modelName: "station"
  1655. },
  1656. this
  1657. ).then(stationModel => {
  1658. stationModel.find(
  1659. {
  1660. $or: [{ "autofill.playlists": payload.playlistId }, { blacklist: payload.playlistId }]
  1661. },
  1662. (err, stations) => {
  1663. if (err) reject(err);
  1664. else resolve({ stationIds: stations.map(station => station._id) });
  1665. }
  1666. );
  1667. });
  1668. });
  1669. }
  1670. /**
  1671. * Clears every queue
  1672. *
  1673. * @returns {Promise} - returns a promise (resolve, reject)
  1674. */
  1675. CLEAR_EVERY_STATION_QUEUE() {
  1676. return new Promise((resolve, reject) => {
  1677. async.waterfall(
  1678. [
  1679. next => {
  1680. StationsModule.stationModel.updateMany({}, { $set: { queue: [] } }, err => {
  1681. if (err) next(err);
  1682. else {
  1683. StationsModule.stationModel.find({}, (err, stations) => {
  1684. if (err) next(err);
  1685. else {
  1686. async.eachLimit(
  1687. stations,
  1688. 1,
  1689. (station, next) => {
  1690. this.publishProgress({
  1691. status: "update",
  1692. message: `Updating station "${station._id}"`
  1693. });
  1694. StationsModule.runJob("UPDATE_STATION", {
  1695. stationId: station._id
  1696. })
  1697. .then(() => next())
  1698. .catch(next);
  1699. CacheModule.runJob("PUB", {
  1700. channel: "station.queueUpdate",
  1701. value: station._id
  1702. })
  1703. .then()
  1704. .catch();
  1705. },
  1706. next
  1707. );
  1708. }
  1709. });
  1710. }
  1711. });
  1712. }
  1713. ],
  1714. err => {
  1715. if (err) reject(err);
  1716. else resolve();
  1717. }
  1718. );
  1719. });
  1720. }
  1721. /**
  1722. * Resets a station queue
  1723. *
  1724. * @param {object} payload - object that contains the payload
  1725. * @param {string} payload.stationId - the station id
  1726. * @returns {Promise} - returns a promise (resolve, reject)
  1727. */
  1728. RESET_QUEUE(payload) {
  1729. return new Promise((resolve, reject) => {
  1730. async.waterfall(
  1731. [
  1732. next => {
  1733. StationsModule.runJob(
  1734. "AUTOFILL_STATION",
  1735. { stationId: payload.stationId, ignoreExistingQueue: true },
  1736. this
  1737. )
  1738. .then(() => {
  1739. CacheModule.runJob("PUB", {
  1740. channel: "station.queueUpdate",
  1741. value: payload.stationId
  1742. })
  1743. .then()
  1744. .catch();
  1745. next();
  1746. })
  1747. .catch(err => {
  1748. if (err === "Autofill is disabled in this station" || err === "Autofill limit reached")
  1749. StationsModule.stationModel
  1750. .updateOne({ _id: payload.stationId }, { $set: { queue: [] } }, this)
  1751. .then(() => next())
  1752. .catch(next);
  1753. else next(err);
  1754. });
  1755. },
  1756. next => {
  1757. StationsModule.runJob("UPDATE_STATION", { stationId: payload.stationId }, this)
  1758. .then(() => next())
  1759. .catch(next);
  1760. },
  1761. next => {
  1762. CacheModule.runJob("PUB", {
  1763. channel: "station.queueUpdate",
  1764. value: payload.stationId
  1765. })
  1766. .then(() => next())
  1767. .catch(next);
  1768. }
  1769. ],
  1770. err => {
  1771. if (err) reject(err);
  1772. else resolve();
  1773. }
  1774. );
  1775. });
  1776. }
  1777. /**
  1778. * Add to a station queue
  1779. *
  1780. * @param {object} payload - object that contains the payload
  1781. * @param {string} payload.stationId - the station id
  1782. * @param {string} payload.mediaSource - the media source
  1783. * @param {string} payload.requestUser - the requesting user id
  1784. * @param {string} payload.requestType - the request type
  1785. * @returns {Promise} - returns a promise (resolve, reject)
  1786. */
  1787. ADD_TO_QUEUE(payload) {
  1788. return new Promise((resolve, reject) => {
  1789. const { stationId, mediaSource, requestUser, requestType } = payload;
  1790. async.waterfall(
  1791. [
  1792. next => {
  1793. StationsModule.runJob("GET_STATION", { stationId }, this)
  1794. .then(station => {
  1795. next(null, station);
  1796. })
  1797. .catch(next);
  1798. },
  1799. (station, next) => {
  1800. if (!station) return next("Station not found.");
  1801. if (!station.requests.enabled) return next("Requests are disabled in this station.");
  1802. if (mediaSource.startsWith("spotify:")) return next("Spotify playback is not supported.");
  1803. if (station.currentSong && station.currentSong.mediaSource === mediaSource)
  1804. return next("That song is currently playing.");
  1805. if (station.queue.find(song => song.mediaSource === mediaSource))
  1806. return next("That song is already in the queue.");
  1807. return next(null, station);
  1808. },
  1809. (station, next) => {
  1810. MediaModule.runJob("GET_MEDIA", { mediaSource }, this)
  1811. .then(response => {
  1812. const { song } = response;
  1813. const { _id, title, skipDuration, artists, thumbnail, duration, verified } = song;
  1814. next(
  1815. null,
  1816. {
  1817. _id,
  1818. mediaSource,
  1819. title,
  1820. skipDuration,
  1821. artists,
  1822. thumbnail,
  1823. duration,
  1824. verified
  1825. },
  1826. station
  1827. );
  1828. })
  1829. .catch(next);
  1830. },
  1831. (song, station, next) => {
  1832. const blacklist = [];
  1833. async.eachLimit(
  1834. station.blacklist,
  1835. 1,
  1836. (playlistId, next) => {
  1837. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }, this)
  1838. .then(playlist => {
  1839. blacklist.push(playlist);
  1840. next();
  1841. })
  1842. .catch(next);
  1843. },
  1844. err => {
  1845. next(err, song, station, blacklist);
  1846. }
  1847. );
  1848. },
  1849. (song, station, blacklist, next) => {
  1850. const blacklistedSongs = blacklist
  1851. .flatMap(blacklistedPlaylist => blacklistedPlaylist.songs)
  1852. .reduce(
  1853. (items, item) =>
  1854. items.find(x => x.mediaSource === item.mediaSource) ? [...items] : [...items, item],
  1855. []
  1856. );
  1857. if (blacklistedSongs.find(blacklistedSong => blacklistedSong.mediaSource === song.mediaSource))
  1858. next("That song is in an blacklisted playlist and cannot be played.");
  1859. else next(null, song, station);
  1860. },
  1861. (song, station, next) => {
  1862. song.requestedBy = requestUser;
  1863. song.requestedAt = Date.now();
  1864. song.requestedType = requestType;
  1865. if (station.queue.length === 0) return next(null, song, station);
  1866. if (
  1867. requestUser &&
  1868. station.queue.filter(queueSong => queueSong.requestedBy === song.requestedBy).length >=
  1869. station.requests.limit
  1870. )
  1871. return next(`The max amount of songs per user is ${station.requests.limit}.`);
  1872. return next(null, song, station);
  1873. },
  1874. // (song, station, next) => {
  1875. // song.requestedBy = session.userId;
  1876. // song.requestedAt = Date.now();
  1877. // let totalDuration = 0;
  1878. // station.queue.forEach(song => {
  1879. // totalDuration += song.duration;
  1880. // });
  1881. // if (totalDuration >= 3600 * 3) return next("The max length of the queue is 3 hours.");
  1882. // return next(null, song, station);
  1883. // },
  1884. // (song, station, next) => {
  1885. // if (station.queue.length === 0) return next(null, song, station);
  1886. // let totalDuration = 0;
  1887. // const userId = station.queue[station.queue.length - 1].requestedBy;
  1888. // station.queue.forEach(song => {
  1889. // if (userId === song.requestedBy) {
  1890. // totalDuration += song.duration;
  1891. // }
  1892. // });
  1893. // if (totalDuration >= 900) return next("The max length of songs per user is 15 minutes.");
  1894. // return next(null, song, station);
  1895. // },
  1896. // (song, station, next) => {
  1897. // if (station.queue.length === 0) return next(null, song);
  1898. // let totalSongs = 0;
  1899. // const userId = station.queue[station.queue.length - 1].requestedBy;
  1900. // station.queue.forEach(song => {
  1901. // if (userId === song.requestedBy) {
  1902. // totalSongs += 1;
  1903. // }
  1904. // });
  1905. // if (totalSongs <= 2) return next(null, song);
  1906. // if (totalSongs > 3)
  1907. // return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1908. // if (
  1909. // station.queue[station.queue.length - 2].requestedBy !== userId ||
  1910. // station.queue[station.queue.length - 3] !== userId
  1911. // )
  1912. // return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1913. // return next(null, song);
  1914. // },
  1915. (song, station, next) => {
  1916. const queueAddBeforeAutofilled = config.get(`experimental.queue_add_before_autofilled`);
  1917. if (
  1918. queueAddBeforeAutofilled === true ||
  1919. (Array.isArray(queueAddBeforeAutofilled) &&
  1920. queueAddBeforeAutofilled.indexOf(stationId) !== -1)
  1921. ) {
  1922. let position = station.queue.length;
  1923. if (station.autofill.enabled && station.queue.length >= station.autofill.limit) {
  1924. position = -station.autofill.limit;
  1925. }
  1926. StationsModule.stationModel.updateOne(
  1927. { _id: stationId },
  1928. {
  1929. $push: {
  1930. queue: {
  1931. $each: [song],
  1932. $position: position
  1933. }
  1934. }
  1935. },
  1936. { runValidators: true },
  1937. next
  1938. );
  1939. return;
  1940. }
  1941. StationsModule.stationModel.updateOne(
  1942. { _id: stationId },
  1943. { $push: { queue: song } },
  1944. { runValidators: true },
  1945. next
  1946. );
  1947. },
  1948. (res, next) => {
  1949. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1950. .then(() => next())
  1951. .catch(next);
  1952. },
  1953. next => {
  1954. CacheModule.runJob(
  1955. "PUB",
  1956. {
  1957. channel: "station.queueUpdate",
  1958. value: stationId
  1959. },
  1960. this
  1961. )
  1962. .then(() => next())
  1963. .catch(next);
  1964. }
  1965. ],
  1966. err => {
  1967. if (err) reject(err);
  1968. else resolve();
  1969. }
  1970. );
  1971. });
  1972. }
  1973. /**
  1974. * Remove from a station queue
  1975. *
  1976. * @param {object} payload - object that contains the payload
  1977. * @param {string} payload.stationId - the station id
  1978. * @param {string} payload.mediaSource - the media source
  1979. * @returns {Promise} - returns a promise (resolve, reject)
  1980. */
  1981. REMOVE_FROM_QUEUE(payload) {
  1982. return new Promise((resolve, reject) => {
  1983. const { stationId, mediaSource } = payload;
  1984. async.waterfall(
  1985. [
  1986. next => {
  1987. StationsModule.runJob("GET_STATION", { stationId }, this)
  1988. .then(station => {
  1989. next(null, station);
  1990. })
  1991. .catch(next);
  1992. },
  1993. (station, next) => {
  1994. if (!station) return next("Station not found.");
  1995. if (!station.queue.find(song => song.mediaSource === mediaSource))
  1996. return next("That song is not currently in the queue.");
  1997. return StationsModule.stationModel.updateOne(
  1998. { _id: stationId },
  1999. { $pull: { queue: { mediaSource } } },
  2000. next
  2001. );
  2002. },
  2003. (res, next) => {
  2004. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  2005. .then(station => {
  2006. if (station.autofill.enabled)
  2007. StationsModule.runJob("AUTOFILL_STATION", { stationId }, this)
  2008. .then(() => next())
  2009. .catch(err => {
  2010. if (
  2011. err === "Autofill is disabled in this station" ||
  2012. err === "Autofill limit reached"
  2013. )
  2014. return next();
  2015. return next(err);
  2016. });
  2017. else next();
  2018. })
  2019. .catch(next);
  2020. },
  2021. next =>
  2022. CacheModule.runJob(
  2023. "PUB",
  2024. {
  2025. channel: "station.queueUpdate",
  2026. value: stationId
  2027. },
  2028. this
  2029. )
  2030. .then(() => next())
  2031. .catch(next)
  2032. ],
  2033. err => {
  2034. if (err) reject(err);
  2035. else resolve();
  2036. }
  2037. );
  2038. });
  2039. }
  2040. /**
  2041. * Add DJ to station
  2042. *
  2043. * @param {object} payload - object that contains the payload
  2044. * @param {string} payload.stationId - the station id
  2045. * @param {string} payload.userId - the dj user id
  2046. * @returns {Promise} - returns a promise (resolve, reject)
  2047. */
  2048. ADD_DJ(payload) {
  2049. return new Promise((resolve, reject) => {
  2050. const { stationId, userId } = payload;
  2051. async.waterfall(
  2052. [
  2053. next => {
  2054. StationsModule.runJob("GET_STATION", { stationId }, this)
  2055. .then(station => {
  2056. next(null, station);
  2057. })
  2058. .catch(next);
  2059. },
  2060. (station, next) => {
  2061. if (!station) return next("Station not found.");
  2062. if (station.djs.find(dj => dj === userId)) return next("That user is already a DJ.");
  2063. return StationsModule.stationModel.updateOne(
  2064. { _id: stationId },
  2065. { $push: { djs: userId } },
  2066. next
  2067. );
  2068. },
  2069. (res, next) => {
  2070. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  2071. .then(() => next())
  2072. .catch(next);
  2073. },
  2074. next =>
  2075. CacheModule.runJob(
  2076. "PUB",
  2077. {
  2078. channel: "station.djs.added",
  2079. value: { stationId, userId }
  2080. },
  2081. this
  2082. )
  2083. .then(() => next())
  2084. .catch(next)
  2085. ],
  2086. err => {
  2087. if (err) reject(err);
  2088. else resolve();
  2089. }
  2090. );
  2091. });
  2092. }
  2093. /**
  2094. * Remove DJ from station
  2095. *
  2096. * @param {object} payload - object that contains the payload
  2097. * @param {string} payload.stationId - the station id
  2098. * @param {string} payload.userId - the dj user id
  2099. * @returns {Promise} - returns a promise (resolve, reject)
  2100. */
  2101. REMOVE_DJ(payload) {
  2102. return new Promise((resolve, reject) => {
  2103. const { stationId, userId } = payload;
  2104. async.waterfall(
  2105. [
  2106. next => {
  2107. StationsModule.runJob("GET_STATION", { stationId }, this)
  2108. .then(station => {
  2109. next(null, station);
  2110. })
  2111. .catch(next);
  2112. },
  2113. (station, next) => {
  2114. if (!station) return next("Station not found.");
  2115. if (!station.djs.find(dj => dj === userId)) return next("That user is not currently a DJ.");
  2116. return StationsModule.stationModel.updateOne(
  2117. { _id: stationId },
  2118. { $pull: { djs: userId } },
  2119. next
  2120. );
  2121. },
  2122. (res, next) => {
  2123. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  2124. .then(() => next())
  2125. .catch(next);
  2126. },
  2127. next =>
  2128. CacheModule.runJob(
  2129. "PUB",
  2130. {
  2131. channel: "station.djs.removed",
  2132. value: { stationId, userId }
  2133. },
  2134. this
  2135. )
  2136. .then(() => next())
  2137. .catch(next)
  2138. ],
  2139. err => {
  2140. if (err) reject(err);
  2141. else resolve();
  2142. }
  2143. );
  2144. });
  2145. }
  2146. }
  2147. export default new _StationsModule();