learn-request-queue/src/app.ts

import * as config from "./config";
import * as requests from "./requests";
import * as twitch from "./twitch";
import * as queries from "./queries";
import { URLSearchParams } from "url";
import express from "express";
import session from "express-session";
import pg from "pg";
import pgSessionStore from "connect-pg-simple";
import fetch, { Response as FetchResponse } from "node-fetch";
import { log, LogLevel } from "./logging"
import cron from "./cron";
import db from "./db";
import errorHandler from "./errors";
import * as version from "./version";


console.log("Starting at " + new Date().toISOString());

// Ensure that any API token we have is valid - if not, destroy the session,
// logging out the user. Should be called before checking whether a user is
// logged in.
async function validateApiToken(session: Express.Session) {
	if (session.tokenpair && ! (await twitch.isApiTokenValid(session.tokenpair))) {
		session.destroy(()=>{});
	}
}

const app = express();
app.use(version.checkVersionMiddleware);
app.use(express.static('public'));
app.use(express.urlencoded({extended: false}));
app.use(express.json());
app.use(session({
	secret: config.sessionSecret,
	saveUninitialized: false,
	resave: false,
	store: new (pgSessionStore(session))({
		pool: db
	})
}));

// API
app.get("/api/getRequests", async (request, response) => {
	if (!request.session) throw new Error ("Missing request.session");
	await validateApiToken(request.session);
	var requestCount = ( request.query.count ? parseInt(request.query.count as string, 10) : 5 );
	var requestOffset = ( request.query.offset ? parseInt(request.query.offset as string, 10) : 0 );
	var sortDirection = ( request.query.sortDirection == "asc" ? "ASC" : "DESC" );
	var inverseSortDirection = ( sortDirection == "ASC" ? "DESC" : "ASC" );
	switch (request.query.sort) {
		case undefined: // Default sort by newest
		case "timestamp":
			var requestSort = `reqTimestamp ${sortDirection}`;
			break;
		case "score":
			var requestSort = `score ${sortDirection}, reqTimestamp ${inverseSortDirection}`;
			break;
		case "title":
			var requestSort = `title ${sortDirection}`
			break;
		case "requester":
			var requestSort = `requester ${sortDirection}, title ${sortDirection}`
			break;
		default:
			response.status(400);
			response.send("Invalid sort");
			return;
	};
	var requestsTotal = await requests.getRequestsTotal();
	if (request.session.user) {
		requests.getRequestsVoted(requestCount,requestOffset,requestSort,request.session.user.id)
			.then((val: Array<any>) => response.send({
					total: requestsTotal,
					requests: val
				}))
			.catch((e: any) => errorHandler(request,response,e));
	} else {
		requests.getRequests(requestCount,requestOffset,requestSort)
			.then((val: Array<any>) => response.send({
					total: requestsTotal,
					requests: val
				}))
			.catch((e: any) => errorHandler(request,response,e));
	}
});

app.get("/api/getAllRequests", async (request, response) => {
	if (!request.session) throw new Error ("Missing request.session");
	await validateApiToken(request.session);
	var requestCount = ( request.query.count ? parseInt(request.query.count as string, 10) : 5 );
	var requestOffset = ( request.query.offset ? parseInt(request.query.offset as string, 10) : 0 );
	var sortDirection = ( request.query.sortDirection == "asc" ? "ASC" : "DESC" );
	switch (request.query.sort) {
		case undefined:
		case "score":
			var requestSort = `score ${sortDirection}, reqTimestamp ASC`;
			break;
		case "timestamp":
			var requestSort = `reqTimestamp ${sortDirection}`;
			break;
		case "alpha":
			var requestSort = `title ${sortDirection}`
			break;
		default:
			response.status(400);
			response.send("Invalid sort");
			return;
	}
	var requestsTotal = await requests.getAllRequestsTotal();
	if (request.session.user) {
		requests.getAllRequestsVoted(requestCount,requestOffset,requestSort,request.session.user.id)
			.then((val: Array<any>) => response.send({
					total: requestsTotal,
					requests: val
				}))
			.catch((e: any) => errorHandler(request,response,e));
	} else {
		requests.getAllRequests(requestCount,requestOffset,requestSort)
			.then((val: Array<any>) => response.send({
					total: requestsTotal,
					requests: val
				}))
			.catch((e: any) => errorHandler(request,response,e));
	}
});

