cleanup

crates/service/obs_gateway/Cargo.toml

@@ -9,7 +9,7 @@ authors = [
 publish = false
 
 [[bin]]
-name = "obs-gateway-server"
+name = "obs-gateway"
 path = "src/main.rs"
 
 [[bin]]
@@ -26,19 +26,18 @@ http_server_common = { path = "../../lib/http_server_common" }
 redis_common = { path = "../../lib/redis_common" }
 
 # External
-futures-timer = "3.0.2"
 anyhow = "1.0.40"
 base64 = "0.13.0"
 chrono = { version = "0.4.19", features = ["serde"] }
 derive_more = "0.99.14"
 dotenv = "0.15.0"
 easyenv = "0.0.6"
+futures-timer = "3.0.2"
 hex = "0.4.3"
 hostname = "0.3.1"
 lazy_static = "1.4.0"
 lfu = "0.2.5"
 log = "0.4.14"
-#newrelic-telemetry = { path = "../../../vendor/newrelic-telemetry-sdk-rust" }
 rand = "0.8.3"
 regex = "1.5.4"
 time = "0.2.26"

crates/service/obs_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs

@@ -7,8 +7,6 @@ use container_common::token::random_crockford_token::random_crockford_token;
 use crate::endpoints_ws::obs_twitch_thread::ObsTwitchThread;
 use crate::redis::obs_active_payload::ObsActivePayload;
 use crate::server_state::ObsGatewayServerState;
-use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher;
-use crate::twitch::polling_websocket_client::PollingTwitchWebsocketClient;
 use crate::twitch::pubsub::build_pubsub_topics_for_user::build_pubsub_topics_for_user;
 use crate::twitch::twitch_user_id::TwitchUserId;
 use crate::twitch::websocket_client::TwitchWebsocketClient;
@@ -95,10 +93,6 @@ pub async fn obs_gateway_websocket_handler(
       })?;
 
   let mut client = TwitchWebsocketClient::new().unwrap();
-  let token_refresher = OauthTokenRefresher::new(
-    twitch_user_id.get_numeric(),
-    &token_record.access_token,
-    token_record.maybe_refresh_token.as_deref());
 
   info!("Connecting to Twitch PubSub...");
   client.connect().await.unwrap();
