From e05d519e5a7a51673ca83fcd926c5ebf9e606d10 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 7 Nov 2021 14:07:01 -0500 Subject: [PATCH] NEW OBS GATEWAY ARC (NOT YET WOKRING) --- Cargo.lock | 77 ++++++- Cargo.toml | 9 + .../down.sql | 4 + .../up.sql | 75 +++++++ src/http_server/endpoints/mod.rs | 1 + src/http_server/endpoints/twitch/mod.rs | 1 + .../endpoints/twitch/pubsub_gateway.rs | 86 ++++++++ src/obs_gateway/mod.rs | 0 src/obs_gateway_server.rs | 200 ++++++++++++++++++ test.js | 10 + 10 files changed, 457 insertions(+), 6 deletions(-) create mode 100644 migrations/2021-09-29-042235_twitch_oauth_tokens/down.sql create mode 100644 migrations/2021-09-29-042235_twitch_oauth_tokens/up.sql create mode 100644 src/http_server/endpoints/twitch/mod.rs create mode 100644 src/http_server/endpoints/twitch/pubsub_gateway.rs create mode 100644 src/obs_gateway/mod.rs create mode 100644 src/obs_gateway_server.rs create mode 100644 test.js diff --git a/Cargo.lock b/Cargo.lock index 06a1e3fcd7..b194588ab9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,31 @@ # It is not intended for manual editing. version = 3 +[[package]] +name = "actix" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1be241f88f3b1e7e9a3fbe3b5a8a0f6915b5a1d7ee0d9a248d3376d01068cc60" +dependencies = [ + "actix-rt 1.1.1", + "actix_derive", + "bitflags", + "bytes 0.5.6", + "crossbeam-channel 0.4.4", + "derive_more", + "futures-channel", + "futures-util", + "log", + "once_cell", + "parking_lot 0.11.1", + "pin-project 0.4.28", + "smallvec 1.6.1", + "tokio 0.2.25", + "tokio-util", + "trust-dns-proto", + "trust-dns-resolver", +] + [[package]] name = "actix-codec" version = "0.3.0" @@ -130,9 +155,9 @@ dependencies = [ [[package]] name = "actix-macros" -version = "0.2.0" +version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dbcb2b608f0accc2f5bcf3dd872194ce13d94ee45b571487035864cf966b04ef" +checksum = "c2f86cd6857c135e6e9fe57b1619a88d1f94a7df34c00e11fe13e64fd3438837" dependencies = [ "quote", "syn", @@ -190,7 +215,7 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bc7d7cd957c9ed92288a7c3c96af81fa5291f65247a76a34dac7b6af74e52ba0" dependencies = [ - "actix-macros 0.2.0", + "actix-macros 0.2.1", "futures-core", "tokio 1.6.0", ] @@ -335,6 +360,22 @@ dependencies = [ "url 2.2.2", ] +[[package]] +name = "actix-web-actors" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f6edf3c2693e2a8c422800c87ee89a6a4eac7dd01109bc172a1093ce1f4f001" +dependencies = [ + "actix", + "actix-codec", + "actix-http", + "actix-web", + "bytes 0.5.6", + "futures-channel", + "futures-core", + "pin-project 0.4.28", +] + [[package]] name = "actix-web-codegen" version = "0.4.0" @@ -346,6 +387,17 @@ dependencies = [ "syn", ] +[[package]] +name = "actix_derive" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b95aceadaf327f18f0df5962fedc1bde2f870566a0b9f65c89508a3b1f79334c" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "adler" version = "1.0.2" @@ -1024,6 +1076,16 @@ dependencies = [ "cfg-if 1.0.0", ] +[[package]] +name = "crossbeam-channel" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b153fe7cbef478c567df0f972e02e6d736db11affe43dfc9c56a9374d1adfb87" +dependencies = [ + "crossbeam-utils 0.7.2", + "maybe-uninit", +] + [[package]] name = "crossbeam-channel" version = "0.5.1" @@ -1655,9 +1717,9 @@ dependencies = [ [[package]] name = "httparse" -version = "1.4.1" +version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3a87b616e37e93c22fb19bcd386f02f3af5ea98a25670ad0fce773de23c5e68" +checksum = "acd94fdbe1d4ff688b67b04eee2e17bd50995534a61539e45adfefb45e5e5503" [[package]] name = "httpdate" @@ -3209,7 +3271,7 @@ dependencies = [ "bytes 0.5.6", "chrono", "crc", - "crossbeam-channel", + "crossbeam-channel 0.5.1", "crossbeam-queue", "crossbeam-utils 0.8.4", "digest", @@ -3346,6 +3408,7 @@ checksum = "213701ba3370744dcd1a12960caa4843b3d68b4d1c0a5d575e0d65b2ee9d16c0" name = "storyteller-web" version = "0.0.1" dependencies = [ + "actix", "actix-cors", "actix-files", "actix-http", @@ -3354,6 +3417,7 @@ dependencies = [ "actix-service", "actix-utils 3.0.0", "actix-web", + "actix-web-actors", "anyhow", "base64 0.13.0", "bcrypt", @@ -3761,6 +3825,7 @@ checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499" dependencies = [ "bytes 0.5.6", "futures-core", + "futures-io", "futures-sink", "log", "pin-project-lite 0.1.12", diff --git a/Cargo.toml b/Cargo.toml index 30f67728b8..b46d2161a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,11 @@ publish = false name = "storyteller-web" path = "src/main.rs" +# OBS websocket server +[[bin]] +name = "obs-gateway-server" +path = "src/obs_gateway_server.rs" + # Jobs for managing downloads and inference [[bin]] name = "tts-download-job" @@ -89,6 +94,10 @@ subprocess = "0.2.6" tempdir = "0.3.7" # Web +actix = "0.10" +#actix-web = "3" +actix-web-actors = "3" + actix-cors = "0.5.4" actix-files= "0.5" actix-http = "2.2.0" diff --git a/migrations/2021-09-29-042235_twitch_oauth_tokens/down.sql b/migrations/2021-09-29-042235_twitch_oauth_tokens/down.sql new file mode 100644 index 0000000000..fdbf0af1b8 --- /dev/null +++ b/migrations/2021-09-29-042235_twitch_oauth_tokens/down.sql @@ -0,0 +1,4 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + diff --git a/migrations/2021-09-29-042235_twitch_oauth_tokens/up.sql b/migrations/2021-09-29-042235_twitch_oauth_tokens/up.sql new file mode 100644 index 0000000000..eed73ee95b --- /dev/null +++ b/migrations/2021-09-29-042235_twitch_oauth_tokens/up.sql @@ -0,0 +1,75 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +-- **NOTE ABOUT DESIGN**: +-- This table contains _many_ OAuth tokens for any single user. (many users : many tokens) +-- An entire history of tokens is kept in the table. + +CREATE TABLE twitch_oauth_tokens( + -- Not used for anything except replication. + id BIGINT(20) NOT NULL AUTO_INCREMENT, + + -- Foreign key to user (Vocodes/Storyteller user) + -- If no user is associated (anonymous Twitch user), then this is null. + maybe_user_token VARCHAR(32) DEFAULT NULL, + + -- The user ID / channel ID are the same + -- https://discuss.dev.twitch.tv/t/what-is-the-difference-between-the-stream--id-and-channel--id/4423 + maybe_twitch_user_id VARCHAR(32) DEFAULT NULL, + + -- Usernames are between 4 and 25 characters. + maybe_twitch_username VARCHAR(32) DEFAULT NULL, + + -- ========== OAUTH SCOPES ========== + + -- "bits:read" + -- "View Bits information for a channel." + has_bits_read BOOLEAN NOT NULL DEFAULT FALSE, + + -- "channel:read:subscriptions" + -- "View a list of all subscribers to a channel and check if a user is subscribed to a channel." + -- eg. Enumerate list of subscribers for a channel + -- eg. PubSub subscribe to `channel-subscribe-events-v1.` (subscribe, resubscribe, gift) + has_channel_read_subscriptions BOOLEAN NOT NULL DEFAULT FALSE, + + -- "channel:read:redemptions" + -- "View Channel Points custom rewards and their redemptions on a channel." + -- eg. PubSub subscribe to `channel-points-channel-v1.` (channel points spends) + has_channel_read_redemptions BOOLEAN NOT NULL DEFAULT FALSE, + + -- "user:read:follows" + -- "View the list of channels a user follows." + -- eg. We can use this to see if a Twitch user follows us. + has_user_read_follows BOOLEAN NOT NULL DEFAULT FALSE, + + + -- THERE ARE MORE SCOPES... PERHAPS WE SHOULD ADD THEM? + + -- ========== VECTOR CLOCK ========== + + -- Incremented with every update. + version INT NOT NULL DEFAULT 0, + + -- ========== TIMESTAMPS ========== + + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + + -- This is when the OAuth token will expire. We need to refresh before this time. + expires_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + + -- If the OAuth token is deleted, we set this. + -- We'll need to set it for all of a user's OAuth tokens since this table stores many tokens. + user_deleted_at TIMESTAMP NULL, + mod_deleted_at TIMESTAMP NULL, + + -- INDICES -- + PRIMARY KEY (id), + KEY fk_maybe_user_token (maybe_user_token) + + --KEY index_has_bits_read (has_bits_read), + --KEY index_creator_ip_address (creator_ip_address) + +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; + diff --git a/src/http_server/endpoints/mod.rs b/src/http_server/endpoints/mod.rs index f585acfc72..38b3143d3f 100644 --- a/src/http_server/endpoints/mod.rs +++ b/src/http_server/endpoints/mod.rs @@ -5,5 +5,6 @@ pub mod misc; pub mod moderation; pub mod root_index; pub mod tts; +pub mod twitch; pub mod users; pub mod w2l; \ No newline at end of file diff --git a/src/http_server/endpoints/twitch/mod.rs b/src/http_server/endpoints/twitch/mod.rs new file mode 100644 index 0000000000..f1d932d175 --- /dev/null +++ b/src/http_server/endpoints/twitch/mod.rs @@ -0,0 +1 @@ +pub mod pubsub_gateway; \ No newline at end of file diff --git a/src/http_server/endpoints/twitch/pubsub_gateway.rs b/src/http_server/endpoints/twitch/pubsub_gateway.rs new file mode 100644 index 0000000000..96f49d96ec --- /dev/null +++ b/src/http_server/endpoints/twitch/pubsub_gateway.rs @@ -0,0 +1,86 @@ +use actix::prelude::*; +use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer}; +use actix_web_actors::ws; +use log::warn; +use log::error; + +pub async fn ws_index(r: HttpRequest, stream: web::Payload) -> Result { + error!(">>>>>> ws_index"); + ws::start(WebSocket::new(), &r, stream) +} + +struct WebSocket {} + +impl Actor for WebSocket { + type Context = ws::WebsocketContext; + + fn started(&mut self, _ctx: &mut Self::Context) { + error!(">>>>>> actor started"); + } +} + +impl StreamHandler> for WebSocket { + fn handle( + &mut self, + msg: Result, + ctx: &mut Self::Context, + ) { + if let Ok(msg) = msg { + error!(">>>>>> streamhandler::handle (msg = {:?})", msg); + match msg { + ws::Message::Text(text) => ctx.text(text), + ws::Message::Binary(bin) => ctx.binary(bin), + ws::Message::Ping(bytes) => ctx.pong(&bytes), + ws::Message::Close(reason) => { + ctx.close(reason); + ctx.stop(); + } + _ => {} + } + } else { + error!(">>>>>> streamhandler::STOP"); + ctx.stop(); + } + } +} + +impl WebSocket { + fn new() -> Self { + Self {} + } +} + +// From the Actix homepage: + +//use actix::{Actor, StreamHandler}; +//use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer}; +//use actix_web_actors::ws; +// +///// Define HTTP actor +//struct MyWs; +// +//impl Actor for MyWs { +// type Context = ws::WebsocketContext; +//} +// +///// Handler for ws::Message message +//impl StreamHandler> for MyWs { +// fn handle( +// &mut self, +// msg: Result, +// ctx: &mut Self::Context, +// ) { +// match msg { +// Ok(ws::Message::Ping(msg)) => ctx.pong(&msg), +// Ok(ws::Message::Text(text)) => ctx.text(text), +// Ok(ws::Message::Binary(bin)) => ctx.binary(bin), +// _ => (), +// } +// } +//} +// +//pub async fn twitch_pubsub_gateway(req: HttpRequest, stream: web::Payload) -> Result { +// let resp = ws::start(MyWs {}, &req, stream); +// println!("{:?}", resp); +// resp +//} \ No newline at end of file diff --git a/src/obs_gateway/mod.rs b/src/obs_gateway/mod.rs new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/obs_gateway_server.rs b/src/obs_gateway_server.rs new file mode 100644 index 0000000000..114f8c4dcf --- /dev/null +++ b/src/obs_gateway_server.rs @@ -0,0 +1,200 @@ +#![forbid(private_in_public)] +#![forbid(unused_must_use)] +//#![forbid(warnings)] + +#![allow(dead_code)] +#![allow(non_snake_case)] +#![allow(unused_imports)] +#![allow(unused_mut)] +#![allow(unused_variables)] + +#[macro_use] extern crate lazy_static; +#[macro_use] extern crate magic_crypt; +#[macro_use] extern crate serde_derive; + +pub mod shared_constants; + +use actix_cors::Cors; +use actix_http::http; +use actix_web::middleware::{Logger, DefaultHeaders}; +use actix_web::{HttpServer, web, HttpResponse, App}; +use crate::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING; +use crate::shared_constants::DEFAULT_REDIS_CONNECTION_STRING; +use crate::shared_constants::DEFAULT_RUST_LOG; +use futures::Future; +use futures::executor::ThreadPool; +use limitation::Limiter; +use log::{info}; +use r2d2_redis::RedisConnectionManager; +use r2d2_redis::r2d2; +use r2d2_redis::redis::Commands; +use sqlx::MySqlPool; +use sqlx::mysql::MySqlPoolOptions; +use std::sync::Arc; +use std::time::Duration; + +const DEFAULT_BIND_ADDRESS : &'static str = "0.0.0.0:12345"; + +pub type AnyhowResult = anyhow::Result; + +/// State that is injected into every endpoint. +#[derive(Clone)] +pub struct ObsGatewayServerState { + /// Configuration from ENV vars. + /// Some of this might not be used. + pub env_config: EnvConfig, + pub hostname: String, +} + +#[derive(Clone)] +pub struct EnvConfig { + // Number of thread workers. + pub num_workers: usize, + pub bind_address: String, + pub cookie_domain: String, + pub cookie_secure: bool, + pub cookie_http_only: bool, + pub website_homepage_redirect: String, +} + +#[actix_web::main] +async fn main() -> AnyhowResult<()> { + easyenv::init_all_with_default_logging(Some(DEFAULT_RUST_LOG)); + + // NB: Do not check this secrets-containing dotenv file into VCS. + // This file should only contain *development* secrets, never production. + let _ = dotenv::from_filename(".env-secrets").ok(); + + info!("Obtaining hostname..."); + + let server_hostname = hostname::get() + .ok() + .and_then(|h| h.into_string().ok()) + .unwrap_or("obs-gateway-server-unknown".to_string()); + + info!("Hostname: {}", &server_hostname); + + info!("Connecting to database..."); + + let db_connection_string = + easyenv::get_env_string_or_default( + "MYSQL_URL", + DEFAULT_MYSQL_CONNECTION_STRING); + + let redis_connection_string = + easyenv::get_env_string_or_default( + "REDIS_URL", + DEFAULT_REDIS_CONNECTION_STRING); + + let pool = MySqlPoolOptions::new() + .max_connections(5) + .connect(&db_connection_string) + .await?; + + let redis_manager = RedisConnectionManager::new(redis_connection_string.clone())?; + + let redis_pool = r2d2::Pool::builder() + .build(redis_manager)?; + + info!("Reading env vars and setting up utils..."); + + let bind_address = easyenv::get_env_string_or_default("BIND_ADDRESS", DEFAULT_BIND_ADDRESS); + let num_workers = easyenv::get_env_num("NUM_WORKERS", 8)?; + let hmac_secret = easyenv::get_env_string_or_default("COOKIE_SECRET", "notsecret"); + let cookie_domain = easyenv::get_env_string_or_default("COOKIE_DOMAIN", ".vo.codes"); + let cookie_secure = easyenv::get_env_bool_or_default("COOKIE_SECURE", true); + let cookie_http_only = easyenv::get_env_bool_or_default("COOKIE_HTTP_ONLY", true); + let website_homepage_redirect = + easyenv::get_env_string_or_default("WEBSITE_HOMEPAGE_REDIRECT", "https://vo.codes/"); + + let server_state = ObsGatewayServerState { + env_config: EnvConfig { + num_workers, + bind_address, + cookie_domain, + cookie_secure, + cookie_http_only, + website_homepage_redirect, + }, + hostname: server_hostname, + }; + + serve(server_state) + .await?; + Ok(()) +} + +pub async fn serve(server_state: ObsGatewayServerState) -> AnyhowResult<()> +{ + let bind_address = server_state.env_config.bind_address.clone(); + let num_workers = server_state.env_config.num_workers.clone(); + let hostname = server_state.hostname.clone(); + + let server_state_arc = web::Data::new(Arc::new(server_state)); + + info!("Starting HTTP service."); + + let log_format = "[%{HOSTNAME}e] %a \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T"; + + HttpServer::new(move || { + App::new() + .app_data(server_state_arc.clone()) + .wrap(Cors::default() + .allowed_origin("http://api.fakeyou.com") + .allowed_origin("http://api.jungle.horse") + .allowed_origin("http://api.vo.codes") + .allowed_origin("http://fakeyou.com") + .allowed_origin("http://jungle.horse") + .allowed_origin("http://jungle.horse:12345") + .allowed_origin("http://jungle.horse:7000") + .allowed_origin("http://localhost:12345") + .allowed_origin("http://localhost:3000") + .allowed_origin("http://localhost:5555") + .allowed_origin("http://localhost:7000") + .allowed_origin("http://localhost:8000") + .allowed_origin("http://localhost:8080") + .allowed_origin("http://mumble.stream") + .allowed_origin("http://trumped.com") + .allowed_origin("http://vo.codes") + .allowed_origin("http://vocodes.com") + .allowed_origin("https://api.fakeyou.com") + .allowed_origin("https://api.jungle.horse") + .allowed_origin("https://api.vo.codes") + .allowed_origin("https://fakeyou.com") + .allowed_origin("https://jungle.horse") + .allowed_origin("https://mumble.stream") + .allowed_origin("https://trumped.com") + .allowed_origin("https://vo.codes") + .allowed_origin("https://vocodes.com") + .allowed_methods(vec!["GET", "POST", "OPTIONS"]) + .supports_credentials() + .allowed_headers(vec![ + http::header::ACCEPT, + http::header::ACCESS_CONTROL_ALLOW_ORIGIN, // Tabulator Ajax + http::header::CONTENT_TYPE, + http::header::ACCESS_CONTROL_ALLOW_CREDENTIALS, // https://stackoverflow.com/a/46412839 + http::header::HeaderName::from_static("x-requested-with") // Tabulator Ajax sends + ]) + .max_age(3600)) + .wrap(Logger::new(&log_format) + .exclude("/liveness") + .exclude("/readiness")) + .wrap(DefaultHeaders::new() + .header("X-Backend-Hostname", &hostname) + .header("X-Build-Sha", "")) + // Twitch + //.service( + // web::resource("/twitch") + // .route(web::get().to(ws_index)) + // .route(web::head().to(|| HttpResponse::Ok())) + //) + //.service(get_root_index) + //.default_service( web::route().to(default_route_404)) + }) + .bind(bind_address)? + .workers(num_workers) + .run() + .await?; + + Ok(()) +} diff --git a/test.js b/test.js new file mode 100644 index 0000000000..e5e40c533b --- /dev/null +++ b/test.js @@ -0,0 +1,10 @@ +//var sock = new WebSocket("ws://api.jungle.horse:80/twitch"); +var sock = new WebSocket("ws://localhost:12345/twitch"); + +sock.onopen = function (event) { + //sock.send("Here's some text that the server is urgently awaiting!"); +}; + +sock.onmessage = function (event) { + console.log('got a message', event.data); +} \ No newline at end of file