app.post("/api/addRequest", async (request, response) => {
	response.type('text/plain');
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var banned = await db.query(Object.assign(queries.checkBan, { values: [request.session.user.id] }))
		.then((result: pg.QueryResult) => result.rowCount > 0);
	if (banned) {
		response.status(401);
		response.send("You are banned; you may not add new requests.");
		return;
	}
	if (!request.body.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	var url = request.body.url as string;
	var requester = request.session.user.id;
	requests.addRequest(url,requester).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.post("/api/updateRequestState", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.body.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	if (!request.body.state) {
		response.status(400);
		response.send("Missing state");
		return;
	}
	var url = request.body.url as string;
	var state = request.body.state as string;
	response.type('text/plain');
	requests.updateRequestState(url,state).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.get("/api/updateRequestMetadata", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.query.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	requests.updateRequestMetadata(request.query.url as string).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.post("/api/updateRequestScoreModifier", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.body.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	if (!request.body.scoreDiff) {
		response.status(400);
		response.send("Missing scoreDiff");
		return;
	}
	var url = request.body.url as string;
	var scoreDiff = parseInt(request.body.scoreDiff as string, 10);
	response.type('text/plain');
	requests.updateRequestScoreModifier(url,scoreDiff).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.post("/api/updatePageTitle", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.body.pageTitle) {
		response.status(400);
		response.send("Missing pageTitle");
		return;
	}
	var pageTitle = request.body.pageTitle as string;
	response.type('text/plain');
	pageTitle = pageTitle.replace(/[&<>"']/g, function(m) {
    switch (m) {
      case '&':
        return '&amp;';
      case '<':
        return '&lt;';
      case '>':
        return '&gt;';
      case '"':
        return '&quot;';
      case "'":
        return '&apos;';
			default:
				return '';
    }
  });
	await db.query(Object.assign(queries.updatePageTitle,{ values: [pageTitle] }))
	.catch((e: any) => errorHandler(request,response,e));
	response.status(200);
	response.send('Successfully updated page title');
});

app.post("/api/updateColors", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.body.bg) {
		response.status(400);
		response.send("Missing bg");
		return;
	}
	if (!request.body.fg) {
		response.status(400);
		response.send("Missing fg");
		return;
	}
	type Colors = { [key: string]: { [key: string]: string} }
	var colors: Colors = { bg: {}, fg: {} };
	console.log(JSON.stringify(request.body,null,2));
	for (var color of ['primary','table','navbar','error']) {
		var setcolor = request.body.bg[color];
		if (/^#[0-9a-fA-F]{6}$/.test(setcolor)) {
			colors.bg[color] = setcolor
		} else {
			response.status(400);
			response.send(`Color 'bg.${color}' missing or invalid`)
			return;
		}
	}
	for (var color of ['primary','ahover','title']) {
		var setcolor = request.body.fg[color];
		if (/^#[0-9a-fA-F]{6}$/.test(setcolor)) {
			colors.fg[color] = setcolor
		} else {
			response.status(400);
			response.send(`Color 'fg.${color}' missing or invalid`)
			return;
		}
	}
	response.type('text/plain');
	await db.query(Object.assign(queries.updateColors,{ values: [JSON.stringify(colors)] }))
	.catch((e: any) => errorHandler(request,response,e));
	response.status(200);
	response.send('Successfully updated colors');
});

app.post("/api/updateVotePoints", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.body.user) {
		response.status(400);
		response.send("Missing user");
		return;
	}
	if (!request.body.follower) {
		response.status(400);
		response.send("Missing follower");
		return;
	}
	if (!request.body.subscriber) {
		response.status(400);
		response.send("Missing subscriber");
		return;
	}
	var user = request.body.user as number;
	var follower = request.body.follower as number;
	var subscriber = request.body.subscriber as number;
	response.type('text/plain');
	await db.query(Object.assign(queries.updateVotePoints,{ values: [user,follower,subscriber] }))
	.catch((e: any) => errorHandler(request,response,e));
	response.status(200);
	response.send('Successfully updated page title');
});

