stations.js 60 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310
  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.mediaSource);
  514. const songsToAdd = [];
  515. let lastSongAdded = null;
  516. if (station.currentSong && station.currentSong.mediaSource)
  517. currentYoutubeIds.push(station.currentSong.mediaSource);
  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.mediaSource)
  537. .filter(mediaSource => currentYoutubeIds.indexOf(mediaSource) === -1);
  538. console.log(4343, getYoutubeIds);
  539. const { songs } = await SongsModule.runJob("GET_SONGS", { mediaSources: getYoutubeIds });
  540. const weightRegex = new RegExp(`${weightTagName}\\[(\\d+)\\]`);
  541. songs.forEach(song => {
  542. let weight = 1;
  543. song.tags.forEach(tag => {
  544. const regexResponse = weightRegex.exec(tag);
  545. if (regexResponse) weight = Number(regexResponse[1]);
  546. });
  547. if (Number.isNaN(weight)) weight = 1;
  548. weight = Math.round(weight);
  549. weight = Math.max(1, weight);
  550. weight = Math.min(10000, weight);
  551. weightMap[song.mediaSource] = weight;
  552. });
  553. const adjustedPlaylistSongs = [];
  554. playlistSongs.forEach(playlistSong => {
  555. Array.from({ length: weightMap[playlistSong.mediaSource] }).forEach(() => {
  556. adjustedPlaylistSongs.push(playlistSong);
  557. });
  558. });
  559. const { array } = await UtilsModule.runJob(
  560. "SHUFFLE",
  561. { array: adjustedPlaylistSongs },
  562. this
  563. );
  564. playlistSongs = array;
  565. }
  566. playlistSongs.every(song => {
  567. if (
  568. songsToAdd.length < songsStillNeeded &&
  569. currentYoutubeIds.indexOf(song.mediaSource) === -1 &&
  570. !songsToAdd.find(songToAdd => songToAdd.mediaSource === song.mediaSource)
  571. ) {
  572. lastSongAdded = song;
  573. songsToAdd.push(song);
  574. return true;
  575. }
  576. if (songsToAdd.length >= songsStillNeeded) return false;
  577. return true;
  578. });
  579. let { currentSongIndex } = station;
  580. if (station.autofill.mode === "sequential" && lastSongAdded) {
  581. const indexOfLastSong = _playlistSongs
  582. .map(song => song.mediaSource)
  583. .indexOf(lastSongAdded.mediaSource);
  584. if (indexOfLastSong !== -1) currentSongIndex = indexOfLastSong;
  585. }
  586. return { currentSongs, songsToAdd, currentSongIndex };
  587. },
  588. ({ currentSongs, songsToAdd, currentSongIndex }, next) => {
  589. const songs = [];
  590. async.eachLimit(
  591. songsToAdd.map(song => song.mediaSource),
  592. 2,
  593. (mediaSource, next) => {
  594. MediaModule.runJob("GET_MEDIA", { mediaSource }, this)
  595. .then(response => {
  596. const { song } = response;
  597. const { _id, title, artists, thumbnail, duration, skipDuration, verified } =
  598. song;
  599. songs.push({
  600. _id,
  601. mediaSource,
  602. title,
  603. artists,
  604. thumbnail,
  605. duration,
  606. skipDuration,
  607. verified
  608. });
  609. next();
  610. })
  611. .catch(next);
  612. },
  613. err => {
  614. if (err) next(err);
  615. else {
  616. const newSongsToAdd = songsToAdd.map(song =>
  617. songs.find(newSong => newSong.mediaSource === song.mediaSource)
  618. );
  619. next(null, currentSongs, newSongsToAdd, currentSongIndex);
  620. }
  621. }
  622. );
  623. },
  624. (currentSongs, songsToAdd, currentSongIndex, next) => {
  625. const newPlaylist = [...currentSongs, ...songsToAdd].map(song => {
  626. if (!song._id) song._id = null;
  627. if (!song.requestedAt) song.requestedAt = Date.now();
  628. if (!song.requestedType) song.requestedType = "autofill";
  629. return song;
  630. });
  631. next(null, newPlaylist, currentSongIndex);
  632. },
  633. (newPlaylist, currentSongIndex, next) => {
  634. StationsModule.stationModel.updateOne(
  635. { _id: stationId },
  636. { $set: { queue: newPlaylist, currentSongIndex } },
  637. { runValidators: true },
  638. err => {
  639. if (err) next(err);
  640. else
  641. StationsModule.runJob(
  642. "UPDATE_STATION",
  643. {
  644. stationId
  645. },
  646. this
  647. )
  648. .then(() => {
  649. next(null);
  650. })
  651. .catch(next);
  652. }
  653. );
  654. }
  655. ],
  656. err => {
  657. if (err) reject(err);
  658. else resolve();
  659. }
  660. );
  661. });
  662. }
  663. /**
  664. * Gets next station song
  665. *
  666. * @param {object} payload - object that contains the payload
  667. * @param {string} payload.stationId - the id of the station
  668. * @returns {Promise} - returns a promise (resolve, reject)
  669. */
  670. GET_NEXT_STATION_SONG(payload) {
  671. return new Promise((resolve, reject) => {
  672. const { stationId } = payload;
  673. async.waterfall(
  674. [
  675. next => {
  676. StationsModule.runJob("GET_STATION", { stationId }, this)
  677. .then(station => {
  678. next(null, station);
  679. })
  680. .catch(next);
  681. },
  682. (station, next) => {
  683. if (station.queue.length === 0) next("No songs available.");
  684. else {
  685. next(null, station.queue[0]);
  686. }
  687. },
  688. (queueSong, next) => {
  689. MediaModule.runJob(
  690. "GET_MEDIA",
  691. {
  692. mediaSource: queueSong.mediaSource
  693. },
  694. this
  695. )
  696. .then(response => {
  697. const { song } = response;
  698. const {
  699. _id,
  700. mediaSource,
  701. title,
  702. skipDuration,
  703. artists,
  704. thumbnail,
  705. duration,
  706. verified
  707. } = song;
  708. next(null, {
  709. _id,
  710. mediaSource,
  711. title,
  712. skipDuration,
  713. artists,
  714. thumbnail,
  715. duration,
  716. verified,
  717. requestedAt: queueSong.requestedAt,
  718. requestedBy: queueSong.requestedBy,
  719. requestedType: queueSong.requestedType,
  720. likes: song.likes || 0,
  721. dislikes: song.dislikes || 0
  722. });
  723. })
  724. .catch(next);
  725. }
  726. ],
  727. (err, song) => {
  728. if (err) reject(err);
  729. else resolve({ song });
  730. }
  731. );
  732. });
  733. }
  734. /**
  735. * Removes first station queue song
  736. *
  737. * @param {object} payload - object that contains the payload
  738. * @param {string} payload.stationId - the id of the station
  739. * @returns {Promise} - returns a promise (resolve, reject)
  740. */
  741. REMOVE_FIRST_QUEUE_SONG(payload) {
  742. return new Promise((resolve, reject) => {
  743. const { stationId } = payload;
  744. async.waterfall(
  745. [
  746. next => {
  747. StationsModule.stationModel.updateOne(
  748. { _id: stationId },
  749. { $pop: { queue: -1 } },
  750. { runValidators: true },
  751. err => {
  752. if (err) next(err);
  753. else
  754. StationsModule.runJob(
  755. "UPDATE_STATION",
  756. {
  757. stationId
  758. },
  759. this
  760. )
  761. .then(() => {
  762. next(null);
  763. })
  764. .catch(next);
  765. }
  766. );
  767. }
  768. ],
  769. err => {
  770. if (err) reject(err);
  771. else resolve();
  772. }
  773. );
  774. });
  775. }
  776. /**
  777. * Process vote to skips for a station
  778. *
  779. * @param {object} payload - object that contains the payload
  780. * @param {string} payload.stationId - the id of the station to process
  781. * @returns {Promise} - returns a promise (resolve, reject)
  782. */
  783. PROCESS_SKIP_VOTES(payload) {
  784. return new Promise((resolve, reject) => {
  785. StationsModule.log("INFO", `Processing skip votes for station ${payload.stationId}.`);
  786. async.waterfall(
  787. [
  788. next => {
  789. StationsModule.runJob(
  790. "GET_STATION",
  791. {
  792. stationId: payload.stationId
  793. },
  794. this
  795. )
  796. .then(station => next(null, station))
  797. .catch(next);
  798. },
  799. (station, next) => {
  800. if (!station) return next("Station not found.");
  801. return next(null, station);
  802. },
  803. (station, next) => {
  804. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: `station.${station._id}` }, this)
  805. .then(sockets => next(null, station, sockets))
  806. .catch(next);
  807. },
  808. (station, sockets, next) => {
  809. const skipVotes =
  810. station.currentSong && station.currentSong.skipVotes
  811. ? station.currentSong.skipVotes.length
  812. : 0;
  813. let shouldSkip = false;
  814. if (skipVotes === 0) {
  815. if (!station.paused && !station.currentSong && station.queue.length > 0) shouldSkip = true;
  816. return next(null, shouldSkip);
  817. }
  818. const users = [];
  819. return async.each(
  820. sockets,
  821. (socketId, next) => {
  822. WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this)
  823. .then(socket => {
  824. if (socket && socket.session && socket.session.userId) {
  825. if (!users.includes(socket.session.userId))
  826. users.push(socket.session.userId);
  827. }
  828. return next();
  829. })
  830. .catch(next);
  831. },
  832. err => {
  833. if (err) return next(err);
  834. if (
  835. !station.paused &&
  836. Math.min(skipVotes, users.length) / users.length >= station.skipVoteThreshold / 100
  837. )
  838. shouldSkip = true;
  839. return next(null, shouldSkip);
  840. }
  841. );
  842. },
  843. (shouldSkip, next) => {
  844. if (shouldSkip)
  845. StationsModule.runJob(
  846. "SKIP_STATION",
  847. {
  848. stationId: payload.stationId,
  849. natural: false,
  850. skipReason: "vote_skip"
  851. },
  852. this
  853. )
  854. .then(() => next())
  855. .catch(next);
  856. else next();
  857. }
  858. ],
  859. err => {
  860. if (err) reject(err);
  861. else resolve();
  862. }
  863. );
  864. });
  865. }
  866. /**
  867. *
  868. *
  869. * @param {*} payload
  870. */
  871. async ADD_STATION_HISTORY_ITEM(payload) {
  872. const { stationId, currentSong, skipReason, skippedAt } = payload;
  873. let document = await StationsModule.stationHistoryModel.create({
  874. stationId,
  875. type: "song_played",
  876. payload: {
  877. song: currentSong,
  878. skippedAt,
  879. skipReason
  880. },
  881. documentVersion: 1
  882. });
  883. if (!document) return;
  884. document = document._doc;
  885. delete document.__v;
  886. delete document.documentVersion;
  887. WSModule.runJob("EMIT_TO_ROOM", {
  888. room: `station.${stationId}`,
  889. args: ["event:station.history.new", { data: { historyItem: document } }]
  890. });
  891. WSModule.runJob("EMIT_TO_ROOM", {
  892. room: `manage-station.${stationId}`,
  893. args: ["event:manageStation.history.new", { data: { stationId, historyItem: document } }]
  894. });
  895. }
  896. /**
  897. * Skips a station
  898. *
  899. * @param {object} payload - object that contains the payload
  900. * @param {string} payload.stationId - the id of the station to skip
  901. * @param {string} payload.natural - whether to skip naturally or forcefully
  902. * @param {string} payload.skipReason - if it was skipped via force skip or via vote skipping or if it was natural
  903. * @returns {Promise} - returns a promise (resolve, reject)
  904. */
  905. SKIP_STATION(payload) {
  906. return new Promise((resolve, reject) => {
  907. StationsModule.log("INFO", `Skipping station ${payload.stationId}.`);
  908. StationsModule.log("STATION_ISSUE", `SKIP_STATION_CB - Station ID: ${payload.stationId}.`);
  909. async.waterfall(
  910. [
  911. // Clears up any existing timers that would skip the station if the song ends
  912. next => {
  913. NotificationsModule.runJob("UNSCHEDULE", {
  914. name: `stations.nextSong?id=${payload.stationId}`
  915. })
  916. .then(() => {
  917. next();
  918. })
  919. .catch(next);
  920. },
  921. // Gets the station object
  922. next => {
  923. StationsModule.runJob(
  924. "GET_STATION",
  925. {
  926. stationId: payload.stationId
  927. },
  928. this
  929. )
  930. .then(station => next(null, station))
  931. .catch(next);
  932. },
  933. (station, next) => {
  934. if (!station) return next("Station not found.");
  935. const { currentSong } = station;
  936. if (!currentSong) return next(null, station);
  937. const stationId = station._id;
  938. const skippedAt = new Date();
  939. const { skipReason } = payload;
  940. return StationsModule.runJob(
  941. "ADD_STATION_HISTORY_ITEM",
  942. {
  943. stationId,
  944. currentSong,
  945. skippedAt,
  946. skipReason
  947. },
  948. this
  949. ).finally(() => {
  950. next(null, station);
  951. });
  952. },
  953. (station, next) => {
  954. if (station.autofill.enabled)
  955. return StationsModule.runJob("AUTOFILL_STATION", { stationId: station._id }, this)
  956. .then(() => next(null, station))
  957. .catch(err => {
  958. if (
  959. err === "Autofill is disabled in this station" ||
  960. err === "Autofill limit reached"
  961. )
  962. return next(null, station);
  963. return next(err);
  964. });
  965. return next(null, station);
  966. },
  967. (station, next) => {
  968. StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)
  969. .then(response => {
  970. StationsModule.runJob("REMOVE_FIRST_QUEUE_SONG", { stationId: station._id }, this)
  971. .then(() => {
  972. next(null, response.song, station);
  973. })
  974. .catch(next);
  975. })
  976. .catch(err => {
  977. if (err === "No songs available.") next(null, null, station);
  978. else next(err);
  979. });
  980. },
  981. async (song, station) => {
  982. const $set = {};
  983. if (song === null) $set.currentSong = null;
  984. else {
  985. // Block for experiment: queue_autofill_skip_last_x_played
  986. if (config.has(`experimental.queue_autofill_skip_last_x_played.${payload.stationId}`)) {
  987. const redisList = `experimental:queue_autofill_skip_last_x_played:${payload.stationId}`;
  988. const maxListLength = Number(
  989. config.get(`experimental.queue_autofill_skip_last_x_played.${payload.stationId}`)
  990. );
  991. // Add mediaSource to list for this station in Redis list
  992. await CacheModule.runJob(
  993. "LPUSH",
  994. {
  995. key: `experimental:queue_autofill_skip_last_x_played:${payload.stationId}`,
  996. value: song.mediaSource
  997. },
  998. this
  999. );
  1000. const currentListLength = await CacheModule.runJob("LLEN", { key: redisList }, this);
  1001. // Removes oldest mediaSource from list for this station in Redis list
  1002. if (currentListLength > maxListLength) {
  1003. const amount = currentListLength - maxListLength;
  1004. const promises = Array.from({ length: amount }).map(() =>
  1005. CacheModule.runJob(
  1006. "RPOP",
  1007. {
  1008. key: `experimental:queue_autofill_skip_last_x_played:${payload.stationId}`
  1009. },
  1010. this
  1011. )
  1012. );
  1013. await Promise.all(promises);
  1014. }
  1015. }
  1016. $set.currentSong = {
  1017. _id: song._id,
  1018. mediaSource: song.mediaSource,
  1019. title: song.title,
  1020. artists: song.artists,
  1021. duration: song.duration,
  1022. skipDuration: song.skipDuration,
  1023. thumbnail: song.thumbnail,
  1024. requestedAt: song.requestedAt,
  1025. requestedBy: song.requestedBy,
  1026. requestedType: song.requestedType,
  1027. verified: song.verified
  1028. };
  1029. }
  1030. $set.startedAt = Date.now();
  1031. $set.timePaused = 0;
  1032. if (station.paused) $set.pausedAt = Date.now();
  1033. return { $set, station };
  1034. },
  1035. ({ $set, station }, next) => {
  1036. StationsModule.stationModel.updateOne({ _id: station._id }, { $set }, err => {
  1037. if (err) return next(err);
  1038. return StationsModule.runJob("UPDATE_STATION", { stationId: station._id }, this)
  1039. .then(station => {
  1040. next(null, station);
  1041. })
  1042. .catch(next);
  1043. });
  1044. },
  1045. (station, next) => {
  1046. if (station.currentSong !== null && station.currentSong.mediaSource !== undefined) {
  1047. station.currentSong.skipVotes = 0;
  1048. }
  1049. next(null, station);
  1050. },
  1051. (station, next) => {
  1052. if (station.autofill.enabled)
  1053. return StationsModule.runJob("AUTOFILL_STATION", { stationId: station._id }, this)
  1054. .then(() => next(null, station))
  1055. .catch(err => {
  1056. if (
  1057. err === "Autofill is disabled in this station" ||
  1058. err === "Autofill limit reached"
  1059. )
  1060. return next(null, station);
  1061. return next(err);
  1062. });
  1063. return next(null, station);
  1064. },
  1065. (station, next) =>
  1066. StationsModule.runJob("UPDATE_STATION", { stationId: station._id }, this)
  1067. .then(station => {
  1068. CacheModule.runJob("PUB", {
  1069. channel: "station.queueUpdate",
  1070. value: payload.stationId
  1071. })
  1072. .then()
  1073. .catch();
  1074. next(null, station);
  1075. })
  1076. .catch(next)
  1077. ],
  1078. async (err, station) => {
  1079. if (err === "Autofill limit reached") return resolve({ station });
  1080. if (err) {
  1081. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1082. StationsModule.log("ERROR", `Skipping station "${payload.stationId}" failed. "${err}"`);
  1083. return reject(new Error(err));
  1084. }
  1085. // TODO Pub/Sub this
  1086. const { currentSong } = station;
  1087. WSModule.runJob("EMIT_TO_ROOM", {
  1088. room: `station.${station._id}`,
  1089. args: [
  1090. "event:station.nextSong",
  1091. {
  1092. data: {
  1093. currentSong,
  1094. startedAt: station.startedAt,
  1095. paused: station.paused,
  1096. timePaused: 0,
  1097. natural: payload.natural
  1098. }
  1099. }
  1100. ]
  1101. });
  1102. WSModule.runJob("EMIT_TO_ROOM", {
  1103. room: `manage-station.${station._id}`,
  1104. args: ["event:station.nextSong", { data: { stationId: station._id, currentSong } }]
  1105. });
  1106. if (station.privacy === "public")
  1107. WSModule.runJob("EMIT_TO_ROOM", {
  1108. room: "home",
  1109. args: ["event:station.nextSong", { data: { stationId: station._id, currentSong } }]
  1110. });
  1111. else {
  1112. const sockets = await WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: "home" }, this);
  1113. sockets.forEach(async socketId => {
  1114. const socket = await WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId });
  1115. if (!socket) return;
  1116. const { session } = socket;
  1117. if (session.sessionId) {
  1118. CacheModule.runJob("HGET", { table: "sessions", key: session.sessionId }).then(
  1119. session => {
  1120. const dispatchNextSong = () => {
  1121. socket.dispatch("event:station.nextSong", {
  1122. data: {
  1123. stationId: station._id,
  1124. currentSong
  1125. }
  1126. });
  1127. };
  1128. hasPermission("stations.index", session, station._id)
  1129. .then(() => dispatchNextSong())
  1130. .catch(() => {
  1131. hasPermission("stations.index.other", session)
  1132. .then(() => dispatchNextSong())
  1133. .catch(() => {});
  1134. });
  1135. }
  1136. );
  1137. }
  1138. });
  1139. }
  1140. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: `station.${station._id}` }).then(sockets => {
  1141. if (station.currentSong !== null && station.currentSong.mediaSource !== undefined) {
  1142. WSModule.runJob("SOCKETS_JOIN_SONG_ROOM", {
  1143. sockets,
  1144. room: `song.${station.currentSong.mediaSource}`
  1145. });
  1146. if (!station.paused) {
  1147. NotificationsModule.runJob("SCHEDULE", {
  1148. name: `stations.nextSong?id=${station._id}`,
  1149. time: station.currentSong.duration * 1000,
  1150. station
  1151. });
  1152. }
  1153. } else WSModule.runJob("SOCKETS_LEAVE_SONG_ROOMS", { sockets });
  1154. });
  1155. return resolve({ station });
  1156. }
  1157. );
  1158. });
  1159. }
  1160. /**
  1161. * Checks if a user can view/access a station
  1162. *
  1163. * @param {object} payload - object that contains the payload
  1164. * @param {object} payload.station - the station object of the station in question
  1165. * @param {string} payload.userId - the id of the user in question
  1166. * @returns {Promise} - returns a promise (resolve, reject)
  1167. */
  1168. CAN_USER_VIEW_STATION(payload) {
  1169. return new Promise((resolve, reject) => {
  1170. async.waterfall(
  1171. [
  1172. next => {
  1173. if (payload.station.privacy === "public" || payload.station.privacy === "unlisted")
  1174. return next(true);
  1175. if (!payload.userId) return next("Not allowed");
  1176. return next();
  1177. },
  1178. next => {
  1179. hasPermission("stations.view", payload.userId, payload.station._id)
  1180. .then(() => next(true))
  1181. .catch(() => next("Not allowed"));
  1182. }
  1183. ],
  1184. async errOrResult => {
  1185. if (errOrResult !== true && errOrResult !== "Not allowed") {
  1186. errOrResult = await UtilsModule.runJob(
  1187. "GET_ERROR",
  1188. {
  1189. error: errOrResult
  1190. },
  1191. this
  1192. );
  1193. reject(new Error(errOrResult));
  1194. } else {
  1195. resolve(errOrResult === true);
  1196. }
  1197. }
  1198. );
  1199. });
  1200. }
  1201. /**
  1202. * Checks if a user has favorited a station or not
  1203. *
  1204. * @param {object} payload - object that contains the payload
  1205. * @param {object} payload.stationId - the id 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. HAS_USER_FAVORITED_STATION(payload) {
  1210. return new Promise((resolve, reject) => {
  1211. async.waterfall(
  1212. [
  1213. next => {
  1214. DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(userModel => {
  1215. userModel.findOne({ _id: payload.userId }, next);
  1216. });
  1217. },
  1218. (user, next) => {
  1219. if (!user) return next("User not found.");
  1220. if (user.favoriteStations.indexOf(payload.stationId) !== -1) return next(null, true);
  1221. return next(null, false);
  1222. }
  1223. ],
  1224. async (err, isStationFavorited) => {
  1225. if (err && err !== true) {
  1226. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1227. return reject(new Error(err));
  1228. }
  1229. return resolve(isStationFavorited);
  1230. }
  1231. );
  1232. });
  1233. }
  1234. /**
  1235. * Returns a list of sockets in a room that can and can't know about a station
  1236. *
  1237. * @param {object} payload - the payload object
  1238. * @param {object} payload.station - the station object
  1239. * @param {string} payload.room - the websockets room to get the sockets from
  1240. * @returns {Promise} - returns a promise (resolve, reject)
  1241. */
  1242. GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION(payload) {
  1243. return new Promise((resolve, reject) => {
  1244. WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: payload.room }, this)
  1245. .then(socketIds => {
  1246. const sockets = [];
  1247. async.eachLimit(
  1248. socketIds,
  1249. 1,
  1250. (socketId, next) => {
  1251. WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this)
  1252. .then(socket => {
  1253. if (socket) sockets.push(socket);
  1254. next();
  1255. })
  1256. .catch(err => {
  1257. reject(err);
  1258. });
  1259. },
  1260. err => {
  1261. if (err) reject(err);
  1262. else {
  1263. let socketsThatCan = [];
  1264. const socketsThatCannot = [];
  1265. if (payload.station.privacy === "public") {
  1266. socketsThatCan = sockets;
  1267. resolve({ socketsThatCan, socketsThatCannot });
  1268. } else {
  1269. async.eachLimit(
  1270. sockets,
  1271. 1,
  1272. (socket, next) => {
  1273. if (!(socket.session && socket.session.sessionId)) {
  1274. socketsThatCannot.push(socket);
  1275. next();
  1276. } else
  1277. hasPermission("stations.view", socket.session, payload.station._id)
  1278. .then(() => {
  1279. socketsThatCan.push(socket);
  1280. next();
  1281. })
  1282. .catch(() => {
  1283. socketsThatCannot.push(socket);
  1284. next();
  1285. });
  1286. },
  1287. err => {
  1288. if (err) reject(err);
  1289. else resolve({ socketsThatCan, socketsThatCannot });
  1290. }
  1291. );
  1292. }
  1293. }
  1294. }
  1295. );
  1296. })
  1297. .catch(reject);
  1298. });
  1299. }
  1300. /**
  1301. * Adds a playlist to autofill a station
  1302. *
  1303. * @param {object} payload - object that contains the payload
  1304. * @param {object} payload.stationId - the id of the station
  1305. * @param {object} payload.playlistId - the id of the playlist
  1306. * @returns {Promise} - returns a promise (resolve, reject)
  1307. */
  1308. AUTOFILL_PLAYLIST(payload) {
  1309. return new Promise((resolve, reject) => {
  1310. async.waterfall(
  1311. [
  1312. next => {
  1313. if (!payload.stationId) next("Please specify a station id");
  1314. else if (!payload.playlistId) next("Please specify a playlist id");
  1315. else next();
  1316. },
  1317. next => {
  1318. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1319. .then(station => {
  1320. next(null, station);
  1321. })
  1322. .catch(next);
  1323. },
  1324. (station, next) => {
  1325. if (station.playlist === payload.playlistId) next("You cannot autofill the station playlist");
  1326. else if (station.autofill.playlists.indexOf(payload.playlistId) !== -1)
  1327. next("This playlist is already autofilling");
  1328. else if (station.blacklist.indexOf(payload.playlistId) !== -1)
  1329. next("This playlist is currently blacklisted");
  1330. else
  1331. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)
  1332. .then(() => {
  1333. next(null);
  1334. })
  1335. .catch(next);
  1336. },
  1337. next => {
  1338. StationsModule.stationModel.updateOne(
  1339. { _id: payload.stationId },
  1340. { $push: { "autofill.playlists": payload.playlistId } },
  1341. next
  1342. );
  1343. },
  1344. (res, next) => {
  1345. StationsModule.runJob(
  1346. "UPDATE_STATION",
  1347. {
  1348. stationId: payload.stationId
  1349. },
  1350. this
  1351. )
  1352. .then(() => {
  1353. next();
  1354. })
  1355. .catch(next);
  1356. }
  1357. ],
  1358. async err => {
  1359. if (err && err !== true) {
  1360. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1361. return reject(new Error(err));
  1362. }
  1363. return resolve();
  1364. }
  1365. );
  1366. });
  1367. }
  1368. /**
  1369. * Removes a playlist from autofill
  1370. *
  1371. * @param {object} payload - object that contains the payload
  1372. * @param {object} payload.stationId - the id of the station
  1373. * @param {object} payload.playlistId - the id of the playlist
  1374. * @returns {Promise} - returns a promise (resolve, reject)
  1375. */
  1376. REMOVE_AUTOFILL_PLAYLIST(payload) {
  1377. return new Promise((resolve, reject) => {
  1378. async.waterfall(
  1379. [
  1380. next => {
  1381. if (!payload.stationId) next("Please specify a station id");
  1382. else if (!payload.playlistId) next("Please specify a playlist id");
  1383. else next();
  1384. },
  1385. next => {
  1386. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1387. .then(station => {
  1388. next(null, station);
  1389. })
  1390. .catch(next);
  1391. },
  1392. (station, next) => {
  1393. if (station.autofill.playlists.indexOf(payload.playlistId) === -1)
  1394. next("This playlist isn't autofilling");
  1395. else next();
  1396. },
  1397. next => {
  1398. StationsModule.stationModel.updateOne(
  1399. { _id: payload.stationId },
  1400. { $pull: { "autofill.playlists": payload.playlistId } },
  1401. next
  1402. );
  1403. },
  1404. (res, next) => {
  1405. StationsModule.runJob(
  1406. "UPDATE_STATION",
  1407. {
  1408. stationId: payload.stationId
  1409. },
  1410. this
  1411. )
  1412. .then(() => {
  1413. next();
  1414. })
  1415. .catch(next);
  1416. }
  1417. ],
  1418. async err => {
  1419. if (err && err !== true) {
  1420. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1421. return reject(new Error(err));
  1422. }
  1423. return resolve();
  1424. }
  1425. );
  1426. });
  1427. }
  1428. /**
  1429. * Add a playlist to station blacklist
  1430. *
  1431. * @param {object} payload - object that contains the payload
  1432. * @param {object} payload.stationId - the id of the station
  1433. * @param {object} payload.playlistId - the id of the playlist
  1434. * @returns {Promise} - returns a promise (resolve, reject)
  1435. */
  1436. BLACKLIST_PLAYLIST(payload) {
  1437. return new Promise((resolve, reject) => {
  1438. async.waterfall(
  1439. [
  1440. next => {
  1441. if (!payload.stationId) next("Please specify a station id");
  1442. else if (!payload.playlistId) next("Please specify a playlist id");
  1443. else next();
  1444. },
  1445. next => {
  1446. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1447. .then(station => {
  1448. next(null, station);
  1449. })
  1450. .catch(next);
  1451. },
  1452. (station, next) => {
  1453. if (station.playlist === payload.playlistId) next("You cannot blacklist the station playlist");
  1454. else if (station.blacklist.indexOf(payload.playlistId) !== -1)
  1455. next("This playlist is already blacklisted");
  1456. else if (station.autofill.playlists.indexOf(payload.playlistId) !== -1)
  1457. next(
  1458. "This playlist is currently autofilling, please remove it from there before blacklisting it"
  1459. );
  1460. else
  1461. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)
  1462. .then(() => {
  1463. next(null);
  1464. })
  1465. .catch(next);
  1466. },
  1467. next => {
  1468. StationsModule.stationModel.updateOne(
  1469. { _id: payload.stationId },
  1470. { $push: { blacklist: payload.playlistId } },
  1471. next
  1472. );
  1473. },
  1474. (res, next) => {
  1475. StationsModule.runJob(
  1476. "UPDATE_STATION",
  1477. {
  1478. stationId: payload.stationId
  1479. },
  1480. this
  1481. )
  1482. .then(() => {
  1483. next();
  1484. })
  1485. .catch(next);
  1486. }
  1487. ],
  1488. async err => {
  1489. if (err && err !== true) {
  1490. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1491. return reject(new Error(err));
  1492. }
  1493. return resolve();
  1494. }
  1495. );
  1496. });
  1497. }
  1498. /**
  1499. * Remove a playlist from station blacklist
  1500. *
  1501. * @param {object} payload - object that contains the payload
  1502. * @param {object} payload.stationId - the id of the station
  1503. * @param {object} payload.playlistId - the id of the playlist
  1504. * @returns {Promise} - returns a promise (resolve, reject)
  1505. */
  1506. REMOVE_BLACKLISTED_PLAYLIST(payload) {
  1507. return new Promise((resolve, reject) => {
  1508. async.waterfall(
  1509. [
  1510. next => {
  1511. if (!payload.stationId) next("Please specify a station id");
  1512. else if (!payload.playlistId) next("Please specify a playlist id");
  1513. else next();
  1514. },
  1515. next => {
  1516. StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)
  1517. .then(station => {
  1518. next(null, station);
  1519. })
  1520. .catch(next);
  1521. },
  1522. (station, next) => {
  1523. if (station.blacklist.indexOf(payload.playlistId) === -1)
  1524. next("This playlist isn't blacklisted");
  1525. else next();
  1526. },
  1527. next => {
  1528. StationsModule.stationModel.updateOne(
  1529. { _id: payload.stationId },
  1530. { $pull: { blacklist: payload.playlistId } },
  1531. next
  1532. );
  1533. },
  1534. (res, next) => {
  1535. StationsModule.runJob(
  1536. "UPDATE_STATION",
  1537. {
  1538. stationId: payload.stationId
  1539. },
  1540. this
  1541. )
  1542. .then(() => {
  1543. next();
  1544. })
  1545. .catch(next);
  1546. }
  1547. ],
  1548. async err => {
  1549. if (err && err !== true) {
  1550. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1551. return reject(new Error(err));
  1552. }
  1553. return resolve();
  1554. }
  1555. );
  1556. });
  1557. }
  1558. /**
  1559. * Removes autofilled or blacklisted playlist from a station
  1560. *
  1561. * @param {object} payload - object that contains the payload
  1562. * @param {string} payload.playlistId - the playlist id
  1563. * @returns {Promise} - returns promise (reject, resolve)
  1564. */
  1565. REMOVE_AUTOFILLED_OR_BLACKLISTED_PLAYLIST_FROM_STATIONS(payload) {
  1566. return new Promise((resolve, reject) => {
  1567. async.waterfall(
  1568. [
  1569. next => {
  1570. if (!payload.playlistId) next("Please specify a playlist id");
  1571. else next();
  1572. },
  1573. next => {
  1574. StationsModule.stationModel.updateMany(
  1575. {
  1576. $or: [{ "autofill.playlists": payload.playlistId }, { blacklist: payload.playlistId }]
  1577. },
  1578. {
  1579. $pull: {
  1580. "autofill.playlists": payload.playlistId,
  1581. blacklist: payload.playlistId
  1582. }
  1583. },
  1584. err => {
  1585. if (err) next(err);
  1586. else next();
  1587. }
  1588. );
  1589. }
  1590. ],
  1591. async err => {
  1592. if (err && err !== true) {
  1593. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1594. return reject(new Error(err));
  1595. }
  1596. return resolve();
  1597. }
  1598. );
  1599. });
  1600. }
  1601. /**
  1602. * Gets stations that autofill or blacklist a specific playlist
  1603. *
  1604. * @param {object} payload - object that contains the payload
  1605. * @param {string} payload.playlistId - the playlist id
  1606. * @returns {Promise} - returns promise (reject, resolve)
  1607. */
  1608. GET_STATIONS_THAT_AUTOFILL_OR_BLACKLIST_PLAYLIST(payload) {
  1609. return new Promise((resolve, reject) => {
  1610. DBModule.runJob(
  1611. "GET_MODEL",
  1612. {
  1613. modelName: "station"
  1614. },
  1615. this
  1616. ).then(stationModel => {
  1617. stationModel.find(
  1618. {
  1619. $or: [{ "autofill.playlists": payload.playlistId }, { blacklist: payload.playlistId }]
  1620. },
  1621. (err, stations) => {
  1622. if (err) reject(err);
  1623. else resolve({ stationIds: stations.map(station => station._id) });
  1624. }
  1625. );
  1626. });
  1627. });
  1628. }
  1629. /**
  1630. * Clears every queue
  1631. *
  1632. * @returns {Promise} - returns a promise (resolve, reject)
  1633. */
  1634. CLEAR_EVERY_STATION_QUEUE() {
  1635. return new Promise((resolve, reject) => {
  1636. async.waterfall(
  1637. [
  1638. next => {
  1639. StationsModule.stationModel.updateMany({}, { $set: { queue: [] } }, err => {
  1640. if (err) next(err);
  1641. else {
  1642. StationsModule.stationModel.find({}, (err, stations) => {
  1643. if (err) next(err);
  1644. else {
  1645. async.eachLimit(
  1646. stations,
  1647. 1,
  1648. (station, next) => {
  1649. this.publishProgress({
  1650. status: "update",
  1651. message: `Updating station "${station._id}"`
  1652. });
  1653. StationsModule.runJob("UPDATE_STATION", {
  1654. stationId: station._id
  1655. })
  1656. .then(() => next())
  1657. .catch(next);
  1658. CacheModule.runJob("PUB", {
  1659. channel: "station.queueUpdate",
  1660. value: station._id
  1661. })
  1662. .then()
  1663. .catch();
  1664. },
  1665. next
  1666. );
  1667. }
  1668. });
  1669. }
  1670. });
  1671. }
  1672. ],
  1673. err => {
  1674. if (err) reject(err);
  1675. else resolve();
  1676. }
  1677. );
  1678. });
  1679. }
  1680. /**
  1681. * Resets a station queue
  1682. *
  1683. * @param {object} payload - object that contains the payload
  1684. * @param {string} payload.stationId - the station id
  1685. * @returns {Promise} - returns a promise (resolve, reject)
  1686. */
  1687. RESET_QUEUE(payload) {
  1688. return new Promise((resolve, reject) => {
  1689. async.waterfall(
  1690. [
  1691. next => {
  1692. StationsModule.runJob(
  1693. "AUTOFILL_STATION",
  1694. { stationId: payload.stationId, ignoreExistingQueue: true },
  1695. this
  1696. )
  1697. .then(() => {
  1698. CacheModule.runJob("PUB", {
  1699. channel: "station.queueUpdate",
  1700. value: payload.stationId
  1701. })
  1702. .then()
  1703. .catch();
  1704. next();
  1705. })
  1706. .catch(err => {
  1707. if (err === "Autofill is disabled in this station" || err === "Autofill limit reached")
  1708. StationsModule.stationModel
  1709. .updateOne({ _id: payload.stationId }, { $set: { queue: [] } }, this)
  1710. .then(() => next())
  1711. .catch(next);
  1712. else next(err);
  1713. });
  1714. },
  1715. next => {
  1716. StationsModule.runJob("UPDATE_STATION", { stationId: payload.stationId }, this)
  1717. .then(() => next())
  1718. .catch(next);
  1719. },
  1720. next => {
  1721. CacheModule.runJob("PUB", {
  1722. channel: "station.queueUpdate",
  1723. value: payload.stationId
  1724. })
  1725. .then(() => next())
  1726. .catch(next);
  1727. }
  1728. ],
  1729. err => {
  1730. if (err) reject(err);
  1731. else resolve();
  1732. }
  1733. );
  1734. });
  1735. }
  1736. /**
  1737. * Add to a station queue
  1738. *
  1739. * @param {object} payload - object that contains the payload
  1740. * @param {string} payload.stationId - the station id
  1741. * @param {string} payload.mediaSource - the media source
  1742. * @param {string} payload.requestUser - the requesting user id
  1743. * @param {string} payload.requestType - the request type
  1744. * @returns {Promise} - returns a promise (resolve, reject)
  1745. */
  1746. ADD_TO_QUEUE(payload) {
  1747. return new Promise((resolve, reject) => {
  1748. const { stationId, mediaSource, requestUser, requestType } = payload;
  1749. console.log(45436546, mediaSource);
  1750. async.waterfall(
  1751. [
  1752. next => {
  1753. StationsModule.runJob("GET_STATION", { stationId }, this)
  1754. .then(station => {
  1755. next(null, station);
  1756. })
  1757. .catch(next);
  1758. },
  1759. (station, next) => {
  1760. if (!station) return next("Station not found.");
  1761. if (!station.requests.enabled) return next("Requests are disabled in this station.");
  1762. if (station.currentSong && station.currentSong.mediaSource === mediaSource)
  1763. return next("That song is currently playing.");
  1764. if (station.queue.find(song => song.mediaSource === mediaSource))
  1765. return next("That song is already in the queue.");
  1766. return next(null, station);
  1767. },
  1768. (station, next) => {
  1769. MediaModule.runJob("GET_MEDIA", { mediaSource }, this)
  1770. .then(response => {
  1771. const { song } = response;
  1772. const { _id, title, skipDuration, artists, thumbnail, duration, verified } = song;
  1773. next(
  1774. null,
  1775. {
  1776. _id,
  1777. mediaSource,
  1778. title,
  1779. skipDuration,
  1780. artists,
  1781. thumbnail,
  1782. duration,
  1783. verified
  1784. },
  1785. station
  1786. );
  1787. })
  1788. .catch(next);
  1789. },
  1790. (song, station, next) => {
  1791. const blacklist = [];
  1792. async.eachLimit(
  1793. station.blacklist,
  1794. 1,
  1795. (playlistId, next) => {
  1796. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }, this)
  1797. .then(playlist => {
  1798. blacklist.push(playlist);
  1799. next();
  1800. })
  1801. .catch(next);
  1802. },
  1803. err => {
  1804. next(err, song, station, blacklist);
  1805. }
  1806. );
  1807. },
  1808. (song, station, blacklist, next) => {
  1809. const blacklistedSongs = blacklist
  1810. .flatMap(blacklistedPlaylist => blacklistedPlaylist.songs)
  1811. .reduce(
  1812. (items, item) =>
  1813. items.find(x => x.mediaSource === item.mediaSource) ? [...items] : [...items, item],
  1814. []
  1815. );
  1816. if (blacklistedSongs.find(blacklistedSong => blacklistedSong.mediaSource === song.mediaSource))
  1817. next("That song is in an blacklisted playlist and cannot be played.");
  1818. else next(null, song, station);
  1819. },
  1820. (song, station, next) => {
  1821. song.requestedBy = requestUser;
  1822. song.requestedAt = Date.now();
  1823. song.requestedType = requestType;
  1824. if (station.queue.length === 0) return next(null, song, station);
  1825. if (
  1826. requestUser &&
  1827. station.queue.filter(queueSong => queueSong.requestedBy === song.requestedBy).length >=
  1828. station.requests.limit
  1829. )
  1830. return next(`The max amount of songs per user is ${station.requests.limit}.`);
  1831. return next(null, song, station);
  1832. },
  1833. // (song, station, next) => {
  1834. // song.requestedBy = session.userId;
  1835. // song.requestedAt = Date.now();
  1836. // let totalDuration = 0;
  1837. // station.queue.forEach(song => {
  1838. // totalDuration += song.duration;
  1839. // });
  1840. // if (totalDuration >= 3600 * 3) return next("The max length of the queue is 3 hours.");
  1841. // return next(null, song, station);
  1842. // },
  1843. // (song, station, next) => {
  1844. // if (station.queue.length === 0) return next(null, song, station);
  1845. // let totalDuration = 0;
  1846. // const userId = station.queue[station.queue.length - 1].requestedBy;
  1847. // station.queue.forEach(song => {
  1848. // if (userId === song.requestedBy) {
  1849. // totalDuration += song.duration;
  1850. // }
  1851. // });
  1852. // if (totalDuration >= 900) return next("The max length of songs per user is 15 minutes.");
  1853. // return next(null, song, station);
  1854. // },
  1855. // (song, station, next) => {
  1856. // if (station.queue.length === 0) return next(null, song);
  1857. // let totalSongs = 0;
  1858. // const userId = station.queue[station.queue.length - 1].requestedBy;
  1859. // station.queue.forEach(song => {
  1860. // if (userId === song.requestedBy) {
  1861. // totalSongs += 1;
  1862. // }
  1863. // });
  1864. // if (totalSongs <= 2) return next(null, song);
  1865. // if (totalSongs > 3)
  1866. // return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1867. // if (
  1868. // station.queue[station.queue.length - 2].requestedBy !== userId ||
  1869. // station.queue[station.queue.length - 3] !== userId
  1870. // )
  1871. // return next("The max amount of songs per user is 3, and only 2 in a row is allowed.");
  1872. // return next(null, song);
  1873. // },
  1874. (song, station, next) => {
  1875. if (config.has(`experimental.queue_add_before_autofilled`)) {
  1876. const queueAddBeforeAutofilled = config.get(`experimental.queue_add_before_autofilled`);
  1877. if (
  1878. queueAddBeforeAutofilled === true ||
  1879. (Array.isArray(queueAddBeforeAutofilled) &&
  1880. queueAddBeforeAutofilled.indexOf(stationId) !== -1)
  1881. ) {
  1882. let position = station.queue.length;
  1883. if (station.autofill.enabled && station.queue.length >= station.autofill.limit) {
  1884. position = -station.autofill.limit;
  1885. }
  1886. StationsModule.stationModel.updateOne(
  1887. { _id: stationId },
  1888. {
  1889. $push: {
  1890. queue: {
  1891. $each: [song],
  1892. $position: position
  1893. }
  1894. }
  1895. },
  1896. { runValidators: true },
  1897. next
  1898. );
  1899. return;
  1900. }
  1901. }
  1902. StationsModule.stationModel.updateOne(
  1903. { _id: stationId },
  1904. { $push: { queue: song } },
  1905. { runValidators: true },
  1906. next
  1907. );
  1908. },
  1909. (res, next) => {
  1910. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1911. .then(() => next())
  1912. .catch(next);
  1913. },
  1914. next => {
  1915. CacheModule.runJob(
  1916. "PUB",
  1917. {
  1918. channel: "station.queueUpdate",
  1919. value: stationId
  1920. },
  1921. this
  1922. )
  1923. .then(() => next())
  1924. .catch(next);
  1925. }
  1926. ],
  1927. err => {
  1928. if (err) reject(err);
  1929. else resolve();
  1930. }
  1931. );
  1932. });
  1933. }
  1934. /**
  1935. * Remove from a station queue
  1936. *
  1937. * @param {object} payload - object that contains the payload
  1938. * @param {string} payload.stationId - the station id
  1939. * @param {string} payload.mediaSource - the media source
  1940. * @returns {Promise} - returns a promise (resolve, reject)
  1941. */
  1942. REMOVE_FROM_QUEUE(payload) {
  1943. return new Promise((resolve, reject) => {
  1944. const { stationId, mediaSource } = payload;
  1945. async.waterfall(
  1946. [
  1947. next => {
  1948. StationsModule.runJob("GET_STATION", { stationId }, this)
  1949. .then(station => {
  1950. next(null, station);
  1951. })
  1952. .catch(next);
  1953. },
  1954. (station, next) => {
  1955. if (!station) return next("Station not found.");
  1956. if (!station.queue.find(song => song.mediaSource === mediaSource))
  1957. return next("That song is not currently in the queue.");
  1958. return StationsModule.stationModel.updateOne(
  1959. { _id: stationId },
  1960. { $pull: { queue: { mediaSource } } },
  1961. next
  1962. );
  1963. },
  1964. (res, next) => {
  1965. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1966. .then(station => {
  1967. if (station.autofill.enabled)
  1968. StationsModule.runJob("AUTOFILL_STATION", { stationId }, this)
  1969. .then(() => next())
  1970. .catch(err => {
  1971. if (
  1972. err === "Autofill is disabled in this station" ||
  1973. err === "Autofill limit reached"
  1974. )
  1975. return next();
  1976. return next(err);
  1977. });
  1978. else next();
  1979. })
  1980. .catch(next);
  1981. },
  1982. next =>
  1983. CacheModule.runJob(
  1984. "PUB",
  1985. {
  1986. channel: "station.queueUpdate",
  1987. value: stationId
  1988. },
  1989. this
  1990. )
  1991. .then(() => next())
  1992. .catch(next)
  1993. ],
  1994. err => {
  1995. if (err) reject(err);
  1996. else resolve();
  1997. }
  1998. );
  1999. });
  2000. }
  2001. /**
  2002. * Add DJ to station
  2003. *
  2004. * @param {object} payload - object that contains the payload
  2005. * @param {string} payload.stationId - the station id
  2006. * @param {string} payload.userId - the dj user id
  2007. * @returns {Promise} - returns a promise (resolve, reject)
  2008. */
  2009. ADD_DJ(payload) {
  2010. return new Promise((resolve, reject) => {
  2011. const { stationId, userId } = payload;
  2012. async.waterfall(
  2013. [
  2014. next => {
  2015. StationsModule.runJob("GET_STATION", { stationId }, this)
  2016. .then(station => {
  2017. next(null, station);
  2018. })
  2019. .catch(next);
  2020. },
  2021. (station, next) => {
  2022. if (!station) return next("Station not found.");
  2023. if (station.djs.find(dj => dj === userId)) return next("That user is already a DJ.");
  2024. return StationsModule.stationModel.updateOne(
  2025. { _id: stationId },
  2026. { $push: { djs: userId } },
  2027. next
  2028. );
  2029. },
  2030. (res, next) => {
  2031. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  2032. .then(() => next())
  2033. .catch(next);
  2034. },
  2035. next =>
  2036. CacheModule.runJob(
  2037. "PUB",
  2038. {
  2039. channel: "station.djs.added",
  2040. value: { stationId, userId }
  2041. },
  2042. this
  2043. )
  2044. .then(() => next())
  2045. .catch(next)
  2046. ],
  2047. err => {
  2048. if (err) reject(err);
  2049. else resolve();
  2050. }
  2051. );
  2052. });
  2053. }
  2054. /**
  2055. * Remove DJ from station
  2056. *
  2057. * @param {object} payload - object that contains the payload
  2058. * @param {string} payload.stationId - the station id
  2059. * @param {string} payload.userId - the dj user id
  2060. * @returns {Promise} - returns a promise (resolve, reject)
  2061. */
  2062. REMOVE_DJ(payload) {
  2063. return new Promise((resolve, reject) => {
  2064. const { stationId, userId } = payload;
  2065. async.waterfall(
  2066. [
  2067. next => {
  2068. StationsModule.runJob("GET_STATION", { stationId }, this)
  2069. .then(station => {
  2070. next(null, station);
  2071. })
  2072. .catch(next);
  2073. },
  2074. (station, next) => {
  2075. if (!station) return next("Station not found.");
  2076. if (!station.djs.find(dj => dj === userId)) return next("That user is not currently a DJ.");
  2077. return StationsModule.stationModel.updateOne(
  2078. { _id: stationId },
  2079. { $pull: { djs: userId } },
  2080. next
  2081. );
  2082. },
  2083. (res, next) => {
  2084. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  2085. .then(() => next())
  2086. .catch(next);
  2087. },
  2088. next =>
  2089. CacheModule.runJob(
  2090. "PUB",
  2091. {
  2092. channel: "station.djs.removed",
  2093. value: { stationId, userId }
  2094. },
  2095. this
  2096. )
  2097. .then(() => next())
  2098. .catch(next)
  2099. ],
  2100. err => {
  2101. if (err) reject(err);
  2102. else resolve();
  2103. }
  2104. );
  2105. });
  2106. }
  2107. }
  2108. export default new _StationsModule();