stations.js 58 KB

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