NEW OBS GATEWAY ARC (NOT YET WOKRING)

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",

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"

migrations/2021-09-29-042235_twitch_oauth_tokens/down.sql

@@ -0,0 +1,4 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+

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.<channel ID>` (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_id>` (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;
+

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;

src/http_server/endpoints/twitch/mod.rs

@@ -0,0 +1 @@
+pub mod pubsub_gateway;

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<HttpResponse, Error> {
+  error!(">>>>>> ws_index");
+  ws::start(WebSocket::new(), &r, stream)
+}
+
+struct WebSocket {}
+
+impl Actor for WebSocket {
+  type Context = ws::WebsocketContext<Self>;
+
+  fn started(&mut self, _ctx: &mut Self::Context) {
+    error!(">>>>>> actor started");
+  }
+}
+
+impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocket {
+  fn handle(
+    &mut self,
+    msg: Result<ws::Message, ws::ProtocolError>,
+    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<Self>;
+//}
+//
+///// Handler for ws::Message message
+//impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWs {
+//  fn handle(
+//    &mut self,
+//    msg: Result<ws::Message, ws::ProtocolError>,
+//    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<HttpResponse, Error> {
+//  let resp = ws::start(MyWs {}, &req, stream);
+//  println!("{:?}", resp);
+//  resp
+//}

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<T> = anyhow::Result<T>;
+
+/// 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(())
+}

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);
+}