much cleaner thread, but redis lease doesn't get renewed

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

@@ -1,7 +1,7 @@
 use crate::redis::lease_payload::LeasePayload;
 use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS;
 use crate::redis::obs_active_payload::ObsActivePayload;
-use crate::threads::twitch_pubsub_user_subscriber_thread::{twitch_pubsub_user_subscriber_thread, TwitchPubsubUserSubscriberThread};
+use crate::threads::twitch_pubsub_user_subscriber_thread::TwitchPubsubUserSubscriberThread;
 use crate::twitch::twitch_user_id::TwitchUserId;
 use log::error;
 use log::info;

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

@@ -16,6 +16,8 @@ use std::thread::sleep;
 use std::time::Duration;
 use twitch_api2::pubsub::Response;
 use time::Instant;
+use container_common::anyhow_result::AnyhowResult;
+use crate::twitch::constants::TWITCH_PING_CADENCE;
 
 // TODO: ERROR HANDLING
 // TODO: ERROR HANDLING
@@ -35,194 +37,137 @@ pub struct TwitchPubsubUserSubscriberThread {
   twitch_user_id: TwitchUserId,
   mysql_pool: Arc<sqlx::Pool<MySql>>,
   redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
+  twitch_websocket_client: TwitchWebsocketClient,
 
-  // Thread state
-  lease_last_renewed_at: Option<Instant>,
+  // ========== Thread state ==========
+
+  // The thread must renew the lease, or another worker will pick it up.
+  // If the lease gets taken by another, we abandon our own workload.
+  redis_lease_last_renewed_at: Option<Instant>,
+
+  /// Twitch PubSub requires PINGs at regular intervals,
+  ///   "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. See details in Handling Connection Failures."
+  twitch_last_pinged_at: Option<Instant>,
 }
 
 impl TwitchPubsubUserSubscriberThread {
-
   pub fn new(
     twitch_user_id: TwitchUserId,
     mysql_pool: Arc<sqlx::Pool<MySql>>,
     redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
   ) -> Self {
+    let twitch_websocket_client = TwitchWebsocketClient::new().unwrap();
     Self {
       twitch_user_id,
       mysql_pool,
       redis_pool,
-      lease_last_renewed_at: None,
+      twitch_websocket_client,
+      redis_lease_last_renewed_at: None,
+      twitch_last_pinged_at: None,
     }
   }
 
   pub async fn start_thread(mut self) {
-    let mut last_renewed_lease = Instant::now();
+    // We lease from outside before starting the thread.
+    self.redis_lease_last_renewed_at = Some(Instant::now());
 
-    self.foo = Some("asdf".to_string());
+    // Lookup the OAuth credentials
+    let oauth_lookup_result = self.lookup_oauth_record().await;
+    let oauth_token_record = match oauth_lookup_result {
+      Err(e) => {
+        info!("Error looking up oauth creds: {:?}", e);
+        return; // TODO: Better flow.
+      }
+      Ok(None) => {
+        info!("Twitch user oauth token does not exit. Auth please.");
+        return; // TODO: Better flow.
+      }
+      Ok(Some(record)) => record,
+    };
+
+    warn!("Connecting to Twitch PubSub for user {}...", &oauth_token_record.twitch_username);
+    self.twitch_websocket_client.connect().await.unwrap(); // TODO: Error handling
 
     loop {
-      info!("Twitch subscriber thread");
-
-      let now = Instant::now();
-      let mut should_renew = false;
-
-      match now.checked_sub(LEASE_RENEW_PERIOD) {
-        None => { should_renew = true }
-        Some(before) => {
-
-        }
-      }
-
-      let mut redis = self.redis_pool.get().unwrap();
-
-      let lease_key = RedisKeys::twitch_pubsub_lease(self.twitch_user_id.get_str());
-      let lease = LeasePayload::new("foo", "bar");
-
-      let serialized = lease.serialize();
-      let _v : Option<String> = redis.set_ex(
-        &lease_key,
-        &serialized,
-        LEASE_TIMEOUT_SECONDS
-      ).unwrap();
-
-      let result = TwitchOauthTokenFinder::new()
-          .scope_twitch_user_id(Some(self.twitch_user_id.get_numeric()))
-          .perform_query(&self.mysql_pool)
-          .await
-          .unwrap();
-
-      let record = match result {
-        None => {
-          info!("Twitch user oauth token does not exit. Auth please");
-          sleep(Duration::from_secs(5));
-          continue;
-        }
-        Some(record) => record
-      };
-
-      let mut client = TwitchWebsocketClient::new().unwrap();
-      //let token_refresher = OauthTokenRefresher::new(
-      //  user_id,
-      //  &token_record.access_token,
-      //  token_record.maybe_refresh_token.as_deref());
-
-      warn!("Connecting to Twitch PubSub for user {}...", &record.twitch_username);
-      client.connect().await.unwrap();
-
-      // TODO: BACKGROUND PINGS UGH!
-      warn!("Sending ping...");
-      client.send_ping().await.unwrap();
+      self.maybe_renew_redis_lease().unwrap();
+      self.maybe_send_twitch_ping().await.unwrap();
 
       warn!("Listen to user...");
       let topics = build_pubsub_topics_for_user(self.twitch_user_id.get_numeric());
-
-      client.listen(&record.access_token, &topics).await.unwrap();
+      self.twitch_websocket_client.listen(&oauth_token_record.access_token, &topics).await.unwrap();
 
       loop {
-        warn!("client.next()");
-        let maybe_event = client.try_next().await.unwrap();
+        let maybe_event = self.twitch_websocket_client.try_next()
+            .await
+            .unwrap();
 
-        if let Some(event) = maybe_event.as_ref() {
-          warn!("event: {:?}", event);
+        match maybe_event {
+          None => {
+            sleep(Duration::from_secs(5));
+          }
+          Some(ref event) => {
+            warn!("event: {:?}", event);
 
-          match event {
-            Response::Response(_) => {}
-            Response::Message { .. } => {}
-            Response::Pong => {}
-            Response::Reconnect => {}
+            match event {
+              Response::Response(_) => {}
+              Response::Message { .. } => {}
+              Response::Pong => {}
+              Response::Reconnect => {}
+            }
           }
         }
-
-        sleep(Duration::from_secs(5));
       }
     }
   }
 
-}
+  fn maybe_renew_redis_lease(&mut self) -> AnyhowResult<()> {
+    let mut should_renew_lease = self.redis_lease_last_renewed_at
+        .map(|last_write| last_write.elapsed().gt(&LEASE_RENEW_PERIOD))
+        .unwrap_or(true);
 
-
-pub async fn twitch_pubsub_user_subscriber_thread(
-  twitch_user_id: TwitchUserId,
-  mysql_pool: Arc<sqlx::Pool<MySql>>,
-  redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
-) {
-
-  let mut last_renewed_lease = Instant::now();
-
-  loop {
-    info!("Twitch subscriber thread");
-
-    let now = Instant::now();
-    let mut should_renew = false;
-
-    match now.checked_sub(LEASE_RENEW_PERIOD) {
-      None => { should_renew = true }
-      Some(before) => {
-
-      }
+    if should_renew_lease {
+      info!("Renewing Redis Lease for user {}", self.twitch_user_id.get_numeric());
+      self.renew_redis_lease()?;
+      self.redis_lease_last_renewed_at = Some(Instant::now());
     }
 
-    let mut redis = redis_pool.get().unwrap();
+    Ok(())
+  }
 
-    let lease_key = RedisKeys::twitch_pubsub_lease(twitch_user_id.get_str());
-    let lease = LeasePayload::new("foo", "bar");
-
-    let serialized = lease.serialize();
+  fn renew_redis_lease(&mut self) -> AnyhowResult<()> {
+    // TODO: Error handling
+    let mut redis = self.redis_pool.get().unwrap();
+    let lease_key = RedisKeys::twitch_pubsub_lease(self.twitch_user_id.get_str());
+    let lease_value = LeasePayload::new("foo", "bar").serialize();
     let _v : Option<String> = redis.set_ex(
       &lease_key,
-      &serialized,
+      &lease_value,
       LEASE_TIMEOUT_SECONDS
     ).unwrap();
+    Ok(())
+  }
 
-    let result = TwitchOauthTokenFinder::new()
-        .scope_twitch_user_id(Some(twitch_user_id.get_numeric()))
-        .perform_query(&mysql_pool)
+  async fn lookup_oauth_record(&mut self) -> AnyhowResult<Option<TwitchOauthTokenRecord>> {
+    TwitchOauthTokenFinder::new()
+        .scope_twitch_user_id(Some(self.twitch_user_id.get_numeric()))
+        .perform_query(&self.mysql_pool)
         .await
-        .unwrap();
+  }
 
-    let record = match result {
-      None => {
-        info!("Twitch user oauth token does not exit. Auth please");
-        sleep(Duration::from_secs(5));
-        continue;
-      }
-      Some(record) => record
-    };
+  async fn maybe_send_twitch_ping(&mut self) -> AnyhowResult<()> {
+    let mut should_send_ping = self.twitch_last_pinged_at
+        .map(|last_ping| last_ping.elapsed().gt(&TWITCH_PING_CADENCE))
+        .unwrap_or(true);
 
-    let mut client = TwitchWebsocketClient::new().unwrap();
-    //let token_refresher = OauthTokenRefresher::new(
-    //  user_id,
-    //  &token_record.access_token,
-    //  token_record.maybe_refresh_token.as_deref());
-
-    warn!("Connecting to Twitch PubSub for user {}...", &record.twitch_username);
-    client.connect().await.unwrap();
-
-    // TODO: BACKGROUND PINGS UGH!
-    warn!("Sending ping...");
-    client.send_ping().await.unwrap();
-
-    warn!("Listen to user...");
-    let topics = build_pubsub_topics_for_user(twitch_user_id.get_numeric());
-
-    client.listen(&record.access_token, &topics).await.unwrap();
-
-    loop {
-      warn!("client.next()");
-      let maybe_event = client.try_next().await.unwrap();
-
-      if let Some(event) = maybe_event.as_ref() {
-        warn!("event: {:?}", event);
-
-        match event {
-          Response::Response(_) => {}
-          Response::Message { .. } => {}
-          Response::Pong => {}
-          Response::Reconnect => {}
-        }
-      }
-
-      sleep(Duration::from_secs(5));
+    if should_send_ping {
+      info!("Sending Twitch ping for user {}", self.twitch_user_id.get_numeric());
+      self.twitch_websocket_client.send_ping().await?;
+      self.twitch_last_pinged_at = Some(Instant::now());
     }
+
+    Ok(())
   }
 }

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

@@ -0,0 +1,11 @@
+use time::Duration;
+
+/// Twitch PubSub requires PINGs at regular intervals,
+///
+///   "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. See details in Handling Connection Failures."
+///
+/// We'll keep this intentionally short so we don't elapse.
+pub const TWITCH_PING_CADENCE: Duration = Duration::minutes(3);

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

@@ -1,3 +1,4 @@
+pub mod constants;
 pub mod oauth;
 pub mod polling_websocket_client;
 pub mod pubsub;