stations.js 62 KB

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