more pubsub

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

@@ -1,22 +1,24 @@
 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;
 use log::error;
 use log::info;
 use log::warn;
 use r2d2_redis::RedisConnectionManager;
-use r2d2_redis::r2d2::Pool;
+use r2d2_redis::r2d2;
 use r2d2_redis::redis::Commands;
 use redis_common::redis_keys::RedisKeys;
+use sqlx::MySql;
 use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
 use tokio::runtime::Runtime;
-use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS;
-use crate::threads::twitch_pubsub_user_subscriber_thread::twitch_pubsub_user_subscriber_thread;
 
 pub async fn listen_for_active_obs_session_thread(
-  redis_pool: Arc<Pool<RedisConnectionManager>>,
-  redis_pubsub_pool: Arc<Pool<RedisConnectionManager>>,
+  mysql_pool: Arc<sqlx::Pool<MySql>>,
+  redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
+  redis_pubsub_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
   runtime: Arc<Runtime>,
 ) {
   // TODO: ERROR HANDLING
@@ -73,8 +75,9 @@ pub async fn listen_for_active_obs_session_thread(
 
     let twitch_user_id = payload.twitch_user_id.clone();
     let redis_pool2 = redis_pool.clone();
+    let mysql_pool2 = mysql_pool.clone();
 
-    runtime.spawn(twitch_pubsub_user_subscriber_thread(twitch_user_id, redis_pool2));
+    runtime.spawn(twitch_pubsub_user_subscriber_thread(twitch_user_id, mysql_pool2, redis_pool2));
 
 
     // Publish: (ActiveSession, user_id)

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

@@ -1,21 +1,27 @@
 use crate::redis::lease_payload::LeasePayload;
+use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS;
+use crate::twitch::pubsub::build_pubsub_topics_for_user::build_pubsub_topics_for_user;
+use crate::twitch::websocket_client::TwitchWebsocketClient;
+use database_queries::twitch_oauth::find::{TwitchOauthTokenRecord, TwitchOauthTokenFinder};
 use log::info;
 use log::warn;
-use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS;
 use r2d2_redis::RedisConnectionManager;
-use r2d2_redis::r2d2::Pool;
+use r2d2_redis::r2d2;
 use r2d2_redis::redis::Commands;
+use redis_common::redis_keys::RedisKeys;
+use sqlx::MySql;
 use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
-use redis_common::redis_keys::RedisKeys;
 
 pub async fn twitch_pubsub_user_subscriber_thread(
   twitch_user_id: String,
-  redis_pool: Arc<Pool<RedisConnectionManager>>
+  mysql_pool: Arc<sqlx::Pool<MySql>>,
+  redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
 ) {
 
   // TODO: Error handling
+  let numeric_twitch_user_id = twitch_user_id.parse::<u32>().unwrap();
 
   loop {
     info!("Twitch subscriber thread");
@@ -32,7 +38,39 @@ pub async fn twitch_pubsub_user_subscriber_thread(
       LEASE_TIMEOUT_SECONDS
     ).unwrap();
 
+    let result = TwitchOauthTokenFinder::new()
+        .scope_twitch_user_id(Some(numeric_twitch_user_id))
+        .perform_query(&mysql_pool)
+        .await
+        .unwrap();
 
-    sleep(Duration::from_secs(3));
+    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();
+
+    warn!("Sending ping...");
+    client.send_ping().await.unwrap();
+
+    warn!("Listen to user...");
+    let topics = build_pubsub_topics_for_user(numeric_twitch_user_id);
+
+    client.listen(&record.access_token, &topics).await.unwrap();
+
+
+    sleep(Duration::from_secs(30));
   }
 }

crates/service/obs_gateway/src/twitch_pubsub_job.rs

@@ -12,7 +12,7 @@
 #[macro_use] extern crate magic_crypt;
 #[macro_use] extern crate serde_derive;
 
-use config::shared_constants::{DEFAULT_RUST_LOG, DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING};
+use config::shared_constants::{DEFAULT_RUST_LOG, DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING, DEFAULT_MYSQL_CONNECTION_STRING};
 use container_common::anyhow_result::AnyhowResult;
 use futures::executor::{ThreadPool, ThreadPoolBuilder};
 use r2d2_redis::r2d2;
@@ -28,6 +28,7 @@ use std::sync::Arc;
 use r2d2_redis::r2d2::Pool;
 use redis_common::redis_keys::RedisKeys;
 use crate::threads::listen_for_active_obs_sessions_thread::listen_for_active_obs_session_thread;
+use sqlx::mysql::MySqlPoolOptions;
 
 pub mod redis;
 pub mod threads;
@@ -38,11 +39,25 @@ pub mod util;
 pub async fn main() -> AnyhowResult<()> {
   easyenv::init_all_with_default_logging(Some(DEFAULT_RUST_LOG));
 
+  let db_connection_string =
+      easyenv::get_env_string_or_default(
+        "MYSQL_URL",
+        DEFAULT_MYSQL_CONNECTION_STRING);
+
   let redis_connection_string =
       easyenv::get_env_string_or_default(
         "REDIS_1_URL",
         DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING);
 
+  info!("Connecting to mysql...");
+
+  let mysql_pool = MySqlPoolOptions::new()
+      .max_connections(5)
+      .connect(&db_connection_string)
+      .await?;
+
+  let mysql_pool = Arc::new(mysql_pool);
+
   info!("Connecting to redis...");
 
   let redis_manager = RedisConnectionManager::new(redis_connection_string.clone())?;
@@ -96,7 +111,7 @@ pub async fn main() -> AnyhowResult<()> {
   //runtime.spawn(watch_user_thread(10));
   //runtime.spawn(watch_user_thread(9999));
   runtime.spawn(listen_for_active_obs_session_thread(
-    redis_pool, redis_pubsub_pool, runtime_2));
+    mysql_pool, redis_pool, redis_pubsub_pool, runtime_2));
 
   loop {
     sleep(Duration::from_millis(10_000));