MusareFork/backend/logic/actions/stations.js

'use strict';

const async   = require('async'),
	  request = require('request'),
	  config  = require('config');

const io = require('../io');
const db = require('../db');
const cache = require('../cache');
const notifications = require('../notifications');
const utils = require('../utils');

let stationsLoaded = {};

notifications.subscribe('station.locked', function(stationName) {
	io.to(`station.${stationName}`).emit("event:station.locked");
});

notifications.subscribe('station.unlocked', function(stationName) {
	io.to(`station.${stationName}`).emit("event:station.unlocked");
});

notifications.subscribe('station.pause', function(stationName) {
	io.to(`station.${stationName}`).emit("event:station.pause");
});

notifications.subscribe('station.resume', function(stationName) {
	io.to(`station.${stationName}`).emit("event:station.resume");
});

/**
 * Loads a station into the cache, and sets up all the related logic
 *
 * @param {String} stationId - the id of the station
 * @param {Function} cb - gets called when this function completes
 */
function initializeAndReturnStation (stationId, cb) {
	async.waterfall([

		// first check the cache for the station
		(next) => cache.hget('stations', stationId, next),

		// if the cached version exist
		(station, next) => {
			if (station) return next(true, station);
			db.models.station.findOne({ id: stationId }, next);
		},

		// if the station exists in the DB, add it to the cache
		(station, next) => {
			if (!station) return cb('Station by that id does not exist');
			station = cache.schemas.station(station);
			cache.hset('stations', station.id, station, (err) => next(err, station));
		}

	], (err, station) => {
		if (err && err !== true) return cb(err);

		// get notified when the next song for this station should play, so that we can notify our sockets
		if (stationsLoaded[stationId] === undefined) {
			stationsLoaded[stationId] = 1;
			let notification = notifications.subscribe(`stations.nextSong?id=${station.id}`, () => {
				// get the station from the cache
				cache.hget('stations', station.name, (err, station) => {
					if (station) {
						console.log(777);
						// notify all the sockets on this station to go to the next song
						io.to(`station.${stationId}`).emit("event:songs.next", {
							currentSong: station.currentSong,
							startedAt: station.startedAt,
							paused: station.paused,
							timePaused: 0
						});
						// schedule a notification to be dispatched when the next song ends
						notifications.schedule(`stations.nextSong?id=${station.id}`, station.currentSong.duration * 1000);
					}
					// the station doesn't exist anymore, unsubscribe from it
					else {
						console.log(888);
						notifications.remove(notification);
						delete stationsLoaded[stationId];
					}
				});
			}, true);

			if (!station.paused) {
				console.log(station);
				notifications.schedule(`stations.nextSong?id=${station.id}`, station.currentSong.duration * 1000);
			}
		}

		return cb(null, station);

		// will need to be added once station namespace thing is decided
		// function generatePlaylist(arr) {
		// 	station.playlist = [];
		// 	return arr.reduce((promise, id) => {
		// 		return promise.then(() => {
		// 			return globals.db.models.song.findOne({ id }, (err, song) => {
		// 				if (err) throw err;
		// 				station.playlist.push(song);
		// 			});
		// 		});
		// 	}, Promise.resolve());
		// }

		// generatePlaylist(station.playlist).then(() => {
		// 	cb(null, station);
		// });
	});
}

