From da5c1893b318ba2ebd212f2c7c29a2b0ce30dc2f Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 31 Jan 2022 01:39:16 -0500 Subject: [PATCH] record channel bits event --- .../src/twitch_pubsub/insert_bits.rs | 29 +++---- .../twitch_pubsub/insert_channel_points.rs | 4 +- .../twitch_pubsub_user_subscriber_thread.rs | 81 ++++++++++++++++--- 3 files changed, 84 insertions(+), 30 deletions(-) diff --git a/crates/lib/database_queries/src/twitch_pubsub/insert_bits.rs b/crates/lib/database_queries/src/twitch_pubsub/insert_bits.rs index 16bb404bdc..73d1239816 100644 --- a/crates/lib/database_queries/src/twitch_pubsub/insert_bits.rs +++ b/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 } diff --git a/crates/lib/database_queries/src/twitch_pubsub/insert_channel_points.rs b/crates/lib/database_queries/src/twitch_pubsub/insert_channel_points.rs index 5bae2f0d12..0a701affda 100644 --- a/crates/lib/database_queries/src/twitch_pubsub/insert_channel_points.rs +++ b/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, @@ -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, diff --git a/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs b/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs index 5fd13eb7b4..8037bcc8b1 100644 --- a/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs +++ b/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> + { + 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()