From 228e31294939b1fe850d0a2237a3deb0423f4e70 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 24 Jan 2022 00:50:22 -0500 Subject: [PATCH] Add refresh_count to model --- .../2022-01-11-042235_twitch_oauth_tokens/up.sql | 6 ++++++ crates/lib/database_queries/src/twitch_oauth/find.rs | 4 ++++ .../lib/database_queries/src/twitch_oauth/insert.rs | 11 ++++++++++- .../threads/twitch_pubsub_user_subscriber_thread.rs | 12 +++++------- 4 files changed, 25 insertions(+), 8 deletions(-) diff --git a/_migrations/2022-01-11-042235_twitch_oauth_tokens/up.sql b/_migrations/2022-01-11-042235_twitch_oauth_tokens/up.sql index ac837ff3a3..0bf8aa6f48 100644 --- a/_migrations/2022-01-11-042235_twitch_oauth_tokens/up.sql +++ b/_migrations/2022-01-11-042235_twitch_oauth_tokens/up.sql @@ -5,6 +5,7 @@ -- **NOTE ABOUT DESIGN**: -- This table contains _many_ OAuth tokens for any single user. (many users : many tokens) -- An entire history of tokens is kept in the table. +-- Apart from "deleted_at", etc., these records should largely be IMMUTABLE. CREATE TABLE twitch_oauth_tokens( -- Not used for anything except replication. @@ -49,6 +50,11 @@ CREATE TABLE twitch_oauth_tokens( -- Null if it does not expire or we were not informed expires_in_seconds INT(10) UNSIGNED DEFAULT NULL, + -- Number of times this token has been refreshed + -- The original token creation will be "0", and each + -- subsequent time will bump by one. Mostly for debugging. + refresh_count INT(10) UNSIGNED DEFAULT 0, + -- ========== OAUTH SCOPES ========== -- "bits:read" diff --git a/crates/lib/database_queries/src/twitch_oauth/find.rs b/crates/lib/database_queries/src/twitch_oauth/find.rs index 4785a33d98..1f544a8243 100644 --- a/crates/lib/database_queries/src/twitch_oauth/find.rs +++ b/crates/lib/database_queries/src/twitch_oauth/find.rs @@ -25,6 +25,7 @@ pub struct TwitchOauthTokenRecord { pub maybe_refresh_token: Option, pub token_type: Option, pub expires_in_seconds: Option, + pub refresh_count: u32, pub has_bits_read: bool, pub has_channel_read_subscriptions: bool, pub has_channel_read_redemptions: bool, @@ -99,6 +100,7 @@ impl TwitchOauthTokenFinder { maybe_refresh_token: record.maybe_refresh_token.clone(), token_type: record.token_type.clone(), expires_in_seconds: record.expires_in_seconds.clone(), + refresh_count: record.refresh_count, has_bits_read: i8_to_bool(record.has_bits_read), has_channel_read_subscriptions: i8_to_bool(record.has_channel_read_subscriptions), has_channel_read_redemptions: i8_to_bool(record.has_channel_read_redemptions), @@ -167,6 +169,7 @@ SELECT twitch_oauth_tokens.maybe_refresh_token, twitch_oauth_tokens.token_type, twitch_oauth_tokens.expires_in_seconds, + twitch_oauth_tokens.refresh_count, twitch_oauth_tokens.has_bits_read, twitch_oauth_tokens.has_channel_read_subscriptions, twitch_oauth_tokens.has_channel_read_redemptions, @@ -256,6 +259,7 @@ pub struct TwitchOauthTokenRecordInternal { pub maybe_refresh_token: Option, pub token_type: Option, pub expires_in_seconds: Option, + pub refresh_count: u32, pub has_bits_read: i8, pub has_channel_read_subscriptions: i8, pub has_channel_read_redemptions: i8, diff --git a/crates/lib/database_queries/src/twitch_oauth/insert.rs b/crates/lib/database_queries/src/twitch_oauth/insert.rs index 85340cc875..e40458a6eb 100644 --- a/crates/lib/database_queries/src/twitch_oauth/insert.rs +++ b/crates/lib/database_queries/src/twitch_oauth/insert.rs @@ -35,6 +35,9 @@ pub struct TwitchOauthTokenInsertBuilder { /// TTL seconds from issue expires_in_seconds: Option, + /// Number of times we've refreshed the token + refresh_count: u32, + has_bits_read: bool, has_channel_read_subscriptions: bool, has_channel_read_redemptions: bool, @@ -55,7 +58,8 @@ impl TwitchOauthTokenInsertBuilder { has_bits_read: false, has_channel_read_subscriptions: false, has_channel_read_redemptions: false, - has_user_read_follows: false + has_user_read_follows: false, + refresh_count: 0, } } @@ -84,6 +88,11 @@ impl TwitchOauthTokenInsertBuilder { self } + pub fn set_refresh_count(mut self, refresh_count: u32) -> Self { + self.refresh_count = refresh_count; + self + } + pub fn set_has_bits_read(mut self, has_bits_read: bool) -> Self { self.has_bits_read = has_bits_read; self 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 7ca88cb515..5fd13eb7b4 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 @@ -1,13 +1,15 @@ use anyhow::{anyhow, Error}; use container_common::anyhow_result::AnyhowResult; use container_common::thread::thread_id::ThreadId; -use crate::redis::lease_payload::LeasePayload; use crate::redis::constants::{LEASE_TIMEOUT_SECONDS, LEASE_RENEW_PERIOD, LEASE_CHECK_PERIOD, OBS_ACTIVE_CHECK_PERIOD}; +use crate::redis::lease_payload::LeasePayload; use crate::twitch::constants::TWITCH_PING_CADENCE; +use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher; 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 database_queries::twitch_oauth::insert::TwitchOauthTokenInsertBuilder; use log::debug; use log::error; use log::info; @@ -22,8 +24,6 @@ use std::thread::sleep; use std::time::Duration; use time::Instant; use twitch_api2::pubsub::Response; -use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher; -use database_queries::twitch_oauth::insert::TwitchOauthTokenInsertBuilder; // TODO: Publish events back to OBS thread // TODO: (cleanup) make the logic clearer to follow. @@ -344,7 +344,6 @@ impl TwitchPubsubUserSubscriberThreadStageTwo { } fn renew_redis_lease(&mut self) -> AnyhowResult<()> { - // TODO: Error handling let mut redis = self.redis_pool.get()?; let lease_key = RedisKeys::twitch_pubsub_lease(self.twitch_user_id.get_str()); @@ -381,7 +380,6 @@ impl TwitchPubsubUserSubscriberThreadStageTwo { } fn check_obs_session_active(&mut self) -> AnyhowResult { - // TODO: Error handling let mut redis = self.redis_pool.get()?; let key = RedisKeys::obs_active_session_keepalive(self.twitch_user_id.get_str()); @@ -404,7 +402,6 @@ impl TwitchPubsubUserSubscriberThreadStageTwo { }, }; - // TODO: Move this somewhere common let refresh_result = self.oauth_token_refresher.refresh_token(refresh_token) .await?; @@ -413,7 +410,7 @@ impl TwitchPubsubUserSubscriberThreadStageTwo { .map(|t| t.secret().to_string()); let expires_seconds = refresh_result.duration.as_secs() as u32; - // TODO: Move this somewhere common + // TODO: Move saving a refreshed record somewhere common let mut query_builder = TwitchOauthTokenInsertBuilder::new( &self.twitch_oauth_token_record.twitch_user_id, &self.twitch_oauth_token_record.twitch_username, @@ -421,6 +418,7 @@ impl TwitchPubsubUserSubscriberThreadStageTwo { .set_refresh_token(refresh_token.as_deref()) .set_user_token(self.twitch_oauth_token_record.maybe_user_token.as_deref()) .set_expires_in_seconds(Some(expires_seconds)) + .set_refresh_count(self.twitch_oauth_token_record.refresh_count.saturating_add(1)) // NB: We don't get these back from the refresh, but it seems like they would stay the same. .set_token_type(self.twitch_oauth_token_record.token_type.as_deref()) .set_has_bits_read(self.twitch_oauth_token_record.has_bits_read)