| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698 | import async from "async";// import mongoose from "mongoose";import CoreClass from "../core";let StationsModule;let CacheModule;let DBModule;let UtilsModule;let WSModule;let SongsModule;let PlaylistsModule;let NotificationsModule;class _StationsModule extends CoreClass {	// eslint-disable-next-line require-jsdoc	constructor() {		super("stations");		StationsModule = this;	}	/**	 * Initialises the stations module	 *	 * @returns {Promise} - returns promise (reject, resolve)	 */	async initialize() {		CacheModule = this.moduleManager.modules.cache;		DBModule = this.moduleManager.modules.db;		UtilsModule = this.moduleManager.modules.utils;		WSModule = this.moduleManager.modules.ws;		SongsModule = this.moduleManager.modules.songs;		PlaylistsModule = this.moduleManager.modules.playlists;		NotificationsModule = this.moduleManager.modules.notifications;		// this.defaultSong = {		// 	youtubeId: "60ItHLz5WEA",		// 	title: "Faded - Alan Walker",		// 	duration: 212,		// 	skipDuration: 0,		// 	likes: -1,		// 	dislikes: -1,		// 	requestedAt: Date.now(),		// 	status: "unverified"		// };		this.userList = {};		this.usersPerStation = {};		this.usersPerStationCount = {};		// TEMP		CacheModule.runJob("SUB", {			channel: "station.pause",			cb: async stationId => {				NotificationsModule.runJob("REMOVE", {					subscription: `stations.nextSong?id=${stationId}`				}).then();			}		});		CacheModule.runJob("SUB", {			channel: "station.resume",			cb: async stationId => {				StationsModule.runJob("INITIALIZE_STATION", { stationId }).then();			}		});		CacheModule.runJob("SUB", {			channel: "station.queueUpdate",			cb: async stationId => {				StationsModule.runJob("GET_STATION", { stationId }).then(station => {					if (!station.currentSong && station.queue.length > 0) {						StationsModule.runJob("INITIALIZE_STATION", {							stationId						}).then();					}				});			}		});		CacheModule.runJob("SUB", {			channel: "station.newOfficialPlaylist",			cb: async stationId => {				CacheModule.runJob("HGET", {					table: "officialPlaylists",					key: stationId				}).then(playlistObj => {					if (playlistObj) {						WSModule.runJob("EMIT_TO_ROOM", {							room: `station.${stationId}`,							args: ["event:newOfficialPlaylist", { data: { playlist: playlistObj.songs } }]						});					}				});			}		});		const stationModel = (this.stationModel = await DBModule.runJob("GET_MODEL", { modelName: "station" }));		const stationSchema = (this.stationSchema = await CacheModule.runJob("GET_SCHEMA", { schemaName: "station" }));		return new Promise((resolve, reject) =>			async.waterfall(				[					next => {						this.setStage(2);						CacheModule.runJob("HGETALL", { table: "stations" })							.then(stations => {								next(null, stations);							})							.catch(next);					},					(stations, next) => {						this.setStage(3);						if (!stations) return next();						const stationIds = Object.keys(stations);						return async.each(							stationIds,							(stationId, next) => {								stationModel.findOne({ _id: stationId }, (err, station) => {									if (err) next(err);									else if (!station) {										CacheModule.runJob("HDEL", {											table: "stations",											key: stationId										})											.then(() => {												next();											})											.catch(next);									} else next();								});							},							next						);					},					next => {						this.setStage(4);						stationModel.find({}, next);					},					(stations, next) => {						this.setStage(5);						async.each(							stations,							(station, next2) => {								async.waterfall(									[										next => {											CacheModule.runJob("HSET", {												table: "stations",												key: station._id,												value: stationSchema(station)											})												.then(station => next(null, station))												.catch(next);										},										(station, next) => {											StationsModule.runJob(												"INITIALIZE_STATION",												{													stationId: station._id												},												null,												-1											)												.then(() => {													next();												})												.catch(next);										}									],									err => {										next2(err);									}								);							},							next						);					}				],				async err => {					if (err) {						err = await UtilsModule.runJob("GET_ERROR", {							error: err						});						reject(new Error(err));					} else {						resolve();					}				}			)		);	}	/**	 * Initialises a station	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - id of the station to initialise	 * @returns {Promise} - returns a promise (resolve, reject)	 */	INITIALIZE_STATION(payload) {		return new Promise((resolve, reject) => {			// if (typeof cb !== 'function') cb = ()=>{};			async.waterfall(				[					next => {						StationsModule.runJob(							"GET_STATION",							{								stationId: payload.stationId							},							this						)							.then(station => {								next(null, station);							})							.catch(next);					},					(station, next) => {						if (!station) return next("Station not found.");						return NotificationsModule.runJob(							"UNSCHEDULE",							{								name: `stations.nextSong?id=${station._id}`							},							this						)							.then()							.catch()							.finally(() => {								NotificationsModule.runJob("SUBSCRIBE", {									name: `stations.nextSong?id=${station._id}`,									cb: () =>										StationsModule.runJob("SKIP_STATION", {											stationId: station._id										}),									unique: true,									station								})									.then()									.catch();								if (station.paused) return next(true, station);								return next(null, station);							});					},					(station, next) => {						if (!station.currentSong) {							return StationsModule.runJob(								"SKIP_STATION",								{									stationId: station._id								},								this							)								.then(station => {									next(true, station);								})								.catch(next)								.finally(() => {});						}						let timeLeft =							station.currentSong.duration * 1000 - (Date.now() - station.startedAt - station.timePaused);						if (Number.isNaN(timeLeft)) timeLeft = -1;						if (station.currentSong.duration * 1000 < timeLeft || timeLeft < 0) {							return StationsModule.runJob(								"SKIP_STATION",								{									stationId: station._id								},								this							)								.then(station => {									next(null, station);								})								.catch(next);						}						// name, time, cb, station						NotificationsModule.runJob("SCHEDULE", {							name: `stations.nextSong?id=${station._id}`,							time: timeLeft,							station						});						return next(null, station);					}				],				async (err, station) => {					if (err && err !== true) {						err = await UtilsModule.runJob(							"GET_ERROR",							{								error: err							},							this						);						reject(new Error(err));					} else resolve(station);				}			);		});	}	/**	 * Attempts to get the station from Redis. If it's not in Redis, get it from Mongo and add it to Redis.	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - id of the station	 * @returns {Promise} - returns a promise (resolve, reject)	 */	GET_STATION(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						CacheModule.runJob("HGET", { table: "stations", key: payload.stationId }, this)							.then(station => next(null, station))							.catch(next);					},					(station, next) => {						if (station) return next(true, station);						return StationsModule.stationModel.findOne({ _id: payload.stationId }, next);					},					(station, next) => {						if (station) {							station = StationsModule.stationSchema(station);							CacheModule.runJob("HSET", {								table: "stations",								key: payload.stationId,								value: station							})								.then()								.catch();							next(true, station);						} else next("Station not found");					}				],				async (err, station) => {					if (err && err !== true) {						err = await UtilsModule.runJob(							"GET_ERROR",							{								error: err							},							this						);						reject(new Error(err));					} else resolve(station);				}			);		});	}	/**	 * Attempts to get a station by name, firstly from Redis. If it's not in Redis, get it from Mongo and add it to Redis.	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationName - the unique name of the station	 * @returns {Promise} - returns a promise (resolve, reject)	 */	async GET_STATION_BY_NAME(payload) {		return new Promise((resolve, reject) =>			async.waterfall(				[					next => {						StationsModule.stationModel.findOne({ name: payload.stationName }, next);					},					(station, next) => {						if (station) {							station = StationsModule.stationSchema(station);							CacheModule.runJob("HSET", {								table: "stations",								key: station._id,								value: station							});							next(true, station);						} else next("Station not found");					}				],				(err, station) => {					if (err && err !== true) return reject(new Error(err));					return resolve(station);				}			)		);	}	/**	 * Updates the station in cache from mongo or deletes station in cache if no longer in mongo.	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - the id of the station to update	 * @returns {Promise} - returns a promise (resolve, reject)	 */	UPDATE_STATION(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						StationsModule.stationModel.findOne({ _id: payload.stationId }, next);					},					(station, next) => {						if (!station) {							CacheModule.runJob("HDEL", {								table: "stations",								key: payload.stationId							})								.then()								.catch();							return next("Station not found");						}						return CacheModule.runJob(							"HSET",							{								table: "stations",								key: payload.stationId,								value: station							},							this						)							.then(station => {								next(null, station);							})							.catch(next);					}				],				async (err, station) => {					if (err && err !== true) {						err = await UtilsModule.runJob(							"GET_ERROR",							{								error: err							},							this						);						reject(new Error(err));					} else resolve(station);				}			);		});	}	/**	 * Fills up the official station playlist queue using the songs from the official station playlist	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - the id of the station	 * @param {string} payload.ignoreExistingQueue - ignore the existing queue songs, replacing the old queue with a completely fresh one	 * @returns {Promise} - returns a promise (resolve, reject)	 */	FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST(payload) {		return new Promise((resolve, reject) => {			const { stationId, ignoreExistingQueue } = payload;			async.waterfall(				[					next => {						PlaylistsModule.runJob("GET_STATION_PLAYLIST", { stationId, includeSongs: true }, this)							.then(response => {								next(null, response.playlist);							})							.catch(next);					},					(playlist, next) => {						StationsModule.runJob("GET_STATION", { stationId }, this)							.then(station => {								if (ignoreExistingQueue) station.queue = [];								next(null, playlist, station);							})							.catch(next);					},					(playlist, station, next) => {						if (station.playMode === "random") {							UtilsModule.runJob("SHUFFLE", { array: playlist.songs }, this)								.then(response => {									next(null, response.array, station);								})								.catch(next);						} else next(null, playlist.songs, station);					},					(_playlistSongs, station, next) => {						let playlistSongs = JSON.parse(JSON.stringify(_playlistSongs));						if (station.playMode === "sequential") {							if (station.currentSongIndex <= playlistSongs.length) {								const songsToAddToEnd = playlistSongs.splice(0, station.currentSongIndex);								playlistSongs = [...playlistSongs, ...songsToAddToEnd];							}						}						const songsStillNeeded = 50 - station.queue.length;						const currentSongs = station.queue;						const currentYoutubeIds = station.queue.map(song => song.youtubeId);						const songsToAdd = [];						let lastSongAdded = null;						playlistSongs							// .map(song => song._doc)							.every(song => {								if (									songsToAdd.length < songsStillNeeded &&									currentYoutubeIds.indexOf(song.youtubeId) === -1								) {									lastSongAdded = song;									songsToAdd.push(song);									return true;								}								if (songsToAdd.length >= songsStillNeeded) return false;								return true;							});						let { currentSongIndex } = station;						if (station.playMode === "sequential" && lastSongAdded) {							const indexOfLastSong = _playlistSongs								.map(song => song.youtubeId)								.indexOf(lastSongAdded.youtubeId);							if (indexOfLastSong !== -1) currentSongIndex = indexOfLastSong;						}						const newPlaylist = [...currentSongs, ...songsToAdd].map(song => {							if (!song._id) song._id = null;							return song;						});						next(null, newPlaylist, currentSongIndex);					},					(newPlaylist, currentSongIndex, next) => {						StationsModule.stationModel.updateOne(							{ _id: stationId },							{ $set: { queue: newPlaylist, currentSongIndex } },							{ runValidators: true },							err => {								if (err) next(err);								else									StationsModule.runJob(										"UPDATE_STATION",										{											stationId										},										this									)										.then(() => {											next(null);										})										.catch(next);							}						);					}				],				err => {					if (err) reject(err);					else resolve();				}			);		});	}	/**	 * Gets next station song	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - the id of the station	 * @returns {Promise} - returns a promise (resolve, reject)	 */	GET_NEXT_STATION_SONG(payload) {		return new Promise((resolve, reject) => {			const { stationId } = payload;			async.waterfall(				[					next => {						StationsModule.runJob("GET_STATION", { stationId }, this)							.then(station => {								next(null, station);							})							.catch(next);					},					(station, next) => {						if (station.queue.length === 0) next("No songs available.");						else {							next(null, station.queue[0]);						}					},					(queueSong, next) => {						if (!queueSong._id) next(null, queueSong);						else							SongsModule.runJob("GET_SONG", { songId: queueSong._id }, this)								.then(response => {									const { song } = response;									if (song) {										const newSong = {											_id: song._id,											youtubeId: song.youtubeId,											title: song.title,											artists: song.artists,											duration: song.duration,											skipDuration: song.skipDuration,											thumbnail: song.thumbnail,											requestedAt: queueSong.requestedAt,											requestedBy: queueSong.requestedBy,											likes: song.likes,											dislikes: song.dislikes,											status: song.status										};										return next(null, newSong);									}									return next(null, song);								})								.catch(err => {									next(err);								});					}				],				(err, song) => {					if (err) console.log(33333, err, payload);					if (err) reject(err);					else resolve({ song });				}			);		});	}	/**	 * Removes first station queue song	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - the id of the station	 * @returns {Promise} - returns a promise (resolve, reject)	 */	REMOVE_FIRST_QUEUE_SONG(payload) {		return new Promise((resolve, reject) => {			const { stationId } = payload;			async.waterfall(				[					next => {						StationsModule.stationModel.updateOne(							{ _id: stationId },							{ $pop: { queue: -1 } },							{ runValidators: true },							err => {								if (err) next(err);								else									StationsModule.runJob(										"UPDATE_STATION",										{											stationId										},										this									)										.then(() => {											next(null);										})										.catch(next);							}						);					}				],				err => {					if (err) reject(err);					else resolve();				}			);		});	}	/**	 * Skips a station	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - the id of the station to skip	 * @returns {Promise} - returns a promise (resolve, reject)	 */	SKIP_STATION(payload) {		return new Promise((resolve, reject) => {			StationsModule.log("INFO", `Skipping station ${payload.stationId}.`);			StationsModule.log("STATION_ISSUE", `SKIP_STATION_CB - Station ID: ${payload.stationId}.`);			async.waterfall(				[					// Clears up any existing timers that would skip the station if the song ends					next => {						NotificationsModule.runJob("UNSCHEDULE", {							name: `stations.nextSong?id=${payload.stationId}`						})							.then(() => {								next();							})							.catch(next);					},					// Gets the station object					next => {						StationsModule.runJob(							"GET_STATION",							{								stationId: payload.stationId							},							this						)							.then(station => next(null, station))							.catch(next);					},					// eslint-disable-next-line consistent-return					(station, next) => {						if (!station) return next("Station not found.");						if (station.type === "community" && station.partyMode && station.queue.length === 0)							return next(null, null, station); // Community station with party mode enabled and no songs in the queue						if (station.type === "community" && station.partyMode && station.queue.length > 0) {							// Community station with party mode enabled and songs in the queue							if (station.paused) return next(null, null, station);							StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)								.then(response => {									StationsModule.runJob(										"REMOVE_FIRST_QUEUE_SONG",										{ stationId: station._id },										this									).then(() => {										next(null, response.song, station);									});								})								.catch(err => {									if (err === "No songs available.") next(null, null, station);									else next(err);								});							// return StationsModule.stationModel.updateOne(							// 	{ _id: payload.stationId },							// 	{							// 		$pull: {							// 			queue: {							// 				_id: station.queue[0]._id							// 			}							// 		}							// 	},							// 	err => {							// 		if (err) return next(err);							// 		return next(null, station.queue[0], -12, station);							// 	}							// );						}						if (station.type === "community" && !station.partyMode) {							StationsModule.runJob(								"FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST",								{ stationId: station._id },								this							)								.then(() => {									StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)										.then(response => {											StationsModule.runJob(												"REMOVE_FIRST_QUEUE_SONG",												{ stationId: station._id },												this											).then(() => {												next(null, response.song, station);											});										})										.catch(err => {											if (err === "No songs available.") next(null, null, station);											else next(err);										});								})								.catch(next);						}						// if (station.type === "community" && !station.partyMode) {						// 	return DBModule.runJob("GET_MODEL", { modelName: "playlist" }, this).then(playlistModel =>						// 		playlistModel.findOne({ _id: station.privatePlaylist }, (err, playlist) => {						// 			if (err) return next(err);						// 			if (!playlist) return next(null, null, -13, station);						// 			playlist = playlist.songs;						// 			if (playlist.length > 0) {						// 				let currentSongIndex;						// 				if (station.currentSongIndex < playlist.length - 1)						// 					currentSongIndex = station.currentSongIndex + 1;						// 				else currentSongIndex = 0;						// 				const callback = (err, song) => {						// 					if (err) return next(err);						// 					if (song) return next(null, song, currentSongIndex, station);						// 					const currentSong = {						// 						youtubeId: playlist[currentSongIndex].youtubeId,						// 						title: playlist[currentSongIndex].title,						// 						duration: playlist[currentSongIndex].duration,						// 						likes: -1,						// 						dislikes: -1,						// 						requestedAt: playlist[currentSongIndex].requestedAt						// 					};						// 					return next(null, currentSong, currentSongIndex, station);						// 				};						// 				if (mongoose.Types.ObjectId.isValid(playlist[currentSongIndex]._id))						// 					return SongsModule.runJob(						// 						"GET_SONG",						// 						{						// 							id: playlist[currentSongIndex]._id						// 						},						// 						this						// 					)						// 						.then(response => callback(null, response.song))						// 						.catch(callback);						// 				return SongsModule.runJob(						// 					"GET_SONG_FROM_YOUTUBE_ID",						// 					{						// 						youtubeId: playlist[currentSongIndex].youtubeId						// 					},						// 					this						// 				)						// 					.then(response => callback(null, response.song))						// 					.catch(callback);						// 			}						// 			return next(null, null, -14, station);						// 		})						// 	);						// }						if (station.type === "official") {							StationsModule.runJob(								"FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST",								{ stationId: station._id },								this							)								.then(() => {									StationsModule.runJob("GET_NEXT_STATION_SONG", { stationId: station._id }, this)										.then(response => {											StationsModule.runJob(												"REMOVE_FIRST_QUEUE_SONG",												{ stationId: station._id },												this											)												.then(() => {													next(null, response.song, station);												})												.catch(next);										})										.catch(err => {											if (err === "No songs available.") next(null, null, station);											else next(err);										});								})								.catch(next);						}					},					(song, station, next) => {						const $set = {};						if (song === null) $set.currentSong = null;						else {							$set.currentSong = {								_id: song._id,								youtubeId: song.youtubeId,								title: song.title,								artists: song.artists,								duration: song.duration,								likes: song.likes,								dislikes: song.dislikes,								skipDuration: song.skipDuration,								thumbnail: song.thumbnail,								requestedAt: song.requestedAt,								requestedBy: song.requestedBy,								status: song.status							};						}						// if (currentSongIndex >= 0) $set.currentSongIndex = currentSongIndex;						$set.startedAt = Date.now();						$set.timePaused = 0;						if (station.paused) $set.pausedAt = Date.now();						next(null, $set, station);					},					($set, station, next) => {						StationsModule.stationModel.updateOne({ _id: station._id }, { $set }, err => {							if (err) return next(err);							return StationsModule.runJob("UPDATE_STATION", { stationId: station._id }, this)								.then(station => {									CacheModule.runJob("PUB", {										channel: "station.queueUpdate",										value: payload.stationId									})										.then()										.catch();									next(null, station);								})								.catch(next);						});					}				],				async (err, station) => {					if (err) {						err = await UtilsModule.runJob(							"GET_ERROR",							{								error: err							},							this						);						StationsModule.log("ERROR", `Skipping station "${payload.stationId}" failed. "${err}"`);						reject(new Error(err));					} else {						if (station.currentSong !== null && station.currentSong.youtubeId !== undefined) {							station.currentSong.skipVotes = 0;						}						// TODO Pub/Sub this						WSModule.runJob("EMIT_TO_ROOM", {							room: `station.${station._id}`,							args: [								"event:songs.next",								{									data: {										currentSong: station.currentSong,										startedAt: station.startedAt,										paused: station.paused,										timePaused: 0									}								}							]						})							.then()							.catch();						if (station.privacy === "public") {							WSModule.runJob("EMIT_TO_ROOM", {								room: "home",								args: [									"event:station.nextSong",									{ data: { stationId: station._id, song: station.currentSong } }								]							})								.then()								.catch();						} else {							const sockets = await WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: "home" }, this);							sockets.forEach(async socketId => {								const socket = await WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this);								const { session } = socket;								if (session.sessionId) {									CacheModule.runJob(										"HGET",										{ table: "sessions", key: session.sessionId },										this										// eslint-disable-next-line no-loop-func									).then(session => {										if (session) {											DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(												userModel => {													userModel.findOne(														{															_id: session.userId														},														(err, user) => {															if (!err && user) {																if (user.role === "admin")																	socket.dispatch("event:station.nextSong", {																		data: {																			stationId: station._id,																			song: station.currentSong																		}																	});																else if (																	station.type === "community" &&																	station.owner === session.userId																)																	socket.dispatch("event:station.nextSong", {																		data: {																			stationId: station._id,																			song: station.currentSong																		}																	});															}														}													);												}											);										}									});								}							});						}						WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: `station.${station._id}` }).then(sockets => {							if (station.currentSong !== null && station.currentSong.youtubeId !== undefined) {								WSModule.runJob("SOCKETS_JOIN_SONG_ROOM", {									sockets,									room: `song.${station.currentSong.youtubeId}`								});								if (!station.paused) {									NotificationsModule.runJob("SCHEDULE", {										name: `stations.nextSong?id=${station._id}`,										time: station.currentSong.duration * 1000,										station									});								}							} else {								WSModule.runJob("SOCKETS_LEAVE_SONG_ROOMS", {									sockets								});							}						});						resolve({ station });					}				}			);		});	}	/**	 * Checks if a user can view/access a station	 *	 * @param {object} payload - object that contains the payload	 * @param {object} payload.station - the station object of the station in question	 * @param {string} payload.userId - the id of the user in question	 * @param {boolean} payload.hideUnlisted - whether the user is allowed to see unlisted stations or not	 * @returns {Promise} - returns a promise (resolve, reject)	 */	CAN_USER_VIEW_STATION(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						if (payload.station.privacy === "public") return next(true);						if (payload.station.privacy === "unlisted")							if (payload.hideUnlisted === true) return next();							else return next(true);						if (!payload.userId) return next("Not allowed");						return next();					},					next => {						DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(userModel => {							userModel.findOne({ _id: payload.userId }, next);						});					},					(user, next) => {						if (!user) return next("Not allowed");						if (user.role === "admin") return next(true);						if (payload.station.type === "official") return next("Not allowed");						if (payload.station.owner === payload.userId) return next(true);						return next("Not allowed");					}				],				async errOrResult => {					if (errOrResult !== true && errOrResult !== "Not allowed") {						errOrResult = await UtilsModule.runJob(							"GET_ERROR",							{								error: errOrResult							},							this						);						reject(new Error(errOrResult));					} else {						resolve(errOrResult === true);					}				}			);		});	}	/**	 * Checks if a user has favorited a station or not	 *	 * @param {object} payload - object that contains the payload	 * @param {object} payload.stationId - the id of the station in question	 * @param {string} payload.userId - the id of the user in question	 * @returns {Promise} - returns a promise (resolve, reject)	 */	HAS_USER_FAVORITED_STATION(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						DBModule.runJob("GET_MODEL", { modelName: "user" }, this).then(userModel => {							userModel.findOne({ _id: payload.userId }, next);						});					},					(user, next) => {						if (!user) return next("User not found.");						if (user.favoriteStations.indexOf(payload.stationId) !== -1) return next(null, true);						return next(null, false);					}				],				async (err, isStationFavorited) => {					if (err && err !== true) {						err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);						return reject(new Error(err));					}					return resolve(isStationFavorited);				}			);		});	}	/**	 * Returns a list of sockets in a room that can and can't know about a station	 *	 * @param {object} payload - the payload object	 * @param {object} payload.station - the station object	 * @param {string} payload.room - the websockets room to get the sockets from	 * @returns {Promise} - returns a promise (resolve, reject)	 */	GET_SOCKETS_THAT_CAN_KNOW_ABOUT_STATION(payload) {		return new Promise((resolve, reject) => {			WSModule.runJob("GET_SOCKETS_FOR_ROOM", { room: payload.room }, this)				.then(socketIds => {					const sockets = [];					async.eachLimit(						socketIds,						1,						(socketId, next) => {							WSModule.runJob("SOCKET_FROM_SOCKET_ID", { socketId }, this)								.then(socket => {									sockets.push(socket);									next();								})								.catch(err => {									reject(err);								});						},						err => {							if (err) reject(err);							else {								let socketsThatCan = [];								const socketsThatCannot = [];								if (payload.station.privacy === "public") {									socketsThatCan = sockets;									resolve({ socketsThatCan, socketsThatCannot });								} else {									async.eachLimit(										sockets,										1,										(socket, next) => {											const { session } = socket;											async.waterfall(												[													next => {														if (!session.sessionId) next("No session id");														else next();													},													next => {														CacheModule.runJob(															"HGET",															{																table: "sessions",																key: session.sessionId															},															this														)															.then(response => {																next(null, response);															})															.catch(next);													},													(session, next) => {														if (!session) next("No session");														else {															DBModule.runJob("GET_MODEL", { modelName: "user" }, this)																.then(userModel => {																	next(null, userModel);																})																.catch(next);														}													},													(userModel, next) => {														if (!userModel) next("No user model");														else															userModel.findOne(																{																	_id: session.userId																},																next															);													},													(user, next) => {														if (!user) next("No user found");														else if (user.role === "admin") {															socketsThatCan.push(socket);															next();														} else if (															payload.station.type === "community" &&															payload.station.owner === session.userId														) {															socketsThatCan.push(socket);															next();														}													}												],												err => {													if (err) socketsThatCannot.push(socket);													next();												}											);										},										err => {											if (err) reject(err);											else resolve({ socketsThatCan, socketsThatCannot });										}									);								}							}						}					);				})				.catch(reject);		});	}	/**	 * Adds a playlist to be included in a station	 *	 * @param {object} payload - object that contains the payload	 * @param {object} payload.stationId - the id of the station to include the playlist in	 * @param {object} payload.playlistId - the id of the playlist to be included	 * @returns {Promise} - returns a promise (resolve, reject)	 */	INCLUDE_PLAYLIST(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						if (!payload.stationId) next("Please specify a station id");						else if (!payload.playlistId) next("Please specify a playlist id");						else next();					},					next => {						StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)							.then(station => {								next(null, station);							})							.catch(next);					},					(station, next) => {						if (station.playlist === payload.playlistId) next("You cannot include the station playlist");						else if (station.includedPlaylists.indexOf(payload.playlistId) !== -1)							next("This playlist is already included");						else if (station.excludedPlaylists.indexOf(payload.playlistId) !== -1)							next(								"This playlist is currently excluded, please remove it from there before including it"							);						else							PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)								.then(() => {									next(null);								})								.catch(next);					},					next => {						DBModule.runJob(							"GET_MODEL",							{								modelName: "station"							},							this						).then(stationModel => {							stationModel.updateOne(								{ _id: payload.stationId },								{ $push: { includedPlaylists: payload.playlistId } },								next							);						});					},					(res, next) => {						StationsModule.runJob(							"UPDATE_STATION",							{								stationId: payload.stationId							},							this						)							.then(() => {								next();							})							.catch(next);					}				],				async err => {					if (err && err !== true) {						err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);						return reject(new Error(err));					}					return resolve();				}			);		});	}	/**	 * Removes a playlist that is included in a station	 *	 * @param {object} payload - object that contains the payload	 * @param {object} payload.stationId - the id of the station	 * @param {object} payload.playlistId - the id of the playlist	 * @returns {Promise} - returns a promise (resolve, reject)	 */	REMOVE_INCLUDED_PLAYLIST(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						if (!payload.stationId) next("Please specify a station id");						else if (!payload.playlistId) next("Please specify a playlist id");						else next();					},					next => {						StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)							.then(station => {								next(null, station);							})							.catch(next);					},					(station, next) => {						if (station.includedPlaylists.indexOf(payload.playlistId) === -1)							next("This playlist isn't included");						else next();					},					next => {						DBModule.runJob(							"GET_MODEL",							{								modelName: "station"							},							this						).then(stationModel => {							stationModel.updateOne(								{ _id: payload.stationId },								{ $pull: { includedPlaylists: payload.playlistId } },								next							);						});					},					(res, next) => {						StationsModule.runJob(							"UPDATE_STATION",							{								stationId: payload.stationId							},							this						)							.then(() => {								next();							})							.catch(next);					}				],				async err => {					if (err && err !== true) {						err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);						return reject(new Error(err));					}					return resolve();				}			);		});	}	/**	 * Adds a playlist to be excluded in a station	 *	 * @param {object} payload - object that contains the payload	 * @param {object} payload.stationId - the id of the station	 * @param {object} payload.playlistId - the id of the playlist	 * @returns {Promise} - returns a promise (resolve, reject)	 */	EXCLUDE_PLAYLIST(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						if (!payload.stationId) next("Please specify a station id");						else if (!payload.playlistId) next("Please specify a playlist id");						else next();					},					next => {						StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)							.then(station => {								next(null, station);							})							.catch(next);					},					(station, next) => {						if (station.playlist === payload.playlistId) next("You cannot exclude the station playlist");						else if (station.excludedPlaylists.indexOf(payload.playlistId) !== -1)							next("This playlist is already excluded");						else if (station.includedPlaylists.indexOf(payload.playlistId) !== -1)							next(								"This playlist is currently included, please remove it from there before excluding it"							);						else							PlaylistsModule.runJob("GET_PLAYLIST", { playlistId: payload.playlistId }, this)								.then(() => {									next(null);								})								.catch(next);					},					next => {						DBModule.runJob(							"GET_MODEL",							{								modelName: "station"							},							this						).then(stationModel => {							stationModel.updateOne(								{ _id: payload.stationId },								{ $push: { excludedPlaylists: payload.playlistId } },								next							);						});					},					(res, next) => {						StationsModule.runJob(							"UPDATE_STATION",							{								stationId: payload.stationId							},							this						)							.then(() => {								next();							})							.catch(next);					}				],				async err => {					if (err && err !== true) {						err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);						return reject(new Error(err));					}					return resolve();				}			);		});	}	/**	 * Removes a playlist that is excluded in a station	 *	 * @param {object} payload - object that contains the payload	 * @param {object} payload.stationId - the id of the station	 * @param {object} payload.playlistId - the id of the playlist	 * @returns {Promise} - returns a promise (resolve, reject)	 */	REMOVE_EXCLUDED_PLAYLIST(payload) {		return new Promise((resolve, reject) => {			console.log(112, payload);			async.waterfall(				[					next => {						if (!payload.stationId) next("Please specify a station id");						else if (!payload.playlistId) next("Please specify a playlist id");						else next();					},					next => {						StationsModule.runJob("GET_STATION", { stationId: payload.stationId }, this)							.then(station => {								next(null, station);							})							.catch(next);					},					(station, next) => {						if (station.excludedPlaylists.indexOf(payload.playlistId) === -1)							next("This playlist isn't excluded");						else next();					},					next => {						DBModule.runJob(							"GET_MODEL",							{								modelName: "station"							},							this						).then(stationModel => {							stationModel.updateOne(								{ _id: payload.stationId },								{ $pull: { excludedPlaylists: payload.playlistId } },								next							);						});					},					(res, next) => {						StationsModule.runJob(							"UPDATE_STATION",							{								stationId: payload.stationId							},							this						)							.then(() => {								next();							})							.catch(next);					}				],				async err => {					if (err && err !== true) {						err = await UtilsModule.runJob("GET_ERROR", { error: err }, this);						return reject(new Error(err));					}					return resolve();				}			);		});	}	/**	 * Gets stations that include or exclude a specific playlist	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.playlistId - the playlist id	 * @returns {Promise} - returns promise (reject, resolve)	 */	GET_STATIONS_THAT_INCLUDE_OR_EXCLUDE_PLAYLIST(payload) {		return new Promise((resolve, reject) => {			DBModule.runJob(				"GET_MODEL",				{					modelName: "station"				},				this			).then(stationModel => {				stationModel.find(					{						$or: [{ includedPlaylists: payload.playlistId }, { excludedPlaylists: payload.playlistId }]					},					(err, stations) => {						if (err) reject(err);						else resolve({ stationIds: stations.map(station => station._id) });					}				);			});		});	}	/**	 * Clears every queue	 *	 * @returns {Promise} - returns a promise (resolve, reject)	 */	CLEAR_EVERY_STATION_QUEUE() {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						StationsModule.stationModel.updateMany({}, { $set: { queue: [] } }, err => {							if (err) next(err);							else {								StationsModule.stationModel.find({}, (err, stations) => {									if (err) next(err);									else {										async.eachLimit(											stations,											1,											(station, next) => {												StationsModule.runJob("UPDATE_STATION", {													stationId: station._id												})													.then(() => next())													.catch(next);												CacheModule.runJob("PUB", {													channel: "station.queueUpdate",													value: station._id												})													.then()													.catch();											},											next										);									}								});							}						});					}				],				err => {					if (err) reject(err);					else resolve();				}			);		});	}	/**	 * Clears and refills a station queue	 *	 * @param {object} payload - object that contains the payload	 * @param {string} payload.stationId - the station id	 * @returns {Promise} - returns a promise (resolve, reject)	 */	CLEAR_AND_REFILL_STATION_QUEUE(payload) {		return new Promise((resolve, reject) => {			async.waterfall(				[					next => {						StationsModule.runJob(							"FILL_UP_STATION_QUEUE_FROM_STATION_PLAYLIST",							{ stationId: payload.stationId, ignoreExistingQueue: true },							this						)							.then(() => {								CacheModule.runJob("PUB", {									channel: "station.queueUpdate",									value: payload.stationId								})									.then()									.catch();								next();							})							.catch(err => {								next(err);							});					}				],				err => {					if (err) reject(err);					else resolve();				}			);		});	}}export default new _StationsModule();
 |