begin work on token renewal

crates/lib/database_queries/src/twitch_oauth/find.rs

@@ -45,6 +45,9 @@ pub struct TwitchOauthTokenFinder {
 
   scope_twitch_user_id: Option<u32>,
   scope_twitch_username_lowercase: Option<String>,
+
+  /// Allow return of expired tokens? Expired tokens can be renewed.
+  allow_expired_tokens: bool,
 }
 
 impl TwitchOauthTokenFinder {
@@ -52,7 +55,8 @@ impl TwitchOauthTokenFinder {
     Self {
       scope_user_token: None,
       scope_twitch_user_id: None,
-      scope_twitch_username_lowercase: None
+      scope_twitch_username_lowercase: None,
+      allow_expired_tokens: false,
     }
   }
 
@@ -73,6 +77,11 @@ impl TwitchOauthTokenFinder {
     self
   }
 
+  pub fn allow_expired_tokens(mut self, allow_expired: bool) -> Self {
+    self.allow_expired_tokens = allow_expired;
+    self
+  }
+
   pub async fn perform_query(
     &self,
     mysql_pool: &MySqlPool
@@ -169,7 +178,6 @@ SELECT
     twitch_oauth_tokens.user_deleted_at,
     twitch_oauth_tokens.mod_deleted_at
 
-
 FROM twitch_oauth_tokens
 LEFT OUTER JOIN users
     ON twitch_oauth_tokens.maybe_user_token = users.token
@@ -211,10 +219,12 @@ LEFT OUTER JOIN users
       }
     }
 
-    if !first_predicate_added {
-      query.push_str(" WHERE ( twitch_oauth_tokens.expires_at IS NULL OR twitch_oauth_tokens.expires_at > NOW() ) ");
-    } else {
-      query.push_str(" AND ( twitch_oauth_tokens.expires_at IS NULL OR twitch_oauth_tokens.expires_at > NOW() ) ");
+    if !self.allow_expired_tokens {
+      if !first_predicate_added {
+        query.push_str(" WHERE ( twitch_oauth_tokens.expires_at IS NULL OR twitch_oauth_tokens.expires_at > NOW() ) ");
+      } else {
+        query.push_str(" AND ( twitch_oauth_tokens.expires_at IS NULL OR twitch_oauth_tokens.expires_at > NOW() ) ");
+      }
     }
 
     // NB: Return the most recent.

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

@@ -53,6 +53,7 @@ pub async fn obs_gateway_websocket_handler(
 ) -> Result<HttpResponse, CommonServerError> {
 
   let mut finder = TwitchOauthTokenFinder::new()
+      .allow_expired_tokens(true)
       .scope_twitch_username(Some(&path.twitch_username));
 
   let lookup_result = finder.perform_query(&server_state.backends.mysql_pool).await;
@@ -164,6 +165,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for ObsGatewayWebSock
             CommonServerError::ServerError
           }).unwrap(); // TODO: Fixme
 
+      // TODO: This should be done *BEFORE* PubSub
       self.write_obs_active().unwrap();
 
       match msg {

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

@@ -28,10 +28,8 @@ use twitch_api2::pubsub::Response;
 // TODO: ERROR HANDLING
 
 // TODO: Publish events back to OBS thread
-// TODO: Disconnect when OBS is done. (keepalive redis key from websocket).
 // TODO: Refresh oauth token.
-// TODO: Handle disconnects.
-// TODO: Server+thread IDs
+// TODO: Error handling, handle disconnects.
 
 pub struct TwitchPubsubUserSubscriberThread {
   thread_id: ThreadId,
@@ -282,6 +280,7 @@ impl TwitchPubsubUserSubscriberThread {
   async fn lookup_oauth_record(&mut self) -> AnyhowResult<Option<TwitchOauthTokenRecord>> {
     TwitchOauthTokenFinder::new()
         .scope_twitch_user_id(Some(self.twitch_user_id.get_numeric()))
+        .allow_expired_tokens(true)
         .perform_query(&self.mysql_pool)
         .await
   }