stations.js 76 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834
  1. import async from "async";
  2. import mongoose from "mongoose";
  3. import config from "config";
  4. import { hasPermission, useHasPermission } from "../hooks/hasPermission";
  5. import isLoginRequired from "../hooks/loginRequired";
  6. import isLoginSometimesRequired from "../hooks/loginSometimesRequired";
  7. // eslint-disable-next-line
  8. import moduleManager from "../../index";
  9. const DBModule = moduleManager.modules.db;
  10. const UtilsModule = moduleManager.modules.utils;
  11. const WSModule = moduleManager.modules.ws;
  12. const PlaylistsModule = moduleManager.modules.playlists;
  13. const CacheModule = moduleManager.modules.cache;
  14. const NotificationsModule = moduleManager.modules.notifications;
  15. const StationsModule = moduleManager.modules.stations;
  16. const ActivitiesModule = moduleManager.modules.activities;
  17. CacheModule.runJob("SUB", {
  18. channel: "station.updateUsers",
  19. cb: ({ stationId, usersPerStation }) => {
  20. WSModule.runJob("EMIT_TO_ROOM", {
  21. room: `station.${stationId}`,
  22. args: ["event:station.users.updated", { data: { users: usersPerStation } }]
  23. });
  24. }
  25. });
  26. CacheModule.runJob("SUB", {
  27. channel: "station.updateUserCount",
  28. cb: ({ stationId, usersPerStationCount }) => {
  29. const count = usersPerStationCount || 0;
  30. WSModule.runJob("EMIT_TO_ROOM", {
  31. room: `station.${stationId}`,
  32. args: ["event:station.userCount.updated", { data: { userCount: count } }]
  33. });
  34. StationsModule.runJob("GET_STATION", { stationId }).then(async station => {
  35. if (station.privacy === "public")
  36. WSModule.runJob("EMIT_TO_ROOM", {
  37. room: "home",
  38. args: ["event:station.userCount.updated", { data: { stationId, userCount: count } }]
  39. });
  40. else {
  41. const sockets = await WSModule.runJob("GET_SOCKETS_FOR_ROOM", {
  42. room: "home"
  43. });
  44. sockets.forEach(async socketId => {
  45. const socket = await WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this);
  46. if (!socket) return;
  47. const { session } = socket;
  48. hasPermission("stations.view", session, stationId)
  49. .then(() => {
  50. socket.dispatch("event:station.userCount.updated", {
  51. data: { stationId, count }
  52. });
  53. })
  54. .catch(() => {});
  55. });
  56. }
  57. });
  58. }
  59. });
  60. CacheModule.runJob("SUB", {
  61. channel: "station.autofillPlaylist",
  62. cb: data => {
  63. const { stationId, playlistId } = data;
  64. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }).then(playlist =>
  65. WSModule.runJob("EMIT_TO_ROOMS", {
  66. rooms: [`station.${stationId}`, `manage-station.${stationId}`],
  67. args: ["event:station.autofillPlaylist", { data: { stationId, playlist } }]
  68. })
  69. );
  70. }
  71. });
  72. CacheModule.runJob("SUB", {
  73. channel: "station.blacklistedPlaylist",
  74. cb: data => {
  75. const { stationId, playlistId } = data;
  76. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }).then(playlist =>
  77. WSModule.runJob("EMIT_TO_ROOMS", {
  78. rooms: [`station.${stationId}`, `manage-station.${stationId}`],
  79. args: ["event:station.blacklistedPlaylist", { data: { stationId, playlist } }]
  80. })
  81. );
  82. }
  83. });
  84. CacheModule.runJob("SUB", {
  85. channel: "station.removedAutofillPlaylist",
  86. cb: data => {
  87. const { stationId, playlistId } = data;
  88. WSModule.runJob("EMIT_TO_ROOMS", {
  89. rooms: [`station.${stationId}`, `manage-station.${stationId}`],
  90. args: ["event:station.removedAutofillPlaylist", { data: { stationId, playlistId } }]
  91. });
  92. }
  93. });
  94. CacheModule.runJob("SUB", {
  95. channel: "station.removedBlacklistedPlaylist",
  96. cb: data => {
  97. const { stationId, playlistId } = data;
  98. WSModule.runJob("EMIT_TO_ROOMS", {
  99. rooms: [`station.${stationId}`, `manage-station.${stationId}`],
  100. args: ["event:station.removedBlacklistedPlaylist", { data: { stationId, playlistId } }]
  101. });
  102. }
  103. });
  104. CacheModule.runJob("SUB", {
  105. channel: "station.pause",
  106. cb: stationId => {
  107. StationsModule.runJob("GET_STATION", { stationId }).then(station => {
  108. WSModule.runJob("EMIT_TO_ROOM", {
  109. room: `station.${stationId}`,
  110. args: ["event:station.pause", { data: { pausedAt: station.pausedAt } }]
  111. });
  112. WSModule.runJob("EMIT_TO_ROOM", {
  113. room: `manage-station.${stationId}`,
  114. args: ["event:station.pause", { data: { stationId, pausedAt: station.pausedAt } }]
  115. });
  116. StationsModule.runJob("GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION", {
  117. room: `home`,
  118. station
  119. }).then(response => {
  120. const { socketsThatCan } = response;
  121. socketsThatCan.forEach(socket => {
  122. socket.dispatch("event:station.pause", { data: { stationId } });
  123. });
  124. });
  125. });
  126. }
  127. });
  128. CacheModule.runJob("SUB", {
  129. channel: "station.resume",
  130. cb: stationId => {
  131. StationsModule.runJob("GET_STATION", { stationId }).then(station => {
  132. WSModule.runJob("EMIT_TO_ROOM", {
  133. room: `station.${stationId}`,
  134. args: ["event:station.resume", { data: { timePaused: station.timePaused } }]
  135. });
  136. WSModule.runJob("EMIT_TO_ROOM", {
  137. room: `manage-station.${stationId}`,
  138. args: ["event:station.resume", { data: { stationId, timePaused: station.timePaused } }]
  139. });
  140. StationsModule.runJob("GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION", {
  141. room: `home`,
  142. station
  143. })
  144. .then(response => {
  145. const { socketsThatCan } = response;
  146. socketsThatCan.forEach(socket => {
  147. socket.dispatch("event:station.resume", { data: { stationId } });
  148. });
  149. })
  150. .catch(console.log);
  151. });
  152. }
  153. });
  154. CacheModule.runJob("SUB", {
  155. channel: "station.queueUpdate",
  156. cb: stationId => {
  157. StationsModule.runJob("GET_STATION", { stationId }).then(station => {
  158. if (!station.currentSong && station.queue.length > 0) {
  159. StationsModule.runJob("INITIALIZE_STATION", {
  160. stationId
  161. }).then();
  162. }
  163. WSModule.runJob("EMIT_TO_ROOM", {
  164. room: `station.${stationId}`,
  165. args: ["event:station.queue.updated", { data: { queue: station.queue } }]
  166. });
  167. WSModule.runJob("EMIT_TO_ROOM", {
  168. room: `manage-station.${stationId}`,
  169. args: ["event:manageStation.queue.updated", { data: { stationId, queue: station.queue } }]
  170. });
  171. });
  172. }
  173. });
  174. CacheModule.runJob("SUB", {
  175. channel: "station.changeQueueOrder",
  176. cb: res => {
  177. WSModule.runJob("EMIT_TO_ROOM", {
  178. room: `station.${res.stationId}`,
  179. args: ["event:station.queue.order.changed", { data: { queueOrder: res.queueOrder } }]
  180. });
  181. WSModule.runJob("EMIT_TO_ROOM", {
  182. room: `manage-station.${res.stationId}`,
  183. args: [
  184. "event:manageStation.queue.order.changed",
  185. { data: { stationId: res.stationId, queueOrder: res.queueOrder } }
  186. ]
  187. });
  188. }
  189. });
  190. CacheModule.runJob("SUB", {
  191. channel: "station.toggleSkipVote",
  192. cb: res => {
  193. const { stationId, voted, userId } = res;
  194. WSModule.runJob("EMIT_TO_ROOM", {
  195. room: `station.${stationId}`,
  196. args: ["event:station.toggleSkipVote", { data: { voted, userId } }]
  197. });
  198. }
  199. });
  200. CacheModule.runJob("SUB", {
  201. channel: "station.remove",
  202. cb: stationId => {
  203. WSModule.runJob("EMIT_TO_ROOM", {
  204. room: `station.${stationId}`,
  205. args: ["event:station.deleted"]
  206. });
  207. WSModule.runJob("EMIT_TO_ROOM", {
  208. room: `manage-station.${stationId}`,
  209. args: ["event:station.deleted"]
  210. });
  211. WSModule.runJob("EMIT_TO_ROOM", {
  212. room: `home`,
  213. args: ["event:station.deleted", { data: { stationId } }]
  214. });
  215. WSModule.runJob("EMIT_TO_ROOM", {
  216. room: "admin.stations",
  217. args: ["event:admin.station.deleted", { data: { stationId } }]
  218. });
  219. }
  220. });
  221. CacheModule.runJob("SUB", {
  222. channel: "station.create",
  223. cb: async stationId => {
  224. StationsModule.runJob("INITIALIZE_STATION", { stationId }).then(async res => {
  225. const { station } = res;
  226. station.userCount = StationsModule.usersPerStationCount[stationId] || 0;
  227. WSModule.runJob("EMIT_TO_ROOM", {
  228. room: "admin.stations",
  229. args: ["event:admin.station.created", { data: { station } }]
  230. });
  231. if (station.privacy === "public")
  232. WSModule.runJob("EMIT_TO_ROOM", {
  233. room: "home",
  234. args: ["event:station.created", { data: { station } }]
  235. });
  236. else {
  237. const sockets = await WSModule.runJob("GET_SOCKETS_FOR_ROOM", {
  238. room: "home"
  239. });
  240. sockets.forEach(async socketId => {
  241. const socket = await WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this);
  242. if (!socket) return;
  243. const { session } = socket;
  244. hasPermission("stations.view", session, stationId)
  245. .then(() => {
  246. socket.dispatch("event:station.created", { data: { station } });
  247. })
  248. .catch(() => {});
  249. });
  250. }
  251. });
  252. }
  253. });
  254. CacheModule.runJob("SUB", {
  255. channel: "station.updated",
  256. cb: async data => {
  257. const stationModel = await DBModule.runJob("GET_MODEL", {
  258. modelName: "station"
  259. });
  260. const { stationId } = data;
  261. stationModel.findOne(
  262. { _id: stationId },
  263. [
  264. "_id",
  265. "name",
  266. "displayName",
  267. "description",
  268. "type",
  269. "privacy",
  270. "owner",
  271. "requests",
  272. "autofill",
  273. "theme",
  274. "skipVoteThreshold"
  275. ],
  276. (err, station) => {
  277. WSModule.runJob("EMIT_TO_ROOMS", {
  278. rooms: [`station.${stationId}`, `manage-station.${stationId}`, "admin.stations"],
  279. args: ["event:station.updated", { data: { station } }]
  280. });
  281. StationsModule.runJob("GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION", {
  282. room: `home`,
  283. station
  284. }).then(response => {
  285. const { socketsThatCan } = response;
  286. socketsThatCan.forEach(socket => {
  287. socket.dispatch("event:station.updated", { data: { station } });
  288. });
  289. });
  290. if (data.previousStation && data.previousStation.privacy !== station.privacy) {
  291. if (station.privacy === "public") {
  292. // Station became public
  293. WSModule.runJob("EMIT_TO_ROOM", {
  294. room: "home",
  295. args: ["event:station.created", { data: { station } }]
  296. });
  297. } else if (data.previousStation.privacy === "public") {
  298. // Station became hidden
  299. StationsModule.runJob("GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION", {
  300. room: `home`,
  301. station
  302. }).then(response => {
  303. const { socketsThatCannot } = response;
  304. socketsThatCannot.forEach(socket => {
  305. socket.dispatch("event:station.deleted", { data: { stationId } });
  306. });
  307. });
  308. }
  309. }
  310. }
  311. );
  312. }
  313. });
  314. export default {
  315. /**
  316. * Get a list of all the stations
  317. * @param {object} session - user session
  318. * @param {boolean} adminFilter - whether to filter out stations admins do not own
  319. * @param {Function} cb - callback
  320. */
  321. index: isLoginSometimesRequired(async function index(session, adminFilter, cb) {
  322. const userModel = await DBModule.runJob("GET_MODEL", { modelName: "user" });
  323. async.waterfall(
  324. [
  325. // get array of the ids of the user's favorite stations
  326. next => {
  327. if (session.userId)
  328. return userModel.findById(session.userId).select({ favoriteStations: -1 }).exec(next);
  329. return next(null, { favoriteStations: [] });
  330. },
  331. (user, next) => {
  332. const favoriteStations = user ? user.favoriteStations : [];
  333. CacheModule.runJob("HGETALL", { table: "stations" }, this).then(stations =>
  334. next(null, stations, favoriteStations)
  335. );
  336. },
  337. (stations, favorited, next) => {
  338. const filteredStations = [];
  339. async.eachLimit(
  340. stations,
  341. 1,
  342. (station, nextStation) => {
  343. async.waterfall(
  344. [
  345. callback => {
  346. // only relevant if user logged in
  347. if (session.userId) {
  348. if (favorited.indexOf(station._id) !== -1) station.isFavorited = true;
  349. return callback();
  350. }
  351. return callback();
  352. },
  353. callback => {
  354. StationsModule.runJob(
  355. "CAN_USER_VIEW_STATION",
  356. {
  357. station,
  358. userId: session.userId
  359. },
  360. this
  361. )
  362. .then(exists => callback(null, exists))
  363. .catch(callback);
  364. },
  365. (exists, callback) => {
  366. if (!exists) callback(null, false, false);
  367. else if (station.privacy === "public") callback(null, true, true);
  368. else
  369. hasPermission("stations.index", session.userId, station._id)
  370. .then(() => callback(null, true, true))
  371. .catch(() => callback(null, true, false));
  372. },
  373. (exists, canIndex, callback) => {
  374. if (!exists) callback(null, false);
  375. else if (!canIndex && !adminFilter)
  376. hasPermission("stations.index.other", session.userId)
  377. .then(() => callback(null, true))
  378. .catch(() => callback(null, false));
  379. else callback(null, canIndex);
  380. }
  381. ],
  382. (err, exists) => {
  383. if (err) return this.log("ERROR", "STATIONS_INDEX", err);
  384. station.userCount = StationsModule.usersPerStationCount[station._id] || 0;
  385. if (exists) filteredStations.push(station);
  386. return nextStation();
  387. }
  388. );
  389. },
  390. () => next(null, filteredStations, favorited)
  391. );
  392. }
  393. ],
  394. async (err, stations, favorited) => {
  395. if (err) {
  396. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  397. this.log("ERROR", "STATIONS_INDEX", `Indexing stations failed. "${err}"`);
  398. return cb({ status: "error", message: err });
  399. }
  400. this.log("SUCCESS", "STATIONS_INDEX", `Indexing stations successful.`, false);
  401. return cb({ status: "success", data: { stations, favorited } });
  402. }
  403. );
  404. }),
  405. /**
  406. * Gets stations, used in the admin stations page by the AdvancedTable component
  407. * @param {object} session - the session object automatically added by the websocket
  408. * @param page - the page
  409. * @param pageSize - the size per page
  410. * @param properties - the properties to return for each station
  411. * @param sort - the sort object
  412. * @param queries - the queries array
  413. * @param operator - the operator for queries
  414. * @param {Function} cb - gets called with the result
  415. */
  416. getData: useHasPermission(
  417. "admin.view.stations",
  418. async function getSet(session, page, pageSize, properties, sort, queries, operator, cb) {
  419. async.waterfall(
  420. [
  421. next => {
  422. DBModule.runJob(
  423. "GET_DATA",
  424. {
  425. page,
  426. pageSize,
  427. properties,
  428. sort,
  429. queries,
  430. operator,
  431. modelName: "station",
  432. blacklistedProperties: [],
  433. specialProperties: {
  434. owner: [
  435. {
  436. $addFields: {
  437. ownerOID: {
  438. $convert: {
  439. input: "$owner",
  440. to: "objectId",
  441. onError: "unknown",
  442. onNull: "unknown"
  443. }
  444. }
  445. }
  446. },
  447. {
  448. $lookup: {
  449. from: "users",
  450. localField: "ownerOID",
  451. foreignField: "_id",
  452. as: "ownerUser"
  453. }
  454. },
  455. {
  456. $unwind: {
  457. path: "$ownerUser",
  458. preserveNullAndEmptyArrays: true
  459. }
  460. },
  461. {
  462. $addFields: {
  463. ownerUsername: {
  464. $cond: [
  465. { $eq: [{ $type: "$owner" }, "string"] },
  466. { $ifNull: ["$ownerUser.username", "unknown"] },
  467. "none"
  468. ]
  469. }
  470. }
  471. },
  472. {
  473. $project: {
  474. ownerOID: 0,
  475. ownerUser: 0
  476. }
  477. }
  478. ]
  479. },
  480. specialQueries: {
  481. owner: newQuery => ({ $or: [newQuery, { ownerUsername: newQuery.owner }] })
  482. }
  483. },
  484. this
  485. )
  486. .then(response => {
  487. next(null, response);
  488. })
  489. .catch(err => {
  490. next(err);
  491. });
  492. }
  493. ],
  494. async (err, response) => {
  495. if (err) {
  496. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  497. this.log("ERROR", "STATIONS_GET_DATA", `Failed to get data from stations. "${err}"`);
  498. return cb({ status: "error", message: err });
  499. }
  500. this.log("SUCCESS", "STATIONS_GET_DATA", `Got data from stations successfully.`);
  501. return cb({ status: "success", message: "Successfully got data from stations.", data: response });
  502. }
  503. );
  504. }
  505. ),
  506. /**
  507. * Obtains basic metadata of a station in order to format an activity
  508. * @param {object} session - user session
  509. * @param {string} stationId - the station id
  510. * @param {Function} cb - callback
  511. */
  512. getStationForActivity: isLoginSometimesRequired(function getStationForActivity(session, stationId, cb) {
  513. async.waterfall(
  514. [
  515. next => {
  516. StationsModule.runJob("GET_STATION", { stationId }, this)
  517. .then(station => {
  518. next(null, station);
  519. })
  520. .catch(next);
  521. }
  522. ],
  523. async (err, station) => {
  524. if (err) {
  525. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  526. this.log(
  527. "ERROR",
  528. "STATIONS_GET_STATION_FOR_ACTIVITY",
  529. `Failed to obtain metadata of station ${stationId} for activity formatting. "${err}"`
  530. );
  531. return cb({ status: "error", message: err });
  532. }
  533. this.log(
  534. "SUCCESS",
  535. "STATIONS_GET_STATION_FOR_ACTIVITY",
  536. `Obtained metadata of station ${stationId} for activity formatting successfully.`
  537. );
  538. return cb({
  539. status: "success",
  540. data: {
  541. title: station.displayName,
  542. thumbnail: station.currentSong ? station.currentSong.thumbnail : ""
  543. }
  544. });
  545. }
  546. );
  547. }),
  548. /**
  549. * Verifies that a station exists from its name
  550. * @param {object} session - user session
  551. * @param {string} stationName - the station name
  552. * @param {Function} cb - callback
  553. */
  554. existsByName: isLoginSometimesRequired(function existsByName(session, stationName, cb) {
  555. async.waterfall(
  556. [
  557. next => {
  558. StationsModule.runJob("GET_STATION_BY_NAME", { stationName }, this)
  559. .then(station => next(null, station))
  560. .catch(next);
  561. },
  562. (station, next) => {
  563. if (!station) return next(null, false);
  564. return StationsModule.runJob("CAN_USER_VIEW_STATION", { station, userId: session.userId }, this)
  565. .then(exists => next(null, exists))
  566. .catch(next);
  567. }
  568. ],
  569. async (err, exists) => {
  570. if (err) {
  571. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  572. this.log(
  573. "ERROR",
  574. "STATION_EXISTS_BY_NAME",
  575. `Checking if station "${stationName}" exists failed. "${err}"`
  576. );
  577. return cb({ status: "error", message: err });
  578. }
  579. this.log(
  580. "SUCCESS",
  581. "STATION_EXISTS_BY_NAME",
  582. `Station "${stationName}" exists successfully.` /* , false */
  583. );
  584. return cb({ status: "success", data: { exists } });
  585. }
  586. );
  587. }),
  588. /**
  589. * Verifies that a station exists from its id
  590. * @param {object} session - user session
  591. * @param {string} stationId - the station id
  592. * @param {Function} cb - callback
  593. */
  594. existsById: isLoginSometimesRequired(function existsById(session, stationId, cb) {
  595. async.waterfall(
  596. [
  597. next => {
  598. StationsModule.runJob("GET_STATION", { stationId }, this)
  599. .then(station => next(null, station))
  600. .catch(next);
  601. },
  602. (station, next) => {
  603. if (!station) return next(null, false);
  604. return StationsModule.runJob("CAN_USER_VIEW_STATION", { station, userId: session.userId }, this)
  605. .then(exists => next(null, exists))
  606. .catch(next);
  607. }
  608. ],
  609. async (err, exists) => {
  610. if (err) {
  611. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  612. this.log(
  613. "ERROR",
  614. "STATION_EXISTS_BY_ID",
  615. `Checking if station "${stationId}" exists failed. "${err}"`
  616. );
  617. return cb({ status: "error", message: err });
  618. }
  619. this.log(
  620. "SUCCESS",
  621. "STATION_EXISTS_BY_ID",
  622. `Station "${stationId}" exists successfully.` /* , false */
  623. );
  624. return cb({ status: "success", data: { exists } });
  625. }
  626. );
  627. }),
  628. /**
  629. * Gets the official playlist for a station
  630. * @param {object} session - user session
  631. * @param {string} stationId - the station id
  632. * @param {Function} cb - callback
  633. */
  634. getPlaylist: isLoginSometimesRequired(function getPlaylist(session, stationId, cb) {
  635. async.waterfall(
  636. [
  637. next => {
  638. StationsModule.runJob("GET_STATION", { stationId }, this)
  639. .then(station => {
  640. next(null, station);
  641. })
  642. .catch(next);
  643. },
  644. (station, next) => {
  645. StationsModule.runJob(
  646. "CAN_USER_VIEW_STATION",
  647. {
  648. station,
  649. userId: session.userId
  650. },
  651. this
  652. )
  653. .then(canView => {
  654. if (canView) return next(null, station);
  655. return next("Insufficient permissions.");
  656. })
  657. .catch(err => next(err));
  658. },
  659. (station, next) => {
  660. if (!station) return next("Station not found.");
  661. if (station.type !== "official") return next("This is not an official station.");
  662. return next();
  663. },
  664. next => {
  665. CacheModule.runJob(
  666. "HGET",
  667. {
  668. table: "officialPlaylists",
  669. key: stationId
  670. },
  671. this
  672. )
  673. .then(playlist => {
  674. next(null, playlist);
  675. })
  676. .catch(next);
  677. },
  678. (playlist, next) => {
  679. if (!playlist) return next("Playlist not found.");
  680. return next(null, playlist);
  681. }
  682. ],
  683. async (err, playlist) => {
  684. if (err) {
  685. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  686. this.log(
  687. "ERROR",
  688. "STATIONS_GET_PLAYLIST",
  689. `Getting playlist for station "${stationId}" failed. "${err}"`
  690. );
  691. return cb({ status: "error", message: err });
  692. }
  693. this.log(
  694. "SUCCESS",
  695. "STATIONS_GET_PLAYLIST",
  696. `Got playlist for station "${stationId}" successfully.`,
  697. false
  698. );
  699. return cb({ status: "success", data: { songs: playlist.songs } });
  700. }
  701. );
  702. }),
  703. /**
  704. * Joins the station by its name
  705. * @param {object} session - user session
  706. * @param {string} stationIdentifier - the station name or station id
  707. * @param {Function} cb - callback
  708. */
  709. join: isLoginSometimesRequired(async function join(session, stationIdentifier, cb) {
  710. const userModel = await DBModule.runJob("GET_MODEL", { modelName: "user" });
  711. async.waterfall(
  712. [
  713. next => {
  714. StationsModule.runJob("GET_STATION_BY_NAME", { stationName: stationIdentifier }, this)
  715. .then(station => next(null, station))
  716. .catch(() =>
  717. // station identifier may be using stationid instead
  718. StationsModule.runJob("GET_STATION", { stationId: stationIdentifier }, this)
  719. .then(station => next(null, station))
  720. .catch(next)
  721. );
  722. },
  723. (station, next) => {
  724. if (!station) return next("Station not found.");
  725. return StationsModule.runJob("CAN_USER_VIEW_STATION", { station, userId: session.userId }, this)
  726. .then(canView => {
  727. if (!canView) next("Not allowed to join station.");
  728. else next(null, station);
  729. })
  730. .catch(err => next(err));
  731. },
  732. (station, next) => {
  733. WSModule.runJob("SOCKET_JOIN_ROOM", {
  734. socketId: session.socketId,
  735. room: `station.${station._id}`
  736. });
  737. const data = {
  738. _id: station._id,
  739. type: station.type,
  740. currentSong: station.currentSong,
  741. startedAt: station.startedAt,
  742. paused: station.paused,
  743. timePaused: station.timePaused,
  744. pausedAt: station.pausedAt,
  745. description: station.description,
  746. displayName: station.displayName,
  747. name: station.name,
  748. privacy: station.privacy,
  749. requests: station.requests,
  750. autofill: station.autofill,
  751. owner: station.owner,
  752. blacklist: station.blacklist,
  753. theme: station.theme,
  754. djs: station.djs,
  755. skipVoteThreshold: station.skipVoteThreshold
  756. };
  757. StationsModule.userList[session.socketId] = station._id;
  758. next(null, data);
  759. },
  760. (data, next) => {
  761. userModel.find({ _id: { $in: data.djs } }, (err, users) => {
  762. if (err) next(err);
  763. else {
  764. data.djs = users.map(user => {
  765. const { _id, name, username, avatar } = user._doc;
  766. return { _id, name, username, avatar };
  767. });
  768. next(null, data);
  769. }
  770. });
  771. },
  772. (data, next) => {
  773. data = JSON.parse(JSON.stringify(data));
  774. data.userCount = StationsModule.usersPerStationCount[data._id] || 0;
  775. data.users = StationsModule.usersPerStation[data._id] || [];
  776. if (!data.currentSong || !data.currentSong.title) return next(null, data);
  777. WSModule.runJob("SOCKET_JOIN_SONG_ROOM", {
  778. socketId: session.socketId,
  779. room: `song.${data.currentSong.mediaSource}`
  780. });
  781. data.currentSong.skipVotes = data.currentSong.skipVotes.length;
  782. return next(null, data);
  783. },
  784. (data, next) => {
  785. // only relevant if user logged in
  786. if (session.userId) {
  787. return StationsModule.runJob(
  788. "HAS_USER_FAVORITED_STATION",
  789. {
  790. userId: session.userId,
  791. stationId: data._id
  792. },
  793. this
  794. )
  795. .then(isStationFavorited => {
  796. data.isFavorited = isStationFavorited;
  797. return next(null, data);
  798. })
  799. .catch(err => next(err));
  800. }
  801. return next(null, data);
  802. }
  803. ],
  804. async (err, data) => {
  805. if (err) {
  806. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  807. this.log("ERROR", "STATIONS_JOIN", `Joining station "${stationIdentifier}" failed. "${err}"`);
  808. return cb({ status: "error", message: err });
  809. }
  810. this.log("SUCCESS", "STATIONS_JOIN", `Joined station "${data._id}" successfully.`);
  811. return cb({ status: "success", data });
  812. }
  813. );
  814. }),
  815. /**
  816. * Gets a station by id
  817. * @param {object} session - user session
  818. * @param {string} stationId - the station id
  819. * @param {Function} cb - callback
  820. */
  821. getStationById: isLoginSometimesRequired(function getStationById(session, stationId, cb) {
  822. async.waterfall(
  823. [
  824. next => {
  825. StationsModule.runJob("GET_STATION", { stationId }, this)
  826. .then(station => {
  827. next(null, station);
  828. })
  829. .catch(next);
  830. },
  831. (station, next) => {
  832. if (!station) return next("Station not found.");
  833. return StationsModule.runJob(
  834. "CAN_USER_VIEW_STATION",
  835. {
  836. station,
  837. userId: session.userId
  838. },
  839. this
  840. )
  841. .then(canView => {
  842. if (!canView) next("Not allowed to get station.");
  843. else next(null, station);
  844. })
  845. .catch(err => next(err));
  846. },
  847. (station, next) => {
  848. // only relevant if user logged in
  849. if (session.userId) {
  850. return StationsModule.runJob(
  851. "HAS_USER_FAVORITED_STATION",
  852. {
  853. userId: session.userId,
  854. stationId
  855. },
  856. this
  857. )
  858. .then(isStationFavorited => {
  859. station.isFavorited = isStationFavorited;
  860. return next(null, station);
  861. })
  862. .catch(err => next(err));
  863. }
  864. return next(null, station);
  865. },
  866. (station, next) => {
  867. const data = {
  868. _id: station._id,
  869. type: station.type,
  870. description: station.description,
  871. displayName: station.displayName,
  872. name: station.name,
  873. privacy: station.privacy,
  874. requests: station.requests,
  875. autofill: station.autofill,
  876. owner: station.owner,
  877. theme: station.theme,
  878. paused: station.paused,
  879. currentSong: station.currentSong,
  880. isFavorited: station.isFavorited,
  881. djs: station.djs,
  882. skipVoteThreshold: station.skipVoteThreshold
  883. };
  884. next(null, data);
  885. }
  886. ],
  887. async (err, data) => {
  888. if (err) {
  889. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  890. this.log("ERROR", "GET_STATION_BY_ID", `Getting station "${stationId}" failed. "${err}"`);
  891. return cb({ status: "error", message: err });
  892. }
  893. this.log("SUCCESS", "GET_STATION_BY_ID", `Got station "${stationId}" successfully.`);
  894. return cb({ status: "success", data: { station: data } });
  895. }
  896. );
  897. }),
  898. /**
  899. * Gets station history
  900. * @param {object} session - user session
  901. * @param {string} stationId - the station id
  902. * @param {Function} cb - callback
  903. */
  904. getHistory: isLoginSometimesRequired(function getHistory(session, stationId, cb) {
  905. async.waterfall(
  906. [
  907. next => {
  908. if (!config.get("experimental.station_history")) return next("Station history is not enabled");
  909. return next();
  910. },
  911. next => {
  912. StationsModule.runJob("GET_STATION", { stationId }, this)
  913. .then(station => {
  914. next(null, station);
  915. })
  916. .catch(next);
  917. },
  918. (station, next) => {
  919. if (!station) return next("Station not found.");
  920. return StationsModule.runJob(
  921. "CAN_USER_VIEW_STATION",
  922. {
  923. station,
  924. userId: session.userId
  925. },
  926. this
  927. )
  928. .then(canView => {
  929. if (!canView) next("Not allowed to access station history.");
  930. else next();
  931. })
  932. .catch(err => next(err));
  933. },
  934. next => {
  935. StationsModule.stationHistoryModel
  936. .find({ stationId }, { documentVersion: 0, __v: 0 })
  937. .sort({ "payload.skippedAt": -1 })
  938. .limit(250)
  939. .then(response => next(null, response))
  940. .catch(next);
  941. }
  942. ],
  943. async (err, history) => {
  944. if (err) {
  945. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  946. this.log(
  947. "ERROR",
  948. "GET_STATION_HISTORY",
  949. `Getting station history for station "${stationId}" failed. "${err}"`
  950. );
  951. return cb({ status: "error", message: err });
  952. }
  953. this.log(
  954. "SUCCESS",
  955. "GET_STATION_HISTORY",
  956. `Got station history for station "${stationId}" successfully.`
  957. );
  958. return cb({ status: "success", data: { history } });
  959. }
  960. );
  961. }),
  962. getStationAutofillPlaylistsById: isLoginSometimesRequired(
  963. function getStationAutofillPlaylistsById(session, stationId, cb) {
  964. async.waterfall(
  965. [
  966. next => {
  967. StationsModule.runJob("GET_STATION", { stationId }, this)
  968. .then(station => {
  969. next(null, station);
  970. })
  971. .catch(next);
  972. },
  973. (station, next) => {
  974. if (!station) return next("Station not found.");
  975. return StationsModule.runJob(
  976. "CAN_USER_VIEW_STATION",
  977. {
  978. station,
  979. userId: session.userId
  980. },
  981. this
  982. )
  983. .then(canView => {
  984. if (!canView) next("Not allowed to get station.");
  985. else next(null, station);
  986. })
  987. .catch(err => next(err));
  988. },
  989. (station, next) => {
  990. const playlists = [];
  991. async.eachLimit(
  992. station.autofill.playlists,
  993. 1,
  994. (playlistId, next) => {
  995. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }, this)
  996. .then(playlist => {
  997. playlists.push(playlist);
  998. next();
  999. })
  1000. .catch(() => {
  1001. playlists.push(null);
  1002. next();
  1003. });
  1004. },
  1005. err => {
  1006. next(err, playlists);
  1007. }
  1008. );
  1009. }
  1010. ],
  1011. async (err, playlists) => {
  1012. if (err) {
  1013. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1014. this.log(
  1015. "ERROR",
  1016. "GET_STATION_AUTOFILL_PLAYLISTS_BY_ID",
  1017. `Getting station "${stationId}"'s autofilling playlists failed. "${err}"`
  1018. );
  1019. return cb({ status: "error", message: err });
  1020. }
  1021. this.log(
  1022. "SUCCESS",
  1023. "GET_STATION_AUTOFILL_PLAYLISTS_BY_ID",
  1024. `Got station "${stationId}"'s autofilling playlists successfully.`
  1025. );
  1026. return cb({ status: "success", data: { playlists } });
  1027. }
  1028. );
  1029. }
  1030. ),
  1031. getStationBlacklistById: isLoginSometimesRequired(function getStationBlacklistById(session, stationId, cb) {
  1032. async.waterfall(
  1033. [
  1034. next => {
  1035. StationsModule.runJob("GET_STATION", { stationId }, this)
  1036. .then(station => {
  1037. next(null, station);
  1038. })
  1039. .catch(next);
  1040. },
  1041. (station, next) => {
  1042. if (!station) return next("Station not found.");
  1043. return StationsModule.runJob(
  1044. "CAN_USER_VIEW_STATION",
  1045. {
  1046. station,
  1047. userId: session.userId
  1048. },
  1049. this
  1050. )
  1051. .then(canView => {
  1052. if (!canView) next("Not allowed to get station.");
  1053. else next(null, station);
  1054. })
  1055. .catch(err => next(err));
  1056. },
  1057. (station, next) => {
  1058. const playlists = [];
  1059. async.eachLimit(
  1060. station.blacklist,
  1061. 1,
  1062. (playlistId, next) => {
  1063. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }, this)
  1064. .then(playlist => {
  1065. playlists.push(playlist);
  1066. next();
  1067. })
  1068. .catch(() => {
  1069. playlists.push(null);
  1070. next();
  1071. });
  1072. },
  1073. err => {
  1074. next(err, playlists);
  1075. }
  1076. );
  1077. }
  1078. ],
  1079. async (err, playlists) => {
  1080. if (err) {
  1081. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1082. this.log(
  1083. "ERROR",
  1084. "GET_STATION_BLACKLIST_BY_ID",
  1085. `Getting station "${stationId}"'s blacklist failed. "${err}"`
  1086. );
  1087. return cb({ status: "error", message: err });
  1088. }
  1089. this.log(
  1090. "SUCCESS",
  1091. "GET_STATION_BLACKLIST_BY_ID",
  1092. `Got station "${stationId}"'s blacklist successfully.`
  1093. );
  1094. return cb({ status: "success", data: { playlists } });
  1095. }
  1096. );
  1097. }),
  1098. /**
  1099. * Toggle votes to skip a station
  1100. * @param {object} session - the session object automatically added by the websocket
  1101. * @param stationId - the station id
  1102. * @param {Function} cb - gets called with the result
  1103. */
  1104. toggleSkipVote: isLoginRequired(async function toggleSkipVote(session, stationId, cb) {
  1105. const stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }, this);
  1106. async.waterfall(
  1107. [
  1108. next => {
  1109. StationsModule.runJob("GET_STATION", { stationId }, this)
  1110. .then(station => next(null, station))
  1111. .catch(next);
  1112. },
  1113. (station, next) => {
  1114. if (!station) return next("Station not found.");
  1115. return StationsModule.runJob("CAN_USER_VIEW_STATION", { station, userId: session.userId }, this)
  1116. .then(canView => {
  1117. if (canView) return next(null, station);
  1118. return next("Insufficient permissions.");
  1119. })
  1120. .catch(err => next(err));
  1121. },
  1122. (station, next) => {
  1123. if (!station.currentSong) return next("There is currently no song to skip.");
  1124. const query = {};
  1125. const voted = station.currentSong.skipVotes.indexOf(session.userId) !== -1;
  1126. if (!voted) query.$push = { "currentSong.skipVotes": session.userId };
  1127. else query.$pull = { "currentSong.skipVotes": session.userId };
  1128. return stationModel.updateOne({ _id: stationId }, query, err => {
  1129. if (err) next(err);
  1130. else next(null, !voted);
  1131. });
  1132. },
  1133. (voted, next) => {
  1134. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1135. .then(station => {
  1136. next(null, station, voted);
  1137. })
  1138. .catch(next);
  1139. },
  1140. (station, voted, next) => {
  1141. if (!station) return next("Station not found.");
  1142. return StationsModule.runJob("PROCESS_SKIP_VOTES", { stationId }, this)
  1143. .then(() => next(null, voted))
  1144. .catch(next);
  1145. }
  1146. ],
  1147. async (err, voted) => {
  1148. if (err) {
  1149. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1150. this.log(
  1151. "ERROR",
  1152. "STATIONS_TOGGLE_SKIP_VOTE",
  1153. `Toggling skip vote on "${stationId}" failed. "${err}"`
  1154. );
  1155. return cb({ status: "error", message: err });
  1156. }
  1157. this.log("SUCCESS", "STATIONS_TOGGLE_SKIP_VOTE", `Toggling skip vote on "${stationId}" successful.`);
  1158. CacheModule.runJob("PUB", {
  1159. channel: "station.toggleSkipVote",
  1160. value: { stationId, voted, userId: session.userId }
  1161. });
  1162. return cb({
  1163. status: "success",
  1164. message: voted
  1165. ? "Successfully voted to skip the song."
  1166. : "Successfully removed vote to skip the song.",
  1167. data: { voted }
  1168. });
  1169. }
  1170. );
  1171. }),
  1172. /**
  1173. * Force skips a station
  1174. * @param {object} session - user session
  1175. * @param {string} stationId - the station id
  1176. * @param {Function} cb - callback
  1177. */
  1178. forceSkip(session, stationId, cb) {
  1179. async.waterfall(
  1180. [
  1181. next => {
  1182. hasPermission("stations.skip", session, stationId)
  1183. .then(() => next())
  1184. .catch(next);
  1185. },
  1186. next => {
  1187. StationsModule.runJob("GET_STATION", { stationId }, this)
  1188. .then(station => {
  1189. next(null, station);
  1190. })
  1191. .catch(next);
  1192. },
  1193. (station, next) => {
  1194. if (!station) return next("Station not found.");
  1195. return next();
  1196. }
  1197. ],
  1198. async err => {
  1199. if (err) {
  1200. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1201. this.log("ERROR", "STATIONS_FORCE_SKIP", `Force skipping station "${stationId}" failed. "${err}"`);
  1202. return cb({ status: "error", message: err });
  1203. }
  1204. StationsModule.runJob("SKIP_STATION", { stationId, natural: false, skipReason: "force_skip" });
  1205. this.log("SUCCESS", "STATIONS_FORCE_SKIP", `Force skipped station "${stationId}" successfully.`);
  1206. return cb({
  1207. status: "success",
  1208. message: "Successfully skipped station."
  1209. });
  1210. }
  1211. );
  1212. },
  1213. /**
  1214. * Leaves the user's current station
  1215. * @param {object} session - user session
  1216. * @param {string} stationId - id of station to leave
  1217. * @param {Function} cb - callback
  1218. */
  1219. leave: isLoginSometimesRequired(function leave(session, stationId, cb) {
  1220. async.waterfall(
  1221. [
  1222. next => {
  1223. StationsModule.runJob("GET_STATION", { stationId }, this)
  1224. .then(station => next(null, station))
  1225. .catch(next);
  1226. },
  1227. (station, next) => {
  1228. if (!station) return next("Station not found.");
  1229. return next();
  1230. }
  1231. ],
  1232. async (err, userCount) => {
  1233. if (err) {
  1234. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1235. this.log("ERROR", "STATIONS_LEAVE", `Leaving station "${stationId}" failed. "${err}"`);
  1236. return cb({ status: "error", message: err });
  1237. }
  1238. this.log("SUCCESS", "STATIONS_LEAVE", `Left station "${stationId}" successfully.`);
  1239. WSModule.runJob("SOCKET_LEAVE_ROOM", { socketId: session.socketId, room: `station.${stationId}` });
  1240. WSModule.runJob("SOCKETS_LEAVE_SONG_ROOMS", { sockets: [session.socketId] });
  1241. delete StationsModule.userList[session.socketId];
  1242. return cb({
  1243. status: "success",
  1244. message: "Successfully left station.",
  1245. data: { userCount }
  1246. });
  1247. }
  1248. );
  1249. }),
  1250. /**
  1251. * Updates a station's settings
  1252. * @param {object} session - user session
  1253. * @param {string} stationId - the station id
  1254. * @param {object} newStation - updated station object
  1255. * @param {Function} cb - callback
  1256. */
  1257. async update(session, stationId, newStation, cb) {
  1258. const stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }, this);
  1259. const playlistModel = await DBModule.runJob("GET_MODEL", { modelName: "playlist" }, this);
  1260. async.waterfall(
  1261. [
  1262. next => {
  1263. hasPermission("stations.update", session, stationId)
  1264. .then(() => next())
  1265. .catch(next);
  1266. },
  1267. next => {
  1268. if (newStation.requests.autorequestLimit > newStation.requests.limit)
  1269. next("The autorequest limit cannot be higher than the request limit.");
  1270. else next();
  1271. },
  1272. next => {
  1273. stationModel.findOne({ _id: stationId }, next);
  1274. },
  1275. (previousStation, next) => {
  1276. const { name, displayName, description, privacy, requests, autofill, theme, skipVoteThreshold } =
  1277. newStation;
  1278. const { enabled, limit, mode } = autofill;
  1279. // This object makes sure only certain properties can be changed by a user
  1280. const setObject = {
  1281. name,
  1282. displayName,
  1283. description,
  1284. privacy,
  1285. requests,
  1286. "autofill.enabled": enabled,
  1287. "autofill.limit": limit,
  1288. "autofill.mode": mode,
  1289. theme,
  1290. skipVoteThreshold
  1291. };
  1292. stationModel.updateOne({ _id: stationId }, { $set: setObject }, { runValidators: true }, err => {
  1293. next(err, previousStation);
  1294. });
  1295. },
  1296. (previousStation, next) => {
  1297. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1298. .then(station => next(null, station, previousStation))
  1299. .catch(next);
  1300. },
  1301. (station, previousStation, next) => {
  1302. if (
  1303. newStation.autofill.enabled &&
  1304. JSON.stringify(newStation.autofill) !== JSON.stringify(previousStation.autofill)
  1305. )
  1306. StationsModule.runJob("AUTOFILL_STATION", { stationId }, this)
  1307. .then(() => {
  1308. CacheModule.runJob("PUB", {
  1309. channel: "station.queueUpdate",
  1310. value: stationId
  1311. })
  1312. .then(() => next(null, station, previousStation))
  1313. .catch(next);
  1314. })
  1315. .catch(err => {
  1316. if (err === "Autofill is disabled in this station" || err === "Autofill limit reached")
  1317. next(null, station, previousStation);
  1318. else next(err);
  1319. });
  1320. else next(null, station, previousStation);
  1321. },
  1322. (station, previousStation, next) => {
  1323. playlistModel.updateOne(
  1324. { _id: station.playlist },
  1325. { $set: { displayName: `Station - ${station.displayName}` } },
  1326. err => {
  1327. next(err, station, previousStation);
  1328. }
  1329. );
  1330. }
  1331. ],
  1332. async (err, station, previousStation) => {
  1333. if (err) {
  1334. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1335. this.log("ERROR", "STATIONS_UPDATE", `Updating station "${stationId}" failed. "${err}"`);
  1336. return cb({ status: "error", message: err });
  1337. }
  1338. this.log("SUCCESS", "STATIONS_UPDATE", `Updated station "${stationId}" successfully.`);
  1339. CacheModule.runJob("PUB", {
  1340. channel: "station.updated",
  1341. value: { stationId, previousStation }
  1342. });
  1343. return cb({
  1344. status: "success",
  1345. message: "Successfully updated the station."
  1346. });
  1347. }
  1348. );
  1349. },
  1350. /**
  1351. * Pauses a station
  1352. * @param {object} session - user session
  1353. * @param {string} stationId - the station id
  1354. * @param {Function} cb - callback
  1355. */
  1356. async pause(session, stationId, cb) {
  1357. const stationModel = await DBModule.runJob(
  1358. "GET_MODEL",
  1359. {
  1360. modelName: "station"
  1361. },
  1362. this
  1363. );
  1364. async.waterfall(
  1365. [
  1366. next => {
  1367. hasPermission("stations.playback.toggle", session, stationId)
  1368. .then(() => next())
  1369. .catch(next);
  1370. },
  1371. next => {
  1372. StationsModule.runJob("GET_STATION", { stationId }, this)
  1373. .then(station => {
  1374. next(null, station);
  1375. })
  1376. .catch(next);
  1377. },
  1378. (station, next) => {
  1379. if (!station) return next("Station not found.");
  1380. if (station.paused) return next("That station was already paused.");
  1381. return stationModel.updateOne(
  1382. { _id: stationId },
  1383. { $set: { paused: true, pausedAt: Date.now() } },
  1384. next
  1385. );
  1386. },
  1387. (res, next) => {
  1388. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1389. .then(() => next())
  1390. .catch(next);
  1391. }
  1392. ],
  1393. async err => {
  1394. if (err) {
  1395. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1396. this.log("ERROR", "STATIONS_PAUSE", `Pausing station "${stationId}" failed. "${err}"`);
  1397. return cb({ status: "error", message: err });
  1398. }
  1399. this.log("SUCCESS", "STATIONS_PAUSE", `Paused station "${stationId}" successfully.`);
  1400. CacheModule.runJob("PUB", {
  1401. channel: "station.pause",
  1402. value: stationId
  1403. });
  1404. NotificationsModule.runJob("UNSCHEDULE", {
  1405. name: `stations.nextSong?id=${stationId}`
  1406. });
  1407. return cb({
  1408. status: "success",
  1409. message: "Successfully paused."
  1410. });
  1411. }
  1412. );
  1413. },
  1414. /**
  1415. * Resumes a station
  1416. * @param {object} session - user session
  1417. * @param {string} stationId - the station id
  1418. * @param {Function} cb - callback
  1419. */
  1420. async resume(session, stationId, cb) {
  1421. const stationModel = await DBModule.runJob(
  1422. "GET_MODEL",
  1423. {
  1424. modelName: "station"
  1425. },
  1426. this
  1427. );
  1428. async.waterfall(
  1429. [
  1430. next => {
  1431. hasPermission("stations.playback.toggle", session, stationId)
  1432. .then(() => next())
  1433. .catch(next);
  1434. },
  1435. next => {
  1436. StationsModule.runJob("GET_STATION", { stationId }, this)
  1437. .then(station => {
  1438. next(null, station);
  1439. })
  1440. .catch(next);
  1441. },
  1442. (station, next) => {
  1443. if (!station) return next("Station not found.");
  1444. if (!station.paused) return next("That station is not paused.");
  1445. station.timePaused += Date.now() - station.pausedAt;
  1446. return stationModel.updateOne(
  1447. { _id: stationId },
  1448. {
  1449. $set: { paused: false },
  1450. $inc: { timePaused: Date.now() - station.pausedAt }
  1451. },
  1452. next
  1453. );
  1454. },
  1455. (res, next) => {
  1456. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1457. .then(() => next())
  1458. .catch(next);
  1459. },
  1460. next => {
  1461. StationsModule.runJob("PROCESS_SKIP_VOTES", { stationId }, this)
  1462. .then(() => next())
  1463. .catch(next);
  1464. }
  1465. ],
  1466. async err => {
  1467. if (err) {
  1468. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1469. this.log("ERROR", "STATIONS_RESUME", `Resuming station "${stationId}" failed. "${err}"`);
  1470. return cb({ status: "error", message: err });
  1471. }
  1472. this.log("SUCCESS", "STATIONS_RESUME", `Resuming station "${stationId}" successfully.`);
  1473. CacheModule.runJob("PUB", {
  1474. channel: "station.resume",
  1475. value: stationId
  1476. });
  1477. return cb({
  1478. status: "success",
  1479. message: "Successfully resumed."
  1480. });
  1481. }
  1482. );
  1483. },
  1484. /**
  1485. * Removes a station
  1486. * @param {object} session - user session
  1487. * @param {string} stationId - the station id
  1488. * @param {Function} cb - callback
  1489. */
  1490. async remove(session, stationId, cb) {
  1491. const stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }, this);
  1492. const userModel = await DBModule.runJob("GET_MODEL", { modelName: "user" }, this);
  1493. async.waterfall(
  1494. [
  1495. next => {
  1496. hasPermission("stations.remove", session, stationId)
  1497. .then(() => next())
  1498. .catch(next);
  1499. },
  1500. next => {
  1501. stationModel.findById(stationId, (err, station) => {
  1502. if (err) return next(err);
  1503. return next(null, station);
  1504. });
  1505. },
  1506. (station, next) => {
  1507. stationModel.deleteOne({ _id: stationId }, err => next(err, station));
  1508. },
  1509. (station, next) => {
  1510. CacheModule.runJob("HDEL", { table: "stations", key: stationId }, this)
  1511. .then(() => next(null, station))
  1512. .catch(next);
  1513. },
  1514. // remove the playlist for the station
  1515. (station, next) => {
  1516. if (station.playlist)
  1517. PlaylistsModule.runJob("DELETE_PLAYLIST", { playlistId: station.playlist })
  1518. .then(() => {})
  1519. .catch(next);
  1520. next(null, station);
  1521. },
  1522. // remove reference to the station id in any array of a user's favorite stations
  1523. (station, next) => {
  1524. userModel.updateMany(
  1525. { favoriteStations: stationId },
  1526. { $pull: { favoriteStations: stationId } },
  1527. err => next(err, station)
  1528. );
  1529. }
  1530. ],
  1531. async (err, station) => {
  1532. if (err) {
  1533. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1534. this.log("ERROR", "STATIONS_REMOVE", `Removing station "${stationId}" failed. "${err}"`);
  1535. return cb({ status: "error", message: err });
  1536. }
  1537. this.log("SUCCESS", "STATIONS_REMOVE", `Removing station "${stationId}" successfully.`);
  1538. CacheModule.runJob("PUB", {
  1539. channel: "station.remove",
  1540. value: stationId
  1541. });
  1542. ActivitiesModule.runJob("ADD_ACTIVITY", {
  1543. userId: session.userId,
  1544. type: "station__remove",
  1545. payload: { message: `Removed a station named ${station.displayName}` }
  1546. });
  1547. ActivitiesModule.runJob("REMOVE_ACTIVITY_REFERENCES", { type: "stationId", stationId });
  1548. return cb({
  1549. status: "success",
  1550. message: "Successfully removed."
  1551. });
  1552. }
  1553. );
  1554. },
  1555. /**
  1556. * Create a station
  1557. * @param {object} session - the session object automatically added by the websocket
  1558. * @param data - the station data
  1559. * @param {Function} cb - gets called with the result
  1560. */
  1561. create: isLoginRequired(async function create(session, data, cb) {
  1562. const stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }, this);
  1563. const playlistModel = await DBModule.runJob("GET_MODEL", { modelName: "playlist" }, this);
  1564. data.name = data.name.toLowerCase();
  1565. let blacklist = [
  1566. "about",
  1567. "support",
  1568. "staff",
  1569. "help",
  1570. "news",
  1571. "terms",
  1572. "privacy",
  1573. "profile",
  1574. "c",
  1575. "community",
  1576. "tos",
  1577. "login",
  1578. "register",
  1579. "p",
  1580. "official",
  1581. "o",
  1582. "faq",
  1583. "team",
  1584. "donate",
  1585. "buy",
  1586. "shop",
  1587. "forums",
  1588. "explore",
  1589. "settings",
  1590. "admin",
  1591. "auth",
  1592. "reset_password",
  1593. "backend",
  1594. "api",
  1595. "songs",
  1596. "playlists",
  1597. "playlist",
  1598. "albums",
  1599. "artists",
  1600. "artist",
  1601. "station"
  1602. ];
  1603. if (data.type === "community") blacklist = [...blacklist, ...config.get("blacklistedCommunityStationNames")];
  1604. async.waterfall(
  1605. [
  1606. next => {
  1607. if (!data) return next("Invalid data.");
  1608. return next();
  1609. },
  1610. next => {
  1611. stationModel.findOne(
  1612. {
  1613. $or: [{ name: data.name }, { displayName: new RegExp(`^${data.displayName}$`, "i") }]
  1614. },
  1615. next
  1616. );
  1617. },
  1618. (station, next) => {
  1619. this.log(station);
  1620. if (station) return next("A station with that name or display name already exists.");
  1621. if (blacklist.indexOf(data.name) !== -1)
  1622. return next("That name is blacklisted. Please use a different name.");
  1623. if (data.type === "official")
  1624. return hasPermission("stations.create.official", session)
  1625. .then(() => next())
  1626. .catch(() => next("Insufficient permissions."));
  1627. return next();
  1628. },
  1629. next => {
  1630. const stationId = mongoose.Types.ObjectId();
  1631. playlistModel.create(
  1632. {
  1633. displayName: `Station - ${data.name}`,
  1634. songs: [],
  1635. createdBy: data.type === "official" ? "Musare" : session.userId,
  1636. createdFor: `${stationId}`,
  1637. createdAt: Date.now(),
  1638. type: "station"
  1639. },
  1640. (err, playlist) => {
  1641. next(err, playlist, stationId);
  1642. }
  1643. );
  1644. },
  1645. (playlist, stationId, next) => {
  1646. const { name, displayName, description, type, privacy } = data;
  1647. if (type === "official") {
  1648. stationModel.create(
  1649. {
  1650. _id: stationId,
  1651. name,
  1652. displayName,
  1653. description,
  1654. playlist: playlist._id,
  1655. type,
  1656. privacy,
  1657. queue: [],
  1658. currentSong: null
  1659. },
  1660. next
  1661. );
  1662. } else {
  1663. stationModel.create(
  1664. {
  1665. _id: stationId,
  1666. name,
  1667. displayName,
  1668. description,
  1669. playlist: playlist._id,
  1670. type,
  1671. privacy,
  1672. owner: session.userId,
  1673. queue: [],
  1674. currentSong: null
  1675. },
  1676. next
  1677. );
  1678. }
  1679. },
  1680. // This extra step is needed because Mongoose decides to create an object with empty arrays for currentSong for some reason
  1681. (station, next) => {
  1682. stationModel.updateOne(
  1683. { _id: station._id },
  1684. {
  1685. $set: {
  1686. currentSong: null
  1687. }
  1688. },
  1689. err => {
  1690. next(err, station);
  1691. }
  1692. );
  1693. }
  1694. ],
  1695. async (err, station) => {
  1696. if (err) {
  1697. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1698. this.log("ERROR", "STATIONS_CREATE", `Creating station failed. "${err}"`);
  1699. cb({ status: "error", message: err });
  1700. } else {
  1701. this.log("SUCCESS", "STATIONS_CREATE", `Created station "${station._id}" successfully.`);
  1702. CacheModule.runJob("PUB", {
  1703. channel: "station.create",
  1704. value: station._id
  1705. });
  1706. ActivitiesModule.runJob("ADD_ACTIVITY", {
  1707. userId: session.userId,
  1708. type: "station__create",
  1709. payload: {
  1710. message: `Created a station named <stationId>${station.displayName}</stationId>`,
  1711. stationId: station._id
  1712. }
  1713. });
  1714. cb({
  1715. status: "success",
  1716. message: "Successfully created station."
  1717. });
  1718. }
  1719. }
  1720. );
  1721. }),
  1722. /**
  1723. * Adds song to station queue
  1724. * @param {object} session - the session object automatically added by the websocket
  1725. * @param stationId - the station id
  1726. * @param mediaSource - the song id
  1727. * @param requestType - whether the song was autorequested or requested normally
  1728. * @param {Function} cb - gets called with the result
  1729. */
  1730. addToQueue: isLoginRequired(async function addToQueue(session, stationId, mediaSource, requestType, cb) {
  1731. async.waterfall(
  1732. [
  1733. next => {
  1734. StationsModule.runJob("GET_STATION", { stationId }, this)
  1735. .then(station => {
  1736. next(null, station);
  1737. })
  1738. .catch(next);
  1739. },
  1740. (station, next) => {
  1741. if (!station) return next("Station not found.");
  1742. if (!station.requests.enabled) return next("Requests are disabled in this station.");
  1743. if (
  1744. station.requests.access === "owner" ||
  1745. (station.requests.access === "user" && station.privacy === "private")
  1746. ) {
  1747. return hasPermission("stations.request", session, stationId)
  1748. .then(() => next(null, station))
  1749. .catch(() => next("You do not have permission to add songs to queue."));
  1750. }
  1751. return next(null, station);
  1752. },
  1753. (station, next) =>
  1754. StationsModule.runJob(
  1755. "CAN_USER_VIEW_STATION",
  1756. {
  1757. station,
  1758. userId: session.userId
  1759. },
  1760. this
  1761. )
  1762. .then(canView => {
  1763. if (canView) return next();
  1764. return next("Insufficient permissions.");
  1765. })
  1766. .catch(err => next(err)),
  1767. next =>
  1768. StationsModule.runJob(
  1769. "ADD_TO_QUEUE",
  1770. {
  1771. stationId,
  1772. mediaSource,
  1773. requestUser: session.userId,
  1774. requestType
  1775. },
  1776. this
  1777. )
  1778. .then(() => next())
  1779. .catch(next)
  1780. ],
  1781. async err => {
  1782. if (err) {
  1783. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1784. this.log(
  1785. "ERROR",
  1786. "STATIONS_ADD_SONG_TO_QUEUE",
  1787. `Adding song "${mediaSource}" to station "${stationId}" queue failed. "${err}"`
  1788. );
  1789. return cb({ status: "error", message: err });
  1790. }
  1791. this.log(
  1792. "SUCCESS",
  1793. "STATIONS_ADD_SONG_TO_QUEUE",
  1794. `Added song "${mediaSource}" to station "${stationId}" successfully.`
  1795. );
  1796. return cb({
  1797. status: "success",
  1798. message: "Successfully added song to queue."
  1799. });
  1800. }
  1801. );
  1802. }),
  1803. /**
  1804. * Removes song from station queue
  1805. * @param {object} session - user session
  1806. * @param {string} stationId - the station id
  1807. * @param {string} mediaSource - the media source
  1808. * @param {Function} cb - callback
  1809. */
  1810. async removeFromQueue(session, stationId, mediaSource, cb) {
  1811. async.waterfall(
  1812. [
  1813. next => {
  1814. hasPermission("stations.queue.remove", session, stationId)
  1815. .then(() => next())
  1816. .catch(next);
  1817. },
  1818. next => {
  1819. if (!mediaSource) return next("Invalid media source.");
  1820. return StationsModule.runJob("REMOVE_FROM_QUEUE", { stationId, mediaSource }, this)
  1821. .then(() => next())
  1822. .catch(next);
  1823. }
  1824. ],
  1825. async err => {
  1826. if (err) {
  1827. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1828. this.log(
  1829. "ERROR",
  1830. "STATIONS_REMOVE_SONG_TO_QUEUE",
  1831. `Removing song "${mediaSource}" from station "${stationId}" queue failed. "${err}"`
  1832. );
  1833. return cb({ status: "error", message: err });
  1834. }
  1835. this.log(
  1836. "SUCCESS",
  1837. "STATIONS_REMOVE_SONG_TO_QUEUE",
  1838. `Removed song "${mediaSource}" from station "${stationId}" successfully.`
  1839. );
  1840. return cb({
  1841. status: "success",
  1842. message: "Successfully removed song from queue."
  1843. });
  1844. }
  1845. );
  1846. },
  1847. /**
  1848. * Gets the queue from a station
  1849. * @param {object} session - user session
  1850. * @param {string} stationId - the station id
  1851. * @param {Function} cb - callback
  1852. */
  1853. getQueue: isLoginSometimesRequired(function getQueue(session, stationId, cb) {
  1854. async.waterfall(
  1855. [
  1856. next => {
  1857. StationsModule.runJob("GET_STATION", { stationId }, this)
  1858. .then(station => next(null, station))
  1859. .catch(next);
  1860. },
  1861. (station, next) => {
  1862. if (!station) return next("Station not found.");
  1863. return next(null, station);
  1864. },
  1865. (station, next) => {
  1866. StationsModule.runJob("CAN_USER_VIEW_STATION", { station, userId: session.userId }, this)
  1867. .then(canView => {
  1868. if (canView) return next(null, station);
  1869. return next("Insufficient permissions.");
  1870. })
  1871. .catch(err => next(err));
  1872. },
  1873. (station, next) => next(null, station.queue)
  1874. ],
  1875. async (err, queue) => {
  1876. if (err) {
  1877. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1878. this.log(
  1879. "ERROR",
  1880. "STATIONS_GET_QUEUE",
  1881. `Getting queue for station "${stationId}" failed. "${err}"`
  1882. );
  1883. return cb({ status: "error", message: err });
  1884. }
  1885. this.log("SUCCESS", "STATIONS_GET_QUEUE", `Got queue for station "${stationId}" successfully.`);
  1886. return cb({
  1887. status: "success",
  1888. message: "Successfully got queue.",
  1889. data: { queue }
  1890. });
  1891. }
  1892. );
  1893. }),
  1894. /**
  1895. * Reposition a song in station queue
  1896. * @param {object} session - user session
  1897. * @param {string} stationId - the station id
  1898. * @param {object} song - contains details about the song that is to be repositioned
  1899. * @param {string} song.mediaSource - the media source of the song
  1900. * @param {number} song.newIndex - the new position for the song in the queue
  1901. * @param {number} song.oldIndex - the old position of the song in the queue
  1902. * @param {Function} cb - callback
  1903. */
  1904. async repositionSongInQueue(session, stationId, song, cb) {
  1905. const stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }, this);
  1906. async.waterfall(
  1907. [
  1908. next => {
  1909. hasPermission("stations.queue.reposition", session, stationId)
  1910. .then(() => next())
  1911. .catch(next);
  1912. },
  1913. next => {
  1914. if (!song || !song.mediaSource) return next("You must provide a song to reposition.");
  1915. return next();
  1916. },
  1917. // remove song from queue
  1918. next => {
  1919. stationModel.updateOne(
  1920. { _id: stationId },
  1921. { $pull: { queue: { mediaSource: song.mediaSource } } },
  1922. next
  1923. );
  1924. },
  1925. // add song back to queue (in new position)
  1926. (res, next) => {
  1927. stationModel.updateOne(
  1928. { _id: stationId },
  1929. { $push: { queue: { $each: [song], $position: song.newIndex } } },
  1930. err => next(err)
  1931. );
  1932. },
  1933. // update the cache representation of the station
  1934. next => {
  1935. StationsModule.runJob("UPDATE_STATION", { stationId }, this)
  1936. .then(station => next(null, station))
  1937. .catch(next);
  1938. }
  1939. ],
  1940. async (err, station) => {
  1941. if (err) {
  1942. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  1943. this.log(
  1944. "ERROR",
  1945. "STATIONS_REPOSITION_SONG_IN_QUEUE",
  1946. `Repositioning song ${song.mediaSource} in queue of station "${stationId}" failed. "${err}"`
  1947. );
  1948. return cb({ status: "error", message: err });
  1949. }
  1950. this.log(
  1951. "SUCCESS",
  1952. "STATIONS_REPOSITION_SONG_IN_QUEUE",
  1953. `Repositioned song ${song.mediaSource} in queue of station "${stationId}" successfully.`
  1954. );
  1955. const queueOrder = station.queue.map(song => song.mediaSource);
  1956. CacheModule.runJob("PUB", {
  1957. channel: "station.changeQueueOrder",
  1958. value: {
  1959. queueOrder,
  1960. stationId
  1961. }
  1962. });
  1963. return cb({
  1964. status: "success",
  1965. message: "Successfully repositioned song in queue."
  1966. });
  1967. }
  1968. );
  1969. },
  1970. /**
  1971. * Autofill a playlist in a station
  1972. * @param {object} session - user session
  1973. * @param {string} stationId - the station id
  1974. * @param {string} playlistId - the playlist id
  1975. * @param {Function} cb - callback
  1976. */
  1977. async autofillPlaylist(session, stationId, playlistId, cb) {
  1978. async.waterfall(
  1979. [
  1980. next => {
  1981. hasPermission("stations.autofill", session, stationId)
  1982. .then(() => next())
  1983. .catch(next);
  1984. },
  1985. next => {
  1986. StationsModule.runJob("GET_STATION", { stationId }, this)
  1987. .then(station => next(null, station))
  1988. .catch(next);
  1989. },
  1990. (station, next) => {
  1991. PlaylistsModule.runJob("GET_PLAYLIST", { playlistId }, this)
  1992. .then(playlist => next(null, station, playlist))
  1993. .catch(next);
  1994. },
  1995. (station, playlist, next) => {
  1996. if (!playlist) return next("Playlist not found");
  1997. if (playlist.privacy !== "public" && playlist.createdBy !== session.userId)
  1998. return hasPermission("playlists.get", session)
  1999. .then(() => next(null, station, playlist))
  2000. .catch(() => next("User unauthorised to view playlist."));
  2001. return next(null, station, playlist);
  2002. },
  2003. (station, playlist, next) => {
  2004. if (!station) return next("Station not found.");
  2005. if (station.type === "official" && ["genre", "admin"].indexOf(playlist.type) === -1)
  2006. return next("Official statuibs are only allowed to autofill genre and admin playlists.");
  2007. if (
  2008. station.type === "community" &&
  2009. ["user", "user-liked", "user-disliked", "genre", "admin"].indexOf(playlist.type) === -1
  2010. )
  2011. return next("Community stations are only allowed to autofill user, genre and admin playlists.");
  2012. if (station.autofill.playlists.indexOf(playlistId) !== -1)
  2013. return next("That playlist is already autofilling.");
  2014. if (station.autofill.mode === "sequential" && station.autofill.playlists.length > 0)
  2015. return next("Error: Only 1 playlist can be autofilling in sequential mode.");
  2016. return next();
  2017. },
  2018. next => {
  2019. StationsModule.runJob("AUTOFILL_PLAYLIST", { stationId, playlistId }, this)
  2020. .then(() => {
  2021. next();
  2022. })
  2023. .catch(next);
  2024. }
  2025. ],
  2026. async err => {
  2027. if (err) {
  2028. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2029. this.log(
  2030. "ERROR",
  2031. "STATIONS_AUTOFILL_PLAYLIST",
  2032. `Including playlist "${playlistId}" for station "${stationId}" failed. "${err}"`
  2033. );
  2034. return cb({ status: "error", message: err });
  2035. }
  2036. this.log(
  2037. "SUCCESS",
  2038. "STATIONS_AUTOFILL_PLAYLIST",
  2039. `Including playlist "${playlistId}" for station "${stationId}" successfully.`
  2040. );
  2041. PlaylistsModule.runJob("AUTOFILL_STATION_PLAYLIST", { stationId }).then().catch();
  2042. CacheModule.runJob("PUB", {
  2043. channel: "station.autofillPlaylist",
  2044. value: {
  2045. playlistId,
  2046. stationId
  2047. }
  2048. });
  2049. return cb({
  2050. status: "success",
  2051. message: "Successfully added autofill playlist."
  2052. });
  2053. }
  2054. );
  2055. },
  2056. /**
  2057. * Remove autofilled playlist from a station
  2058. * @param {object} session - user session
  2059. * @param {string} stationId - the station id
  2060. * @param {string} playlistId - the playlist id
  2061. * @param {Function} cb - callback
  2062. */
  2063. async removeAutofillPlaylist(session, stationId, playlistId, cb) {
  2064. async.waterfall(
  2065. [
  2066. next => {
  2067. hasPermission("stations.autofill", session, stationId)
  2068. .then(() => next())
  2069. .catch(next);
  2070. },
  2071. next => {
  2072. StationsModule.runJob("GET_STATION", { stationId }, this)
  2073. .then(station => next(null, station))
  2074. .catch(next);
  2075. },
  2076. (station, next) => {
  2077. if (!station) return next("Station not found.");
  2078. if (station.autofill.playlists.indexOf(playlistId) === -1)
  2079. return next("That playlist is not autofilling.");
  2080. return next();
  2081. },
  2082. next => {
  2083. StationsModule.runJob("REMOVE_AUTOFILL_PLAYLIST", { stationId, playlistId }, this)
  2084. .then(() => {
  2085. next();
  2086. })
  2087. .catch(next);
  2088. }
  2089. ],
  2090. async err => {
  2091. if (err) {
  2092. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2093. this.log(
  2094. "ERROR",
  2095. "STATIONS_REMOVE_AUTOFILL_PLAYLIST",
  2096. `Removing autofill playlist "${playlistId}" for station "${stationId}" failed. "${err}"`
  2097. );
  2098. return cb({ status: "error", message: err });
  2099. }
  2100. this.log(
  2101. "SUCCESS",
  2102. "STATIONS_REMOVE_AUTOFILL_PLAYLIST",
  2103. `Removing autofill playlist "${playlistId}" for station "${stationId}" successfully.`
  2104. );
  2105. PlaylistsModule.runJob("AUTOFILL_STATION_PLAYLIST", { stationId }).then().catch();
  2106. CacheModule.runJob("PUB", {
  2107. channel: "station.removedAutofillPlaylist",
  2108. value: {
  2109. playlistId,
  2110. stationId
  2111. }
  2112. });
  2113. return cb({
  2114. status: "success",
  2115. message: "Successfully removed autofill playlist."
  2116. });
  2117. }
  2118. );
  2119. },
  2120. /**
  2121. * Blacklist a playlist in a station
  2122. * @param {object} session - user session
  2123. * @param {string} stationId - the station id
  2124. * @param {string} playlistId - the playlist id
  2125. * @param {Function} cb - callback
  2126. */
  2127. async blacklistPlaylist(session, stationId, playlistId, cb) {
  2128. async.waterfall(
  2129. [
  2130. next => {
  2131. hasPermission("stations.blacklist", session, stationId)
  2132. .then(() => next())
  2133. .catch(next);
  2134. },
  2135. next => {
  2136. StationsModule.runJob("GET_STATION", { stationId }, this)
  2137. .then(station => next(null, station))
  2138. .catch(next);
  2139. },
  2140. (station, next) => {
  2141. if (!station) return next("Station not found.");
  2142. if (station.blacklist.indexOf(playlistId) !== -1)
  2143. return next("That playlist is already blacklisted.");
  2144. return next();
  2145. },
  2146. next => {
  2147. StationsModule.runJob("BLACKLIST_PLAYLIST", { stationId, playlistId }, this)
  2148. .then(() => {
  2149. next();
  2150. })
  2151. .catch(next);
  2152. }
  2153. ],
  2154. async err => {
  2155. if (err) {
  2156. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2157. this.log(
  2158. "ERROR",
  2159. "STATIONS_BLACKLIST_PLAYLIST",
  2160. `Blacklisting playlist "${playlistId}" for station "${stationId}" failed. "${err}"`
  2161. );
  2162. return cb({ status: "error", message: err });
  2163. }
  2164. this.log(
  2165. "SUCCESS",
  2166. "STATIONS_BLACKLIST_PLAYLIST",
  2167. `Blacklisting playlist "${playlistId}" for station "${stationId}" successfully.`
  2168. );
  2169. PlaylistsModule.runJob("AUTOFILL_STATION_PLAYLIST", { stationId }).then().catch();
  2170. CacheModule.runJob("PUB", {
  2171. channel: "station.blacklistedPlaylist",
  2172. value: {
  2173. playlistId,
  2174. stationId
  2175. }
  2176. });
  2177. return cb({
  2178. status: "success",
  2179. message: "Successfully blacklisted playlist."
  2180. });
  2181. }
  2182. );
  2183. },
  2184. /**
  2185. * Remove blacklisted a playlist from a station
  2186. * @param {object} session - user session
  2187. * @param {string} stationId - the station id
  2188. * @param {string} playlistId - the playlist id
  2189. * @param {Function} cb - callback
  2190. */
  2191. async removeBlacklistedPlaylist(session, stationId, playlistId, cb) {
  2192. async.waterfall(
  2193. [
  2194. next => {
  2195. hasPermission("stations.blacklist", session, stationId)
  2196. .then(() => next())
  2197. .catch(next);
  2198. },
  2199. next => {
  2200. StationsModule.runJob("GET_STATION", { stationId }, this)
  2201. .then(station => next(null, station))
  2202. .catch(next);
  2203. },
  2204. (station, next) => {
  2205. if (!station) return next("Station not found.");
  2206. if (station.blacklist.indexOf(playlistId) === -1) return next("That playlist is not blacklisted.");
  2207. return next();
  2208. },
  2209. next => {
  2210. StationsModule.runJob("REMOVE_BLACKLISTED_PLAYLIST", { stationId, playlistId }, this)
  2211. .then(() => {
  2212. next();
  2213. })
  2214. .catch(next);
  2215. }
  2216. ],
  2217. async err => {
  2218. if (err) {
  2219. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2220. this.log(
  2221. "ERROR",
  2222. "STATIONS_REMOVE_BLACKLISTED_PLAYLIST",
  2223. `Removing blacklisted playlist "${playlistId}" for station "${stationId}" failed. "${err}"`
  2224. );
  2225. return cb({ status: "error", message: err });
  2226. }
  2227. this.log(
  2228. "SUCCESS",
  2229. "STATIONS_REMOVE_BLACKLISTED_PLAYLIST",
  2230. `Removing blacklisted playlist "${playlistId}" for station "${stationId}" successfully.`
  2231. );
  2232. PlaylistsModule.runJob("AUTOFILL_STATION_PLAYLIST", { stationId }).then().catch();
  2233. CacheModule.runJob("PUB", {
  2234. channel: "station.removedBlacklistedPlaylist",
  2235. value: {
  2236. playlistId,
  2237. stationId
  2238. }
  2239. });
  2240. return cb({
  2241. status: "success",
  2242. message: "Successfully removed blacklisted playlist."
  2243. });
  2244. }
  2245. );
  2246. },
  2247. /**
  2248. * Favorites a station
  2249. * @param {object} session - the session object automatically added by the websocket
  2250. * @param {string} stationId - the station to favorite
  2251. * @param {Function} cb - gets called with the result
  2252. */
  2253. favoriteStation: isLoginRequired(async function favoriteStation(session, stationId, cb) {
  2254. const userModel = await DBModule.runJob("GET_MODEL", { modelName: "user" }, this);
  2255. async.waterfall(
  2256. [
  2257. next => {
  2258. StationsModule.runJob("GET_STATION", { stationId }, this)
  2259. .then(station => next(null, station))
  2260. .catch(next);
  2261. },
  2262. (station, next) => {
  2263. if (!station) return next("Station not found.");
  2264. return StationsModule.runJob("CAN_USER_VIEW_STATION", { station, userId: session.userId }, this)
  2265. .then(canView => {
  2266. if (canView) return next(null, station);
  2267. return next("Insufficient permissions.");
  2268. })
  2269. .catch(err => next(err));
  2270. },
  2271. (station, next) => {
  2272. userModel.updateOne(
  2273. { _id: session.userId },
  2274. { $addToSet: { favoriteStations: stationId } },
  2275. (err, res) => next(err, station, res)
  2276. );
  2277. },
  2278. (station, res, next) => {
  2279. if (res.nModified === 0) return next("The station was already favorited.");
  2280. return next(null, station);
  2281. }
  2282. ],
  2283. async (err, station) => {
  2284. if (err) {
  2285. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2286. this.log("ERROR", "FAVORITE_STATION", `Favoriting station "${stationId}" failed. "${err}"`);
  2287. return cb({ status: "error", message: err });
  2288. }
  2289. this.log("SUCCESS", "FAVORITE_STATION", `Favorited station "${stationId}" successfully.`);
  2290. CacheModule.runJob("PUB", {
  2291. channel: "user.favoritedStation",
  2292. value: {
  2293. userId: session.userId,
  2294. stationId
  2295. }
  2296. });
  2297. ActivitiesModule.runJob("ADD_ACTIVITY", {
  2298. userId: session.userId,
  2299. type: "station__favorite",
  2300. payload: {
  2301. message: `Favorited station <stationId>${station.displayName}</stationId>`,
  2302. stationId
  2303. }
  2304. });
  2305. return cb({
  2306. status: "success",
  2307. message: "Succesfully favorited station."
  2308. });
  2309. }
  2310. );
  2311. }),
  2312. /**
  2313. * Unfavorites a station
  2314. * @param {object} session - the session object automatically added by the websocket
  2315. * @param {string} stationId - the station to unfavorite
  2316. * @param {Function} cb - gets called with the result
  2317. */
  2318. unfavoriteStation: isLoginRequired(async function unfavoriteStation(session, stationId, cb) {
  2319. const userModel = await DBModule.runJob("GET_MODEL", { modelName: "user" }, this);
  2320. async.waterfall(
  2321. [
  2322. next => {
  2323. userModel.updateOne({ _id: session.userId }, { $pull: { favoriteStations: stationId } }, next);
  2324. },
  2325. (res, next) => {
  2326. if (res.nModified === 0) return next("The station wasn't favorited.");
  2327. return next();
  2328. },
  2329. next => {
  2330. StationsModule.runJob("GET_STATION", { stationId }, this)
  2331. .then(station => next(null, station))
  2332. .catch(next);
  2333. }
  2334. ],
  2335. async (err, station) => {
  2336. if (err) {
  2337. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2338. this.log("ERROR", "UNFAVORITE_STATION", `Unfavoriting station "${stationId}" failed. "${err}"`);
  2339. return cb({ status: "error", message: err });
  2340. }
  2341. this.log("SUCCESS", "UNFAVORITE_STATION", `Unfavorited station "${stationId}" successfully.`);
  2342. CacheModule.runJob("PUB", {
  2343. channel: "user.unfavoritedStation",
  2344. value: {
  2345. userId: session.userId,
  2346. stationId
  2347. }
  2348. });
  2349. ActivitiesModule.runJob("ADD_ACTIVITY", {
  2350. userId: session.userId,
  2351. type: "station__unfavorite",
  2352. payload: {
  2353. message: `Unfavorited station <stationId>${station.displayName}</stationId>`,
  2354. stationId
  2355. }
  2356. });
  2357. return cb({
  2358. status: "success",
  2359. message: "Succesfully unfavorited station."
  2360. });
  2361. }
  2362. );
  2363. }),
  2364. /**
  2365. * Clears every station queue
  2366. * @param {object} session - the session object automatically added by the websocket
  2367. * @param {Function} cb - gets called with the result
  2368. */
  2369. clearEveryStationQueue: useHasPermission(
  2370. "stations.clearEveryStationQueue",
  2371. async function clearEveryStationQueue(session, cb) {
  2372. this.keepLongJob();
  2373. this.publishProgress({
  2374. status: "started",
  2375. title: "Clear every station queue",
  2376. message: "Clearing every station queue.",
  2377. id: this.toString()
  2378. });
  2379. await CacheModule.runJob("RPUSH", { key: `longJobs.${session.userId}`, value: this.toString() }, this);
  2380. await CacheModule.runJob(
  2381. "PUB",
  2382. {
  2383. channel: "longJob.added",
  2384. value: { jobId: this.toString(), userId: session.userId }
  2385. },
  2386. this
  2387. );
  2388. async.waterfall(
  2389. [
  2390. next => {
  2391. StationsModule.runJob("CLEAR_EVERY_STATION_QUEUE", {}, this)
  2392. .then(() => next())
  2393. .catch(next);
  2394. }
  2395. ],
  2396. async err => {
  2397. if (err) {
  2398. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2399. this.log("ERROR", "CLEAR_EVERY_STATION_QUEUE", `Clearing every station queue failed. "${err}"`);
  2400. this.publishProgress({
  2401. status: "error",
  2402. message: err
  2403. });
  2404. return cb({ status: "error", message: err });
  2405. }
  2406. this.log("SUCCESS", "CLEAR_EVERY_STATION_QUEUE", "Clearing every station queue was successful.");
  2407. this.publishProgress({
  2408. status: "success",
  2409. message: "Successfully cleared every station queue."
  2410. });
  2411. return cb({ status: "success", message: "Successfully cleared every station queue." });
  2412. }
  2413. );
  2414. }
  2415. ),
  2416. /**
  2417. * Reset a station queue
  2418. * @param {object} session - the session object automatically added by the websocket
  2419. * @param {string} stationId - the station id
  2420. * @param {Function} cb - gets called with the result
  2421. */
  2422. async resetQueue(session, stationId, cb) {
  2423. async.waterfall(
  2424. [
  2425. next => {
  2426. hasPermission("stations.queue.reset", session, stationId)
  2427. .then(() => next())
  2428. .catch(next);
  2429. },
  2430. next => {
  2431. StationsModule.runJob("RESET_QUEUE", { stationId }, this)
  2432. .then(() => next())
  2433. .catch(next);
  2434. }
  2435. ],
  2436. async err => {
  2437. if (err) {
  2438. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2439. this.log("ERROR", "RESET_QUEUE", `Resetting station queue failed. "${err}"`);
  2440. return cb({ status: "error", message: err });
  2441. }
  2442. this.log("SUCCESS", "RESET_QUEUE", "Resetting station queue was successful.");
  2443. return cb({ status: "success", message: "Successfully reset station queue." });
  2444. }
  2445. );
  2446. },
  2447. /**
  2448. * Gets skip votes for a station
  2449. * @param {object} session - the session object automatically added by the websocket
  2450. * @param {string} stationId - the station id
  2451. * @param {string} songId - the song id to get skipvotes for
  2452. * @param {Function} cb - gets called with the result
  2453. */
  2454. getSkipVotes: isLoginRequired(async function getSkipVotes(session, stationId, songId, cb) {
  2455. async.waterfall(
  2456. [
  2457. next => {
  2458. StationsModule.runJob("GET_STATION", { stationId }, this)
  2459. .then(res => next(null, res.currentSong))
  2460. .catch(console.log);
  2461. },
  2462. (currentSong, next) => {
  2463. if (currentSong && currentSong._id === songId)
  2464. next(null, {
  2465. skipVotes: currentSong.skipVotes.length,
  2466. skipVotesCurrent: true,
  2467. voted: currentSong.skipVotes.indexOf(session.userId) !== -1
  2468. });
  2469. else
  2470. next(null, {
  2471. skipVotes: 0,
  2472. skipVotesCurrent: false,
  2473. voted: false
  2474. });
  2475. }
  2476. ],
  2477. async (err, data) => {
  2478. if (err) {
  2479. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2480. this.log(
  2481. "ERROR",
  2482. "STATIONS_GET_SKIP_VOTES",
  2483. `User "${session.userId}" failed to get skip votes for ${stationId}. "${err}"`
  2484. );
  2485. return cb({ status: "error", message: err });
  2486. }
  2487. return cb({
  2488. status: "success",
  2489. data
  2490. });
  2491. }
  2492. );
  2493. }),
  2494. /**
  2495. * Add DJ to station
  2496. * @param {object} session - the session object automatically added by the websocket
  2497. * @param {string} stationId - the station id
  2498. * @param {string} userId - the dj user id
  2499. * @param {Function} cb - gets called with the result
  2500. */
  2501. async addDj(session, stationId, userId, cb) {
  2502. async.waterfall(
  2503. [
  2504. next => {
  2505. hasPermission("stations.djs.add", session, stationId)
  2506. .then(() => next())
  2507. .catch(next);
  2508. },
  2509. next => {
  2510. StationsModule.runJob("ADD_DJ", { stationId, userId }, this)
  2511. .then(() => next())
  2512. .catch(next);
  2513. }
  2514. ],
  2515. async err => {
  2516. if (err) {
  2517. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2518. this.log("ERROR", "ADD_DJ", `Adding DJ failed. "${err}"`);
  2519. return cb({ status: "error", message: err });
  2520. }
  2521. this.log("SUCCESS", "ADD_DJ", "Adding DJ was successful.");
  2522. return cb({ status: "success", message: "Successfully added DJ." });
  2523. }
  2524. );
  2525. },
  2526. /**
  2527. * Remove DJ from station
  2528. * @param {object} session - the session object automatically added by the websocket
  2529. * @param {string} stationId - the station id
  2530. * @param {string} userId - the dj user id
  2531. * @param {Function} cb - gets called with the result
  2532. */
  2533. async removeDj(session, stationId, userId, cb) {
  2534. async.waterfall(
  2535. [
  2536. next => {
  2537. hasPermission("stations.djs.remove", session, stationId)
  2538. .then(() => next())
  2539. .catch(next);
  2540. },
  2541. next => {
  2542. StationsModule.runJob("REMOVE_DJ", { stationId, userId }, this)
  2543. .then(() => next())
  2544. .catch(next);
  2545. }
  2546. ],
  2547. async err => {
  2548. if (err) {
  2549. err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);
  2550. this.log("ERROR", "REMOVE_DJ", `Removing DJ failed. "${err}"`);
  2551. return cb({ status: "error", message: err });
  2552. }
  2553. this.log("SUCCESS", "REMOVE_DJ", "Removing DJ was successful.");
  2554. return cb({ status: "success", message: "Successfully removed DJ." });
  2555. }
  2556. );
  2557. },
  2558. /**
  2559. * Sets the state of the current user session
  2560. * @param {object} session - the session object automatically added by the websocket
  2561. * @param {string} newStationState - the new state
  2562. * @param {Function} cb - gets called with the result
  2563. */
  2564. setStationState(session, newStationState, cb) {
  2565. session.stationState = newStationState;
  2566. cb({ status: "success" });
  2567. }
  2568. };