dxs
/
learn-request-queue
1
0
Fork 0
Code Issues 12 Pull Requests Projects Releases 11 Wiki Activity

Streamer-related fixes

master
Dessa Simpson 2020-09-19 11:24:39 -07:00
parent e9696fac25
commit 53da0eb687
3 changed files with 55 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
src/app.ts
View File

@ -257,17 +257,15 @@ app.get("/callback", async (request, response) => {
if (typeof tokenResponse == 'undefined') throw new Error('tokenResponse 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.tokenpair = { access_token: tokenResponse.access_token, refresh_token: tokenResponse.refresh_token };
request.session.user = (await twitch.apiRequest(request.session.tokenpair,"/users")).data[0]; request.session.user = (await twitch.apiRequest(request.session.tokenpair,"/users")).data[0];
var query = Object.assign(queries.updateUser,{ values: [request.session.user.id,request.session.user.display_name,request.session.user.profile_image_url] }); db.query(Object.assign(queries.updateUser,{ values: [request.session.user.id,request.session.user.display_name,request.session.user.profile_image_url] }));
db.query(query); var streamer = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]);
var streamerid = await db.query(queries.getStreamerId).then((result: pg.QueryResult) => result.rows[0]['userid']); if (typeof streamer == 'undefined' || request.session.user.id == streamer['userid']) {
if (typeof (tokenResponse as any).scope != 'undefined') { // Scopes requested - update streamer info if (typeof (tokenResponse as any).scope != 'undefined') { // Scopes requested - update streamer info
if (request.session.user.id == streamerid) { db.query(Object.assign(queries.updateStreamer,{ values: [request.session.user.id,JSON.stringify(request.session.tokenpair)] }));
var query = Object.assign(queries.updateStreamer,{ values: [request.session.user.id,JSON.stringify(request.session.tokenpair)] }); } else { // Scopes not requested - redirect and request them
db.query(query); 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;
} }
} else if (request.session.user.id == streamerid) {
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, '/'); response.redirect(307, '/');
}); });
@ -276,6 +274,7 @@ app.get("/callback", async (request, response) => {
app.get("/", async (request, response) => { app.get("/", async (request, response) => {
if (request.session) await validateApiToken(request.session); if (request.session) await validateApiToken(request.session);
var streamerInfo = await db.query(queries.getStreamerInfo).then((result: pg.QueryResult) => result.rows[0]); var streamerInfo = await db.query(queries.getStreamerInfo).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`);
if (!request.session || !request.session.user) { if (!request.session || !request.session.user) {
response.render('main.eta', { response.render('main.eta', {
loggedIn: false, loggedIn: false,
@ -306,25 +305,24 @@ app.get("/logout", async (request, response) => request.session!.destroy(() => r
async function processBannedUsers() { async function processBannedUsers() {
log(LogLevel.INFO,"processBannedUsers run at " + new Date().toISOString()); log(LogLevel.INFO,"processBannedUsers run at " + new Date().toISOString());
var streamer = await db.query(queries.getStreamerIdToken).then((result: pg.QueryResult) => result.rows[0]); var streamer = await db.query(queries.getStreamerIdToken).then((result: pg.QueryResult) => result.rows[0]);
if (streamer['tokenpair'] != null) { if (typeof streamer == 'undefined') return;
var response = await twitch.apiRequest(streamer['tokenpair'],"/moderation/banned?broadcaster_id=" + streamer['userid']); var response = await twitch.streamerApiRequest("/moderation/banned?broadcaster_id=" + streamer['userid']);
var dbconn = await db.connect(); var dbconn = await db.connect();
try { try {
await dbconn.query('BEGIN'); await dbconn.query('BEGIN');
for (var ban of response.data) { for (var ban of response.data) {
if (ban.expires_at == '') dbconn.query(Object.assign(queries.insertBan,{ values: [ban.user_id] })) if (ban.expires_at == '') dbconn.query(Object.assign(queries.insertBan,{ values: [ban.user_id] }))
}
await dbconn.query('COMMIT');
} catch (e) {
await dbconn.query('ROLLBACK');
} finally {
dbconn.release();
} }
await dbconn.query('COMMIT');
} catch (e) {
await dbconn.query('ROLLBACK');
} finally {
dbconn.release();
} }
setTimeout(processBannedUsers,3600000+Math.floor(Math.random()*900000)) // Run every 1-1.25 hours to balance load setTimeout(processBannedUsers,3600000+Math.floor(Math.random()*900000)) // Run every 1-1.25 hours to balance load
} }
processBannedUsers(); setTimeout(processBannedUsers,3600000+Math.floor(Math.random()*900000)) // Run every 1-1.25 hours to balance load
app.listen(config.port, () => { app.listen(config.port, () => {
console.log(`Listening on port ${config.port}`); console.log(`Listening on port ${config.port}`);

17
src/config.ts
View File

@ -1,29 +1,36 @@
if (typeof process.env.PORT == undefined) { import * as logging from './logging';
import { log, LogLevel } from "./logging"
if (typeof process.env.PORT == 'undefined') {
console.log("Missing environment variable PORT"); console.log("Missing environment variable PORT");
process.exit(1); process.exit(1);
} }
export const port: number = parseInt(process.env.PORT as string, 10); export const port: number = parseInt(process.env.PORT as string, 10);
if (typeof process.env.URL_PREFIX == undefined) { if (typeof process.env.URL_PREFIX == 'undefined') {
console.log("Missing environment variable URL_PREFIX"); console.log("Missing environment variable URL_PREFIX");
process.exit(1); process.exit(1);
} }
export const urlPrefix: string = process.env.URL_PREFIX!; export const urlPrefix: string = process.env.URL_PREFIX!;
if (typeof process.env.TWITCH_CLIENTID == undefined) { if (typeof process.env.TWITCH_CLIENTID == 'undefined') {
console.log("Missing environment variable TWITCH_CLIENTID"); console.log("Missing environment variable TWITCH_CLIENTID");
process.exit(1); process.exit(1);
} }
export const twitchClientId: string = process.env.TWITCH_CLIENTID!; export const twitchClientId: string = process.env.TWITCH_CLIENTID!;
if (typeof process.env.TWITCH_SECRET == undefined) { if (typeof process.env.TWITCH_SECRET == 'undefined') {
console.log("Missing environment variable TWITCH_SECRET"); console.log("Missing environment variable TWITCH_SECRET");
process.exit(1); process.exit(1);
} }
export const twitchSecret: string = process.env.TWITCH_SECRET!; export const twitchSecret: string = process.env.TWITCH_SECRET!;
if (typeof process.env.SESSION_SECRET == undefined) { if (typeof process.env.SESSION_SECRET == 'undefined') {
console.log("Missing environment variable SESSION_SECRET"); console.log("Missing environment variable SESSION_SECRET");
process.exit(1); process.exit(1);
} }
export const sessionSecret: string = process.env.SESSION_SECRET!; export const sessionSecret: string = process.env.SESSION_SECRET!;
export const logLevel = (process.env.LOG_LEVEL ? LogLevel[process.env.LOG_LEVEL as keyof typeof LogLevel] : LogLevel.ERROR );
console.log("Running with logLevel = " + logLevel)

23
src/twitch.ts
View File

@ -1,6 +1,9 @@
import * as config from "./config"; import * as config from "./config";
import * as queries from "./queries";
import { log, LogLevel } from "./logging" import { log, LogLevel } from "./logging"
import fetch, { Response as FetchResponse } from "node-fetch"; import fetch, { Response as FetchResponse } from "node-fetch";
import pg from "pg";
import db from "./db";
export interface TokenPair { export interface TokenPair {
access_token: string; access_token: string;
@ -44,11 +47,10 @@ export async function apiRequest(tokens: TokenPair, endpoint: string): Promise <
"Client-ID": config.twitchClientId "Client-ID": config.twitchClientId
}; };
return fetch("https://api.twitch.tv/helix" + endpoint, { headers: headers }) return fetch("https://api.twitch.tv/helix" + endpoint, { headers: headers })
.then((res: FetchResponse) => { .then(async (res: FetchResponse) => {
if (res.status == 200) { if (res.status == 200) {
return res.json(); return res.json();
} else { } else {
if (refreshApiToken(tokens)) {
log(LogLevel.WARNING,"Failed API request (pre-refresh):"); log(LogLevel.WARNING,"Failed API request (pre-refresh):");
log(LogLevel.WARNING,"Request URL: https://api.twitch.tv/helix" + endpoint); log(LogLevel.WARNING,"Request URL: https://api.twitch.tv/helix" + endpoint);
log(LogLevel.WARNING,"Headers:"); log(LogLevel.WARNING,"Headers:");
@ -56,6 +58,11 @@ export async function apiRequest(tokens: TokenPair, endpoint: string): Promise <
log(LogLevel.WARNING,"Response:"); log(LogLevel.WARNING,"Response:");
log(LogLevel.WARNING,JSON.stringify(await res.json(),null,2)); log(LogLevel.WARNING,JSON.stringify(await res.json(),null,2));
log(LogLevel.WARNING,"Attempting refresh"); log(LogLevel.WARNING,"Attempting refresh");
if (await refreshApiToken(tokens)) {
headers = {
"Authorization": "Bearer " + tokens.access_token,
"Client-ID": config.twitchClientId
};
return fetch("https://api.twitch.tv/helix" + endpoint, { headers: headers }) return fetch("https://api.twitch.tv/helix" + endpoint, { headers: headers })
.then(async (res: FetchResponse) => { .then(async (res: FetchResponse) => {
if (res.status == 200) { if (res.status == 200) {
@ -78,6 +85,18 @@ export async function apiRequest(tokens: TokenPair, endpoint: string): Promise <
}) })
} }
export async function streamerApiRequest(endpoint: string) {
log(LogLevel.DEBUG,`Call: streamerApiRequest(${endpoint})`);
var streamer = await db.query(queries.getStreamerIdToken).then((result: pg.QueryResult) => result.rows[0]);
var tokenpair = streamer.tokenpair;
var originaltoken = tokenpair.access_token;
log(LogLevel.DEBUG,"Original token: " + tokenpair.access_token);
var response = await apiRequest(tokenpair,endpoint);
log(LogLevel.DEBUG,"New token: " + tokenpair.access_token);
if (tokenpair.access_token != originaltoken) await db.query(Object.assign(queries.updateStreamer,{ values: [streamer.userid,tokenpair] }))
return response;
}
// Check if API token is valid. Checks Twitch's OAuth validation endpoint. If // Check if API token is valid. Checks Twitch's OAuth validation endpoint. If
// success, return true. If failure, return the result of attempting to refresh // success, return true. If failure, return the result of attempting to refresh
// the API token. // the API token.