module.exports = {

	/**
	 * Get a list of all the stations
	 *
	 * @param session
	 * @param cb
	 * @return {{ status: String, stations: Array }}
	 */
	index: (session, cb) => {
		// TODO: the logic should be a bit more personalized to the users preferred genres
		// and it should probably just a different cache table then 'stations'
		cache.hgetall('stations', (err, stations) => {

			if (err && err !== true) {
				return cb({
					status: 'error',
					message: 'An error occurred while obtaining the stations'
				});
			}

			let arr = [];
			for (let prop in stations) {
				arr.push(stations[prop]);
			}

			cb({ status: 'success', stations: arr });
		});
	},

	/**
	 * Joins the station by its id
	 *
	 * @param session
	 * @param stationId - the station id
	 * @param cb
	 * @return {{ status: String, userCount: Integer }}
	 */
	join: (session, stationId, cb) => {
		let ns = io.io.of("/");
		if (ns) {
			for (let id in ns.connected) {
				console.log(ns.connected[id]);
				console.log(ns.connected[id].testProp);
			}
		}


		initializeAndReturnStation(stationId, (err, station) => {

			if (err && err !== true) {
				return cb({ status: 'error', message: 'An error occurred while joining the station' });
			}

			if (station) {

				if (session) session.stationId = stationId;

				//TODO Loop through all sockets, see if socket with same session exists, and if so leave all other station rooms and join this stationRoom

				cache.client.hincrby('station.userCounts', stationId, 1, (err, userCount) => {
					if (err) return cb({ status: 'error', message: 'An error occurred while joining the station' });
					utils.socketJoinRoom(session);
					//TODO Emit to cache, listen on cache
					cb({ status: 'success', currentSong: station.currentSong, startedAt: station.startedAt, paused: station.paused, timePaused: station.timePaused });
				});
			}
			else {
				cb({ status: 'failure', message: `That station doesn't exist` });
			}
		});
	},

	/**
	 * Skips the users current station
	 *
	 * @param session
	 * @param stationId - the station id
	 * @param cb
	 * @return {{ status: String, skipCount: Integer }}
	 */
	skip: (session, stationId, cb) => {

		if (!session) return cb({ status: 'failure', message: 'You must be logged in to skip a song!' });

		initializeAndReturnStation(stationId, (err, station) => {

			if (err && err !== true) {
				return cb({ status: 'error', message: 'An error occurred while skipping the station' });
			}

			if (station) {
				cache.client.hincrby('station.skipCounts', session.stationId, 1, (err, skipCount) => {

					session.skippedSong = true;

					if (err) return cb({ status: 'error', message: 'An error occurred while skipping the station' });

					cache.hget('station.userCounts', session.stationId, (err, userCount) => {
						cb({ status: 'success', skipCount });
					});
				});
			}
			else {
				cb({ status: 'failure', message: `That station doesn't exist` });
			}
		});
	},

	/**
	 * Leaves the users current station
	 *
	 * @param session
	 * @param cb
	 * @return {{ status: String, userCount: Integer }}
	 */
	leave: (session, cb) => {
		let stationId = "edm";
		initializeAndReturnStation(stationId, (err, station) => {

			if (err && err !== true) {
				return cb({ status: 'error', message: 'An error occurred while leaving the station' });
			}

			if (session) session.stationId = null;
			else if (station) {
				cache.client.hincrby('station.userCounts', stationId, -1, (err, userCount) => {
					if (err) return cb({ status: 'error', message: 'An error occurred while leaving the station' });
					utils.socketLeaveRooms(session);
					cb({ status: 'success', userCount });
				});
			} else {
				cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
			}
		});
	},

	lock: (session, stationId, cb) => {
		//TODO Require admin
		initializeAndReturnStation(stationId, (err, station) => {
			if (err && err !== true) {
				return cb({ status: 'error', message: 'An error occurred while locking the station' });
			} else if (station) {
				// Add code to update Mongo and Redis
				cb({ status: 'success' });
			} else {
				cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
			}
		});
	},

	unlock: (session, stationId, cb) => {
		//TODO Require admin
		initializeAndReturnStation(stationId, (err, station) => {
			if (err && err !== true) {
				return cb({ status: 'error', message: 'An error occurred while unlocking the station' });
			} else if (station) {
				// Add code to update Mongo and Redis
				cb({ status: 'success' });
			} else {
				cb({ status: 'failure', message: `That station doesn't exist, it may have been deleted` });
			}
		});
	},

};