stations.js 60 KB

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