@@ -128,7 +122,6 @@ pub async fn obs_gateway_websocket_handler(
   let websocket = ObsGatewayWebSocket::new(
     twitch_user_id.clone(),
     client,
-    token_refresher,
     server_state_arc
   );
 
@@ -150,12 +143,10 @@ impl ObsGatewayWebSocket {
   fn new(
     twitch_user_id: TwitchUserId,
     twitch_client: TwitchWebsocketClient,
-    oauth_token_refresher: OauthTokenRefresher,
     server_state: Arc<ObsGatewayServerState>,
   ) -> Self {
     let twitch_thread = Arc::new(ObsTwitchThread::new(
       twitch_user_id.clone(),
-      oauth_token_refresher,
       twitch_client));
     Self {
       twitch_user_id,

crates/service/obs_gateway/src/endpoints_ws/obs_twitch_thread.rs

@@ -1,5 +1,3 @@
-use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher;
-use crate::twitch::polling_websocket_client::PollingTwitchWebsocketClient;
 use crate::twitch::twitch_user_id::TwitchUserId;
 use crate::twitch::websocket_client::TwitchWebsocketClient;
 use futures::lock::Mutex;
@@ -29,7 +27,6 @@ pub struct ObsTwitchThread {
 
 struct InnerData {
   twitch_user_id: TwitchUserId,
-  token_refresher: OauthTokenRefresher,
   is_connected: bool,
   twitch_client: TwitchWebsocketClient,
   last_ping: SystemTime,
@@ -38,14 +35,12 @@ struct InnerData {
 impl ObsTwitchThread {
   pub fn new(
     twitch_user_id: TwitchUserId,
-    token_refresher: OauthTokenRefresher,
     twitch_client: TwitchWebsocketClient
   ) -> Self {
     let now = Instant::now();
     Self {
       inner_data: Arc::new(Mutex::new(InnerData {
         twitch_user_id,
-        token_refresher,
         is_connected: false,
         twitch_client,
         last_ping: UNIX_EPOCH,

crates/service/obs_gateway/src/main.rs

@@ -32,8 +32,6 @@ use crate::endpoints::oauth_begin_redirect::oauth_begin_enroll_redirect;
 use crate::endpoints::oauth_end::oauth_end_enroll_from_redirect;
 use crate::endpoints_ws::obs_gateway_websocket_handler::obs_gateway_websocket_handler;
 use crate::server_state::{ObsGatewayServerState, EnvConfig, TwitchOauthSecrets, TwitchOauthTemp, BackendsConfig};
-use crate::twitch::polling_websocket_client::PollingTwitchWebsocketClient;
-use crate::twitch::twitch_client_wrapper::TwitchClientWrapper;
 use crate::twitch::twitch_secrets::TwitchSecrets;
 use futures::Future;
 use futures::executor::ThreadPool;
@@ -234,121 +232,3 @@ pub async fn serve(server_state: ObsGatewayServerState) -> AnyhowResult<()>
 
   Ok(())
 }
-
-
-
-// ========================================================
-
-// TODO: Temp comment out to debug actix-web vs. runtime
-//
-//info!("Getting app access token...");
-
-////let scopes = Scope::all();
-//let scopes = vec![
-//  Scope::BitsRead,
-//  Scope::UserReadEmail,
-//];
-
-//let mut twitch_client = TwitchClientWrapper::new(client_id.clone(), client_secret.clone());
-//twitch_client.request_access_token(scopes).await?;
-
-//info!("Getting user id ...");
-
-////let user_id = twitch_client.get_user_id_from_username("testytest512").await?;
-//let user_id = twitch_client.get_user_id_from_username("vocodes").await?;
-
-//info!("User ID: {}", user_id);
-
-//std::thread::sleep(Duration::from_secs(5000));
-
-
-//  // ==================== OAUTH FLOW ====================
-
-
-
-//
-//  let auth_token = user_token.access_token.as_str();
-//
-//  // ==================== PUBSUB SUBSCRIPTION AND MAIN LOOP ====================
-//
-//  let mut client = PollingTwitchWebsocketClient::new()?;
-//
-//  println!("Connecting PubSub...");
-//  client.connect().await?;
-//
-//  println!("Connected");
-//
-//  //println!("Starting polling thread...");
-//  //client.start_ping_thread().await;
-//
-//  println!("Sending PING...");
-//
-//  client.send_ping().await?;
-//
-//  println!("Try read next...");
-//  let r = client.try_next().await?;
-//  println!("Result: {:?}", r);
-//
-//
-//  println!("Begin LISTEN...");
-//
-//  let bit_topic = pubsub::channel_bits::ChannelBitsEventsV2 {
-//    channel_id: user_id,
-//  }.into_topic();
-//
-//  let topics = [bit_topic];
-//
-//  client.listen(&auth_token, &topics).await?;
-//
-//  println!("Try read next...");
-//  let r = client.try_next().await?;
-//  println!("Result: {:?}", r);
-//
-//  /*
-//  Result: Some(Message { data: ChannelBitsEventsV2 { topic: ChannelBitsEventsV2 {
-//  channel_id: 652567283 }, reply: BitsEvent { data: BitsEventData { badge_entitlement:
-//  Some(BadgeEntitlement { new_version: 100, previous_version: 1 }), bits_used: 100,
-//  channel_id: "652567283", channel_name: "vocodes", chat_message: "Cheer100 testing the cheer",
-//  context: Cheer, is_anonymous: false, time: "2021-09-27T04:30:53.627717085Z",
-//  total_bits_used: 101, user_id: "650154491", user_name: "testytest512" },
-//  message_id: "793f745e-8f3e-5f71-bc41-4808c4b49a53", version: "1.0", is_anonymous: false } } })
-//
-//  Result: Some(Message { data: ChannelBitsEventsV2 { topic: ChannelBitsEventsV2 {
-//  channel_id: 652567283 }, reply: BitsEvent { data: BitsEventData { badge_entitlement: None,
-//  bits_used: 1, channel_id: "652567283", channel_name: "vocodes",
-//  chat_message: "mario: hello everybody Cheer1", context: Cheer, is_anonymous: false,
-//  time: "2021-09-27T04:47:32.943872952Z", total_bits_used: 102, user_id: "650154491",
-//  user_name: "testytest512" }, message_id: "5844328d-495b-5585-9d6c-3c99949f9a0a", version: "1.0",
-//  is_anonymous: false } } })
-//   */
-//
-//
-//
-//  println!("Try read next...");
-//  let r = client.try_next().await?;
-//  println!("Result: {:?}", r);
-//
-//  println!("Sleep...");
-//  std::thread::sleep(Duration::from_millis(30000));
-//
-//  //let text = msg.into_text()?;
-//  //println!("Text: {}", text);
-//
-//  // ========================================================
-
-//  info!("Connecting to database...");
-//
-//  let db_connection_string =
-//      easyenv::get_env_string_or_default(
-//        "MYSQL_URL",
-//        DEFAULT_MYSQL_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)?;

crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs

@@ -24,9 +24,7 @@ use crate::twitch::constants::TWITCH_PING_CADENCE;
 // TODO: ERROR HANDLING
 // TODO: ERROR HANDLING
 
-// TODO: Timer to ping().
-// TODO: Timer to renew lease.
-// TODO: Publish events back to OBS thread.
+// TODO: Publish events back to OBS thread (keepalive redis key from websocket).
 // TODO: Disconnect when OBS is done.
 // TODO: Refresh oauth token.
 // TODO: Handle disconnects.
@@ -137,24 +135,8 @@ impl TwitchPubsubUserSubscriberThread {
 
         sleep(Duration::from_secs(1));
 
-//        match maybe_event {
-//          None => {
-//          }
-//          Some(ref event) => {
-//            warn!("event: {:?}", event);
-//
-//            match event {
-//              Response::Response(_) => {}
-//              Response::Message { .. } => {}
-//              Response::Pong => {}
-//              Response::Reconnect => {}
-//            }
-//          }
-//        }
-
         self.maybe_renew_redis_lease().unwrap();
         self.maybe_send_twitch_ping().await.unwrap();
-
       }
     }
   }

crates/service/obs_gateway/src/twitch/mod.rs

@@ -1,8 +1,6 @@
 pub mod constants;
 pub mod oauth;
-pub mod polling_websocket_client;
 pub mod pubsub;
-pub mod twitch_client_wrapper;
 pub mod twitch_secrets;
 pub mod twitch_user_id;
 pub mod websocket_client;

crates/service/obs_gateway/src/twitch/oauth/mod.rs

@@ -1,2 +1 @@
-pub mod oauth_token_builder;
-pub mod oauth_token_refresher;
+pub mod oauth_token_builder;

crates/service/obs_gateway/src/twitch/oauth/oauth_token_refresher.rs

@@ -1,24 +0,0 @@
-// TODO: I know this is something we'll need, but I haven't built it yet.
-//  We'll probably want to write back to the database.
-
-#[derive(Clone)]
-pub struct OauthTokenRefresher {
-  user_id: u32,
-  auth_token: String,
-  refresh_token: Option<String>,
-}
-
-impl OauthTokenRefresher {
-
-  pub fn new(user_id: u32, auth_token: &str, refresh_token: Option<&str>) -> Self {
-    Self {
-      user_id,
-      auth_token: auth_token.to_string(),
-      refresh_token: refresh_token.map(|str| str.to_string()),
-    }
-  }
-
-  pub fn get_auth_token(&self) -> String {
-    self.auth_token.clone()
-  }
-}

crates/service/obs_gateway/src/twitch/polling_websocket_client.rs

@@ -1,146 +0,0 @@
-use anyhow::anyhow;
-use container_common::anyhow_result::AnyhowResult;
-use crate::twitch::websocket_client::TwitchWebsocketClient;
-use std::sync::atomic::AtomicBool;
-use std::sync::{Arc, Mutex};
-use std::time::Duration;
-use twitch_api2::pubsub::{Response, TopicData, Topic, Topics};
-
-const PING_SECONDS : u64 = 1 * 60; // Twitch calls for every 5 minutes, we'll do every 1 minute.
-
-pub struct PollingTwitchWebsocketClient {
-  client: Arc<Mutex<TwitchWebsocketClient>>,
-  is_polling: AtomicBool,
-}
-
-impl PollingTwitchWebsocketClient {
-  pub fn new() -> AnyhowResult<Self> {
-    Ok(Self {
-      client: Arc::new(Mutex::new(TwitchWebsocketClient::new()?)),
-      is_polling: AtomicBool::new(false),
-    })
-  }
-
-  pub async fn connect(&mut self) -> AnyhowResult<()> {
-    match self.client.lock() {
-      Err(e) => Err(anyhow!("unlock client err: {:?}", e)),
-      Ok(mut client) => {
-        client.connect().await
-      }
-    }
-  }
-
-  pub async fn listen<'t>(&mut self, auth_token: &str, topics: &'t [Topics]) -> AnyhowResult<()> {
-    match self.client.lock() {
-      Err(e) => Err(anyhow!("unlock client err: {:?}", e)),
-      Ok(mut client) => {
-        client.listen(auth_token, topics).await
-      }
-    }
-  }
-
-  pub async fn send_ping(&mut self) -> AnyhowResult<()> {
-    match self.client.lock() {
-      Err(e) => Err(anyhow!("unlock client err: {:?}", e)),
-      Ok(mut client) => {
-        client.send_ping().await
-      }
-    }
-  }
-
-  /// Start a pinging background thread.
-  /// Per Twitch PubSub docs,
-  ///   To keep the server from closing the connection, clients must send a PING command
-  ///   at least once every 5 minutes. If a client does not receive a PONG message within
-  ///   10 seconds of issuing a PING command, it should reconnect to the server.
-  pub async fn start_ping_thread(&self) {
-    let client = self.client.clone();
-    tokio::spawn(async move {
-      loop {
-        match client.lock() {
-          Ok(mut c) => {
-            // TODO: Future isn't Send
-            //let _r = c.send_ping().await;
-            // TODO: Need to read from all call sites due to ordering.
-            //c.read_response();
-          }
-          Err(_) => {
-            // TODO
-          }
-        }
-        tokio::time::sleep(Duration::from_secs(PING_SECONDS)).await;
-      }
-    });
-  }
-
-  pub async fn next(&mut self) -> AnyhowResult<Response> {
-    match self.client.lock() {
-      Err(e) => Err(anyhow!("unlock client err: {:?}", e)),
-      Ok(mut client) => {
-        client.next().await
-      }
-    }
-  }
-
-  pub async fn try_next(&mut self) -> AnyhowResult<Option<Response>> {
-    match self.client.lock() {
-      Err(e) => Err(anyhow!("unlock client err: {:?}", e)),
-      Ok(mut client) => {
-        client.try_next().await
-      }
-    }
-  }
-
-  /*async fn start_polling(&self) {
-    if self.is_polling.load(Ordering::Relaxed) {
-      return;
-    }
-
-    tokio::spawn(async move {
-      self.is_polling.store(true, Ordering::Relaxed);
-
-      let command = pubsub::channel_sub_gifts(
-        &topics,
-        "authtoken",
-        "random nonce string",
-      )?;
-
-    });
-     /
-  }*/
-
-
-  /*let message = Message::Text(PING_COMMAND.to_string());
-  socket.send(message).await?;
-  match self.read_response().await {
-    Err(e) => println!("Error getting PONG: {:?}", e),
-    Ok(typed_response) => {
-      match typed_response {
-        Response::Pong => { println!("Pong") }
-        Response::Reconnect => { println!("Reconnect") }
-        Response::Response(_0) => { println!("Sub/Unsub (pubsub?)") }
-        Response::Message { data } => {
-          match data {
-            TopicData::ChannelCheerEventsPublicV1 { topic, reply } => {}
-            TopicData::ChannelSubGiftsV1 { topic, reply } => {}
-            TopicData::ChannelSubscribeEventsV1 { topic, reply } => {}
-            TopicData::CommunityPointsChannelV1 { topic, reply } => {}
-            TopicData::Following { topic, reply} => {}
-            // Not interested in these events
-            TopicData::AutoModQueue { .. } => {}
-            TopicData::ChannelBitsBadgeUnlocks { .. } => {}
-            TopicData::ChannelBitsEventsV2 { .. } => {}
-            TopicData::ChannelPointsChannelV1 { .. } => {}
-            TopicData::ChatModeratorActions { .. } => {}
-            TopicData::HypeTrainEventsV1 { .. } => {}
-            TopicData::HypeTrainEventsV1Rewards { .. } => {}
-            TopicData::Raid { .. } => {}
-            TopicData::UserModerationNotifications { .. } => {}
-            TopicData::VideoPlayback { .. } => {}
-            TopicData::VideoPlaybackById { .. } => {}
-          }
-        }
-      }
-    }
-  }*/
-}

crates/service/obs_gateway/src/twitch/twitch_client_wrapper.rs

@@ -1,65 +0,0 @@
-//! The Helix client is independent of PubSub
-//! This just wraps `twitch_api2` for more convenience.
-
-use anyhow::anyhow;
-use container_common::anyhow_result::AnyhowResult;
-use twitch_api2::TwitchClient;
-use twitch_api2::helix::users::GetUsersRequest;
-use twitch_api2::types::{Nickname, UserId};
-use twitch_oauth2::{ClientSecret, ClientId, AccessToken, AppAccessToken, Scope};
-
-pub struct TwitchClientWrapper<'a> {
-  client_id: ClientId,
-  client_secret: ClientSecret,
-  twitch_client: TwitchClient<'a, reqwest::Client>,
-  access_token: Option<AppAccessToken>,
-}
-
-impl <'a> TwitchClientWrapper <'a> {
-  pub fn new(client_id: ClientId, client_secret: ClientSecret) -> Self {
-    Self {
-      client_id,
-      client_secret,
-      twitch_client: TwitchClient::default(),
-      access_token: None,
-    }
-  }
-
-  pub async fn request_access_token(&mut self, scopes: Vec<Scope>) -> AnyhowResult<()> {
-    self.get_access_token(scopes).await?;
-    Ok(())
-  }
-
-  async fn get_access_token(&mut self, scopes: Vec<Scope>) -> AnyhowResult<()> {
-    let mut http_client = reqwest::Client::new();
-    let access_token = twitch_oauth2::AppAccessToken::get_app_access_token(
-      &http_client,
-      self.client_id.clone(),
-      self.client_secret.clone(),
-      scopes,
-    ).await?;
-
-    self.access_token = Some(access_token);
-    Ok(())
-  }
-
-  /// We frequently need to look up IDs from usernames.
-  pub async fn get_user_id_from_username(&self, username: &str) -> AnyhowResult<u32> {
-    let access_token = match self.access_token {
-      None => return Err(anyhow!("no access token has been granted")),
-      Some(ref token) => token,
-    };
-
-    let req = GetUsersRequest::builder()
-        .login(vec![Nickname::new(username)])
-        .build();
-
-    let response = &self.twitch_client.helix.req_get(req, access_token)
-        .await?;
-
-    match response.data.get(0) {
-      None => Err(anyhow!("No such user: `{}`.", username)),
-      Some(user) => Ok(user.id.as_str().parse::<u32>()?),
-    }
-  }
-}

crates/service/obs_gateway/src/twitch_pubsub_job.rs

@@ -14,22 +14,20 @@
 
 use config::shared_constants::{DEFAULT_RUST_LOG, DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING, DEFAULT_MYSQL_CONNECTION_STRING};
 use container_common::anyhow_result::AnyhowResult;
+use crate::threads::listen_for_active_obs_sessions_thread::listen_for_active_obs_session_thread;
+use crate::twitch::websocket_client::TwitchWebsocketClient;
 use futures::executor::{ThreadPool, ThreadPoolBuilder};
-use r2d2_redis::r2d2;
+use futures::task::SpawnExt;
 use log::info;
+use r2d2_redis::RedisConnectionManager;
+use r2d2_redis::r2d2::Pool;
+use r2d2_redis::r2d2;
+use redis_common::redis_keys::RedisKeys;
+use sqlx::mysql::MySqlPoolOptions;
+use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
-use crate::twitch::twitch_client_wrapper::TwitchClientWrapper;
-use crate::twitch::websocket_client::TwitchWebsocketClient;
-use futures::task::SpawnExt;
 use tokio::runtime::{Builder, Runtime};
-use r2d2_redis::RedisConnectionManager;
-use std::sync::Arc;
-use r2d2_redis::r2d2::Pool;
-use redis_common::redis_keys::RedisKeys;
-use crate::threads::listen_for_active_obs_sessions_thread::listen_for_active_obs_session_thread;
-use sqlx::mysql::MySqlPoolOptions;
-
 pub mod redis;
 pub mod threads;
 pub mod twitch;