app.post("/api/deleteRequest", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.body.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	var url = request.body.url as string;
	response.type('text/plain');
	requests.deleteRequest(url).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.post("/api/addVote", async (request,response) => {
	response.type('text/plain');
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var banned = await db.query(Object.assign(queries.checkBan,{ values: [request.session.user.id] })).then((result: pg.QueryResult) => result.rowCount > 0);
	if (banned) {
		response.status(401);
		response.send("You are banned; you may not vote on requests.");
		return;
	}
	if (!request.body.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	var url = request.body.url as string;
	var user = request.session.user.id;
	requests.addVote(url,user).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.post("/api/deleteVote", async (request,response) => {
	response.type('text/plain');
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	if (!request.body.url) {
		response.status(400);
		response.send("Missing url");
		return;
	}
	var url = request.body.url as string;
	var user = request.session.user.id;
	requests.deleteVote(url,user).then((val: [number,string]) => {
		response.status(val[0]);
		response.send(val[1]);
	})
	.catch((e: any) => errorHandler(request,response,e));
});

app.get("/api/cronRequest", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	if (!request.session || !request.session.user) {
		response.status(401);
		response.send("Session expired; please log in again");
		return;
	}
	var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']);
	if (request.session.user.id != streamerid) {
		response.status(401);
		response.send("You are not the streamer");
		return;
	}
	if (!request.query.job) {
		response.status(400);
		response.send("Missing job");
		return;
	}
	var job = request.body.job as string;
	try {
		cron.validateJob(job)
	} catch (e) {
		response.status(400);
		response.send("Invalid job")
		return;
	}
	response.type('text/plain');
	cron.request(job).catch((e: any) => errorHandler(request,response,e));
});


// Twitch callback
app.get("/callback", async (request, response) => {
	if (request.query.error) {
		response.redirect(307, '/');
		return;
	}
	var authcode = request.query.code as string;
	var tokenResponse = await fetch("https://id.twitch.tv/oauth2/token", { method: "POST", body: new URLSearchParams({
		client_id: config.twitchClientId,
		client_secret: config.twitchSecret,
		code: authcode,
		grant_type: "authorization_code",
		redirect_uri: `${config.urlPrefix}/callback`
	})}).then((res: FetchResponse) => res.json() as Promise<twitch.TokenPair>)
	.catch((e: any) => errorHandler(request,response,e));
	if (typeof request.session == 'undefined') throw new Error('Session is undefined');
	if (typeof tokenResponse == 'undefined') throw new Error('tokenResponse is undefined');
	request.session.tokenpair = { access_token: tokenResponse.access_token, refresh_token: tokenResponse.refresh_token };
	request.session.user = (await twitch.apiRequest(request.session.tokenpair,"/users")).data[0];
	await db.query(Object.assign(queries.updateUser,{ values: [request.session.user.id,request.session.user.display_name,request.session.user.profile_image_url] }));
	var streamer = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]);
	if (typeof streamer == 'undefined' || request.session.user.id == streamer['userid']) {
		if (typeof (tokenResponse as any).scope != 'undefined') { // Scopes requested - update streamer info
			db.query(Object.assign(queries.updateStreamer,{ values: [request.session.user.id,JSON.stringify(request.session.tokenpair)] }));
		} else { // Scopes not requested - redirect and request them
			response.redirect(307, `https://id.twitch.tv/oauth2/authorize?client_id=${config.twitchClientId}&redirect_uri=${config.urlPrefix}/callback&response_type=code&scope=channel:read:subscriptions moderation:read`);
			return;
		}
	}
	response.redirect(307, '/');
});

// Frontend templates
app.get("/", async (request, response) => {
	if (request.session) await validateApiToken(request.session);
	var streamerInfo = await db.query(queries.getStreamerInfo).then((result: pg.QueryResult) => result.rows[0]);
	var streamerConfig = await db.query(queries.getConfig).then((result: pg.QueryResult) => result.rows[0]);
	if (typeof streamerInfo == 'undefined') {
		response.redirect(307, `https://id.twitch.tv/oauth2/authorize?client_id=${config.twitchClientId}&redirect_uri=${config.urlPrefix}/callback&response_type=code&scope=channel:read:subscriptions moderation:read`);
		return;
	}
	if (!request.session || !request.session.user) {
		response.render('main.eta', {
			loggedIn: false,
			clientId: config.twitchClientId,
			urlPrefix: config.urlPrefix,
			streamerName: streamerInfo['displayname'],
			streamerProfilePicture: streamerInfo['imageurl'],
			config: streamerConfig,
		});
	} else {
		var validStates = JSON.stringify((await db.query(queries.getValidStates).then((result: pg.QueryResult) => result.rows)).map((row: any) => row.state));
		response.render('main.eta', {
			loggedIn: true,
			userName: request.session.user.display_name,
			userProfilePicture: request.session.user.profile_image_url,
			validStates: validStates,
			isStreamer: streamerInfo['userid'] == request.session.user.id,
			streamerName: streamerInfo['displayname'],
			streamerProfilePicture: streamerInfo['imageurl'],
			config: streamerConfig,
		});
	}
});

app.get("/colors.css", async (_request, response) => {
	var streamerInfo = await db.query(queries.getStreamerInfo).then((result: pg.QueryResult) => result.rows[0]);
	var colors = await db.query(queries.getConfig).then((result: pg.QueryResult) => result.rows[0]['colors']);
	if (typeof streamerInfo == 'undefined') return;
	response.contentType("text/css");
	response.render('colors.eta', colors);
});

// Streamer Panel
//app.get("/streamer/", async (request, response) => {
//
//});

// Logout
app.get("/logout", async (request, response) =>
  request.session!.destroy(() => response.redirect(307, '/')));

// Check version then listen
version.checkVersion().then(_ => app.listen(config.port, () => {
	cron.run();
	setInterval(cron.run,config.cronInterval);
  console.log(`Listening on port ${config.port}`);
}))
.catch((e) => {
	log(LogLevel.ERROR,e)
	process.exit(1);
});