more cleanup

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

@@ -65,7 +65,7 @@ pub async fn obs_gateway_websocket_handler(
     },
   };
 
-  let twitch_user_id = TwitchUserId::from_string(&token_record.twitch_user_id)
+  let twitch_user_id = TwitchUserId::from_str(&token_record.twitch_user_id)
       .map_err(|e| {
         error!("Error converting twitch user id: {}, id= {}", e, &token_record.twitch_user_id);
         CommonServerError::ServerError

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

@@ -14,6 +14,7 @@ use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
 use tokio::runtime::Runtime;
+use crate::twitch::twitch_user_id::TwitchUserId;
 
 pub async fn listen_for_active_obs_session_thread(
   mysql_pool: Arc<sqlx::Pool<MySql>>,
@@ -27,26 +28,13 @@ pub async fn listen_for_active_obs_session_thread(
   let channel = RedisKeys::obs_active_session_topic();
   pubsub.subscribe(channel).unwrap();
 
-  let mut count = 0;
-
-  //loop {
-  //  let payload : String = try!(msg.get_payload());
-  //  println!("channel '{}': {}", msg.get_channel_name(), payload);
-  //}
-
   loop {
-    //info!("[PubSub]");
-
     // TODO: ERROR HANDLING
     let message = pubsub.get_message().unwrap();
     let payload : String = message.get_payload().unwrap();
 
-    //info!("Message: {}", payload);
-
     let payload = ObsActivePayload::from_json_str(&payload).unwrap();
 
-    //info!("Message decoded: {:?}", payload);
-
     let lease_key = RedisKeys::twitch_pubsub_lease(&payload.twitch_user_id);
 
     let mut redis = redis_pool.get().unwrap();
@@ -54,11 +42,7 @@ pub async fn listen_for_active_obs_session_thread(
     let lease_value : Option<String> = redis.get(&lease_key).unwrap();
 
     if let Some(value) = lease_value.as_deref() {
-      //info!("Already has lease.");
       let lease = LeasePayload::deserialize(value).unwrap();
-
-      //info!("Lease value: {:?}", &lease);
-
       continue;
     }
 
@@ -73,65 +57,11 @@ pub async fn listen_for_active_obs_session_thread(
           LEASE_TIMEOUT_SECONDS
     ).unwrap();
 
-    let twitch_user_id = payload.twitch_user_id.clone();
+    let twitch_user_id = TwitchUserId::from_str(&payload.twitch_user_id).unwrap();
+
     let redis_pool2 = redis_pool.clone();
     let mysql_pool2 = mysql_pool.clone();
 
     runtime.spawn(twitch_pubsub_user_subscriber_thread(twitch_user_id, mysql_pool2, redis_pool2));
-
-
-    // Publish: (ActiveSession, user_id)
-    //  - On: First connect
-    //  - Every 30 seconds
-    //  - 1,000 users  x  10 servers  =  10,000  = 333 QPS
-    //  - 100 users  x  5 servers  =  500  = 16 QPS
-    // Redis Benchmarks
-    //   SET: 552028.75 requests per second (high spec machine)
-    //   GET: 707463.75 requests per second (high spec machine)
-    //   SET: 100007 requests completed in 0.88 seconds (no pipeline)
-    //   GET: 100000 requests completed in 1.23 seconds (no pipeline)
-
-    // Lease Key w/ 60s TTL: "ActiveSessionLease:{user_id}"
-    // Lease Value: "{serverNonce}:{threadNonce}" (to kill same-server dupes)
-
-
-
-
-    // Data structures:
-    // - HashSet of user ids (maybe not needed?)
-    // - Random threads (don't need to be tracked)
-    // - Redis leases that expire every 30 seconds - 1 minute
-    //   --- But polling these leases in a big cluster is high QPS
-
-    // > New UserId:
-    // If not present in HashSet, try to create a Redis lease
-    // If lease already taken, ignore.
-    // If lease created, create a new thread.
-    // Thread continually renews lease (every 10 seconds)
-    // Puts in a threadlocal random nonce into lease. If differs at
-    //  any point, we know some other process somehow got a lease. Abort.
-
-
-    // frontend --> lease key
-    //   -- no, frontend won't know about disconnects, just drought of events
-
-/*
-    THIS WORKS
-    if count < 3 {
-      info!("Spawning....");
-
-      let count2 = count;
-
-      runtime.spawn(async move {
-        let count3 = count2;
-        loop {
-          info!(".....spawned..... {}", count3);
-          sleep(Duration::from_millis(1_000));
-        }
-      });
-    }
-
-    count += 1;
- */
   }
 }

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

@@ -1,6 +1,7 @@
 use crate::redis::lease_payload::LeasePayload;
 use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS;
 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;
 use database_queries::twitch_oauth::find::{TwitchOauthTokenRecord, TwitchOauthTokenFinder};
 use log::info;
@@ -17,20 +18,19 @@ use twitch_api2::pubsub::Response;
 
 /// Receive both forms of twitch userid
 pub async fn twitch_pubsub_user_subscriber_thread(
-  twitch_user_id: String,
+  twitch_user_id: TwitchUserId,
   mysql_pool: Arc<sqlx::Pool<MySql>>,
   redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
 ) {
 
   // TODO: Error handling
-  let numeric_twitch_user_id = twitch_user_id.parse::<u32>().unwrap();
 
   loop {
     info!("Twitch subscriber thread");
 
     let mut redis = redis_pool.get().unwrap();
 
-    let lease_key = RedisKeys::twitch_pubsub_lease(&twitch_user_id);
+    let lease_key = RedisKeys::twitch_pubsub_lease(twitch_user_id.get_str());
     let lease = LeasePayload::new("foo", "bar");
 
     let serialized = lease.serialize();
@@ -41,7 +41,7 @@ pub async fn twitch_pubsub_user_subscriber_thread(
     ).unwrap();
 
     let result = TwitchOauthTokenFinder::new()
-        .scope_twitch_user_id(Some(numeric_twitch_user_id))
+        .scope_twitch_user_id(Some(twitch_user_id.get_numeric()))
         .perform_query(&mysql_pool)
         .await
         .unwrap();
@@ -69,7 +69,7 @@ pub async fn twitch_pubsub_user_subscriber_thread(
     client.send_ping().await.unwrap();
 
     warn!("Listen to user...");
-    let topics = build_pubsub_topics_for_user(numeric_twitch_user_id);
+    let topics = build_pubsub_topics_for_user(twitch_user_id.get_numeric());
 
     client.listen(&record.access_token, &topics).await.unwrap();
 

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

@@ -10,7 +10,7 @@ pub struct TwitchUserId {
 
 impl TwitchUserId {
 
-  pub fn from_string(twitch_user_id: &str) -> AnyhowResult<Self> {
+  pub fn from_str(twitch_user_id: &str) -> AnyhowResult<Self> {
     let numeric_id = twitch_user_id.parse::<u32>()?;
     Ok(Self {
       string_user_id: twitch_user_id.to_string(),