record channel bits event

crates/lib/database_queries/src/twitch_pubsub/insert_bits.rs

@@ -40,29 +40,24 @@ impl TwitchPubsubBitsInsertBuilder {
     }
   }
 
-  pub fn set_sender_twitch_user_id(mut self, value: Option<&str>) -> Self {
-    self.sender_twitch_user_id = value.map(|s| s.to_string());
+  pub fn set_sender_twitch_user_id(mut self, value: &str) -> Self {
+    self.sender_twitch_user_id = Some(value.to_string());
     self
   }
 
-  pub fn set_sender_twitch_username(mut self, value: Option<&str>) -> Self {
-    if let Some(username) = value {
-      self.sender_twitch_username = Some(username.to_string());
-      self.sender_twitch_username_lowercase = Some(username.to_uppercase());
-    } else {
-      self.sender_twitch_username = None;
-      self.sender_twitch_username_lowercase = None;
-    }
+  pub fn set_sender_twitch_username(mut self, value: &str) -> Self {
+    self.sender_twitch_username = Some(value.to_string());
+    self.sender_twitch_username_lowercase = Some(value.to_uppercase());
     self
   }
 
-  pub fn set_destination_channel_id(mut self, value: Option<&str>) -> Self {
-    self.destination_channel_id = value.map(|s| s.to_string());
+  pub fn set_destination_channel_id(mut self, value: &str) -> Self {
+    self.destination_channel_id = Some(value.to_string());
     self
   }
 
-  pub fn set_destination_channel_name(mut self, value: Option<&str>) -> Self {
-    self.destination_channel_name = value.map(|s| s.to_string());
+  pub fn set_destination_channel_name(mut self, value: &str) -> Self {
+    self.destination_channel_name = Some(value.to_string());
     self
   }
 
@@ -81,10 +76,8 @@ impl TwitchPubsubBitsInsertBuilder {
     self
   }
 
-  pub fn set_chat_message(mut self, value: Option<&str>) -> Self {
-    self.chat_message = value
-        .map(|s| s.to_string())
-        .unwrap_or("".to_string());
+  pub fn set_chat_message(mut self, value: &str) -> Self {
+    self.chat_message = value.to_string();
     self
   }
 

crates/lib/database_queries/src/twitch_pubsub/insert_channel_points.rs

@@ -5,7 +5,7 @@ use log::info;
 use log::warn;
 use sqlx::MySqlPool;
 
-pub struct TwitchPubsubBitsInsertBuilder {
+pub struct TwitchPubsubChannelPointsInsertBuilder {
   // ===== Required Fields =====
 
   sender_twitch_user_id: Option<String>,
@@ -32,7 +32,7 @@ pub struct TwitchPubsubBitsInsertBuilder {
   max_per_user_per_stream: u64,
 }
 
-impl TwitchPubsubBitsInsertBuilder {
+impl TwitchPubsubChannelPointsInsertBuilder {
   pub fn new() -> Self {
     Self {
       sender_twitch_user_id: None,

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

@@ -10,6 +10,8 @@ use crate::twitch::twitch_user_id::TwitchUserId;
 use crate::twitch::websocket_client::TwitchWebsocketClient;
 use database_queries::twitch_oauth::find::{TwitchOauthTokenRecord, TwitchOauthTokenFinder};
 use database_queries::twitch_oauth::insert::TwitchOauthTokenInsertBuilder;
+use database_queries::twitch_pubsub::insert_bits::TwitchPubsubBitsInsertBuilder;
+use database_queries::twitch_pubsub::insert_channel_points::TwitchPubsubChannelPointsInsertBuilder;
 use log::debug;
 use log::error;
 use log::info;
@@ -23,7 +25,8 @@ use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
 use time::Instant;
-use twitch_api2::pubsub::Response;
+use twitch_api2::pubsub::{Response, TwitchResponse, TopicData};
+use twitch_api2::pubsub::channel_bits::ChannelBitsEventsV2Reply;
 
 // TODO: Publish events back to OBS thread
 // TODO: (cleanup) make the logic clearer to follow.
@@ -258,23 +261,81 @@ impl TwitchPubsubUserSubscriberThreadStageTwo {
 
     match event {
       Response::Response(response) => {
-        match response.error.as_deref() {
-          Some("ERR_BADAUTH") => {
-            warn!("Invalid token. Bad auth. Need to refresh");
-            let token_record = self.refresh_twitch_oauth_token().await?;
-            return Ok(Some(LoopEndedReason::RefreshedOauthToken { token: token_record }));
-          }
-          _ => {},
-        }
+        // NB: Auth failure might cause the loop to end.
+        return self.handle_maybe_auth_error_event(response).await;
+      }
+      Response::Message { data } => {
+        self.handle_pubsub_topic_event(data).await?;
       }
-      Response::Message { .. } => {}
       Response::Pong => {}
       Response::Reconnect => {}
     }
 
+    Ok(None) // Don't end loop
+  }
+
+  async fn handle_maybe_auth_error_event(&mut self, response: TwitchResponse)
+    -> AnyhowResult<Option<LoopEndedReason>>
+  {
+    let error = match response.error.as_deref() {
+      None => return Ok(None),
+      Some(e) => e,
+    };
+    match error {
+      "ERR_BADAUTH" => {
+        warn!("Invalid token. Bad auth. Need to refresh");
+        let token_record = self.refresh_twitch_oauth_token().await?;
+        return Ok(Some(LoopEndedReason::RefreshedOauthToken { token: token_record }));
+      }
+      _ => warn!("Unknown Twitch PubSub error: {:?}", error),
+    }
     Ok(None)
   }
 
+  async fn handle_pubsub_topic_event(&mut self, topic_data: TopicData) -> AnyhowResult<()> {
+    match topic_data {
+      // Unimplemented
+      TopicData::AutoModQueue { .. } => {}
+      TopicData::ChannelBitsBadgeUnlocks { .. } => {}
+      TopicData::ChatModeratorActions { .. } => {}
+      TopicData::ChannelSubscribeEventsV1 { .. } => {}
+      TopicData::CommunityPointsChannelV1 { .. } => {}
+      TopicData::ChannelCheerEventsPublicV1 { .. } => {}
+      TopicData::ChannelSubGiftsV1 { .. } => {}
+      TopicData::VideoPlayback { .. } => {}
+      TopicData::VideoPlaybackById { .. } => {}
+      TopicData::HypeTrainEventsV1 { .. } => {}
+      TopicData::HypeTrainEventsV1Rewards { .. } => {}
+      TopicData::Following { .. } => {}
+      TopicData::Raid { .. } => {}
+      TopicData::UserModerationNotifications { .. } => {}
+      // Implemented
+      TopicData::ChannelBitsEventsV2 { topic, reply } => {
+        let mut event_builder = TwitchPubsubBitsInsertBuilder::new();
+        match *reply {
+          ChannelBitsEventsV2Reply::BitsEvent { data, message_id, version, is_anonymous } => {
+            let user_id = data.user_id.to_string();
+            let user_name = data.user_name.to_string();
+            let mut event_builder = event_builder.set_sender_twitch_user_id(&user_id)
+                .set_sender_twitch_username(&user_name)
+                .set_destination_channel_id(&data.channel_id.to_string())
+                .set_destination_channel_name(&data.channel_name.to_string())
+                .set_bits_used(data.bits_used as u64)
+                .set_total_bits_used(data.total_bits_used as u64)
+                .set_is_anonymous(is_anonymous)
+                .set_chat_message(&data.chat_message);
+            event_builder.insert(&self.mysql_pool).await?;
+          }
+          _ => {}
+        }
+      }
+      TopicData::ChannelPointsChannelV1 { topic, reply } => {
+      }
+    }
+
+    Ok(())
+  }
+
   async fn maybe_send_twitch_ping(&mut self) -> AnyhowResult<()> {
     let mut should_send_ping = self.twitch_last_pinged_at
         .elapsed()