diff --git a/Cargo.lock b/Cargo.lock index e4da3dd..c21dcd5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -162,6 +162,15 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" +[[package]] +name = "backon" +version = "1.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "592277618714fbcecda9a02ba7a8781f319d26532a88553bbacc77ba5d2b3a8d" +dependencies = [ + "fastrand", +] + [[package]] name = "backtrace" version = "0.3.74" @@ -1467,8 +1476,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8034fb926579ff49d3fe58d288d5dcb580bf11e9bccd33224b45adebf0fd0c23" dependencies = [ "arc-swap", + "backon", "bytes", "combine", + "futures-channel", "futures-util", "itoa", "num-bigint", diff --git a/Cargo.toml b/Cargo.toml index 5312bfd..2450e03 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,6 @@ bsky-sdk = "0.1.16" glob = "0.3.2" grep = "0.3.2" rand = "0.9.0" -redis = { version = "0.29.1", features = ["aio", "tokio-comp"] } +redis = { version = "0.29.1", features = ["aio", "connection-manager", "tokio-comp"] } tokio = { version = "1.44.0", features = ["full"] } tokio-cron-scheduler = "0.13.0" diff --git a/config.toml b/config.toml new file mode 100644 index 0000000..95ada90 --- /dev/null +++ b/config.toml @@ -0,0 +1,14 @@ +[quotes] +default = { path = "quotes/**/*.txt", content = ".*", dates = [] } + +[quotes.father] +path = "quotes/**/*.txt" +content = "\b(?i:father|dad|daddy|papa|pops)\b" +dates = [ + # whichever date is Father's day +] + +[quotes.mother] +path = "quotes/**/*.txt" +content = "\b(?i:mother|mommy|mama|mom)\b" +dates = [] diff --git a/src/main.rs b/src/main.rs index ddd5971..1b5fc19 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,13 @@ -use bsky_sdk::BskyAgent; +use bsky_sdk::api::app; use bsky_sdk::api::app::bsky::feed::post; use bsky_sdk::api::types::string::Datetime; +use bsky_sdk::{BskyAgent, api::types::Object}; use glob::glob; use grep::{matcher::Matcher, regex, searcher::sinks}; use rand::random_range; use rand::seq::SliceRandom; +use redis::aio::ConnectionManagerConfig; use std::{sync::Arc, time::Duration}; use tokio::sync::Mutex; @@ -21,7 +23,7 @@ const EVENT_QUEUE: &str = "queue:event"; // See https://cron.help for what these strings mean const POSTING_INTERVAL_CRON: &str = "0 0,30 * * * *"; -const POSTING_INTERVAL_DEBUG: &str = "0,30 * * * * *"; +const POSTING_INTERVAL_DEBUG: &str = "1/10 * * * * *"; const EVENT_UPDATE_INTERVAL: &str = "55 23 * * *"; const POSTING_RETRIES: i32 = 5; @@ -40,165 +42,256 @@ fn prepare_post>(text: I) -> post::RecordData { } } +#[derive(Clone, Debug)] struct QuoteFilter { path: String, content: String, dates: Vec, } -fn read_files(filter: &QuoteFilter) -> Vec { - let matcher = regex::RegexMatcher::new(&filter.content).unwrap(); - let mut searcher = grep::searcher::Searcher::new(); - let mut results = Vec::new(); - - for file in glob(&filter.path).unwrap() { - let file = match file { - Ok(file) => file, - Err(_) => continue, - }; - - let mut matched = false; - let sink = sinks::Lossy(|_lnum, _line| { - matched = true; - Ok(false) - }); +impl QuoteFilter { + pub async fn get_quote( + &self, + mut con: impl redis::aio::ConnectionLike + AsyncCommands + Clone, + ) -> Result { + // 1: Attempt to read from the event (priority) queue + let event_quote: Option = con.lpop(EVENT_QUEUE, None).await.ok(); + if let Some(quote) = event_quote { + return Ok(quote); + } + + // 2: Otherwise, we read from the regular queue, repopulating it if it's empty + self.reshuffle_quotes(con.clone(), DEFAULT_QUEUE).await?; + con.lpop(DEFAULT_QUEUE, None).await.map_err(|_| ()) + } + + async fn reshuffle_quotes( + &self, + mut con: impl redis::aio::ConnectionLike + AsyncCommands, + output_queue: &str, + ) -> Result<(), ()> { + let len: u64 = con.llen(output_queue).await.map_err(|_| ())?; + // NOTE: The following assumes the queue hasn't been repopulated by any other client + // in-between the call to llen and the execution of the pipeline. + // Hopefully won't be a problem :) + if len == 0 { + let mut file_contents = self.read_files(); + + { + let mut rand = rand::rng(); + file_contents.shuffle(&mut rand); + } + + let mut pipeline = redis::pipe(); + for file_contents in file_contents.into_iter() { + pipeline.lpush(output_queue, file_contents.as_str()); + } + let _: () = pipeline.query_async(&mut con).await.map_err(|_| ())?; + } + + Ok(()) + } - let search_result = searcher.search_path(&matcher, &file, sink); - if !matched || search_result.is_err() { - continue; + fn read_files(&self) -> Vec { + let matcher = regex::RegexMatcher::new(&self.content).unwrap(); + let mut searcher = grep::searcher::Searcher::new(); + let mut results = Vec::new(); + + for file in glob(&self.path).unwrap() { + let file = match file { + Ok(file) => file, + Err(_) => continue, + }; + + let mut matched = false; + let sink = sinks::Lossy(|_lnum, _line| { + matched = true; + Ok(false) + }); + + let search_result = searcher.search_path(&matcher, &file, sink); + if !matched || search_result.is_err() { + continue; + } + + let contents = std::fs::read_to_string(file).unwrap(); + results.push(contents.trim().to_string()); } - let contents = std::fs::read_to_string(file).unwrap(); - results.push(contents.trim().to_string()); + results } +} - results +#[derive(Clone)] +struct RedisState { + con_manager: redis::aio::ConnectionManager, } -async fn reshuffle_quotes( - filter: &QuoteFilter, - mut con: impl redis::aio::ConnectionLike + AsyncCommands, - output_queue: &str, -) -> Result<(), ()> { - let len: u64 = con.llen(output_queue).await.map_err(|_| ())?; - // NOTE: The following assumes the queue hasn't been repopulated by any other client - // in-between the call to llen and the execution of the pipeline. - // Hopefully won't be a problem :) - if len == 0 { - let mut file_contents = read_files(filter); - - { - let mut rand = rand::rng(); - file_contents.shuffle(&mut rand); +impl RedisState { + pub async fn new(url: String) -> Result { + let redis = redis::Client::open(url).map_err(|_| ())?; + let config = ConnectionManagerConfig::new() + .set_response_timeout(std::time::Duration::from_secs(10)) + .set_number_of_retries(3); + let con_manager = redis::aio::ConnectionManager::new_with_config(redis, config) + .await + .map_err(|_| ())?; + + Ok(RedisState { con_manager }) + } + + pub async fn fetch_quote(&self, filter: &QuoteFilter) -> Result { + loop { + match filter.get_quote(self.con_manager.clone()).await { + Ok(text) => return Ok(text), + Err(_) => eprintln!("Error fetching quote from redis storage. Retrying..."), + }; } + } +} + +#[derive(Clone)] +struct BlueskyState { + bsky_agent: BskyAgent, + bsky_session: Object, +} + +impl BlueskyState { + pub async fn new_session(username: String, password: String) -> Result { + let agent = BskyAgent::builder().build().await.map_err(|_| ())?; + let session = agent.login(username, password).await.map_err(|_| ())?; + + Ok(Self { + bsky_agent: agent, + bsky_session: session, + }) + } + + pub async fn submit_post(self, post: String) -> Result<(), ()> { + let post = prepare_post(post.as_str()); - let mut pipeline = redis::pipe(); - for file_contents in file_contents.into_iter() { - pipeline.lpush(output_queue, file_contents.as_str()); + for current_try in 0..POSTING_RETRIES { + if let Err(e) = self.bsky_agent.create_record(post.clone()).await { + eprintln!("Could not post quote: `{e}`"); + eprintln!("Attempting to refresh login..."); + + if let Err(e) = self + .bsky_agent + .resume_session(self.bsky_session.clone()) + .await + { + eprintln!("Failed to resume sessions due to following error: {e}") + } + } else { + if current_try > 0 { + eprintln!("Successfully posted quote on retry #{current_try}"); + } + return Ok(()); + } } - let _: () = pipeline.query_async(&mut con).await.map_err(|_| ())?; + + Err(()) } +} - Ok(()) +#[derive(Clone)] +struct State { + redis: RedisState, + bsky_session: Option, } -async fn get_quote( - filter: &QuoteFilter, - mut con: impl redis::aio::ConnectionLike + AsyncCommands + Clone, -) -> Result { - // 1: Attempt to read from the event (priority) queue - let event_quote: Option = con.lpop(EVENT_QUEUE, None).await.ok(); - if let Some(quote) = event_quote { - return Ok(quote); +impl State { + pub fn redis(&self) -> &RedisState { + &self.redis } - // 2: Otherwise, we read from the regular queue, repopulating it if it's empty - reshuffle_quotes(filter, con.clone(), DEFAULT_QUEUE).await?; - con.lpop(DEFAULT_QUEUE, None).await.map_err(|_| ()) + pub fn bsky(&self) -> Option<&BlueskyState> { + self.bsky_session.as_ref() + } } #[tokio::main] async fn main() -> Result<(), Box> { - let redis = - redis::Client::open(std::env::var("REDIS_URL").unwrap_or("redis://localhost".to_string()))?; - let con = redis.get_multiplexed_async_connection().await?; - let debug_mode = std::env::var("DEBUG").unwrap_or("0".to_string()) == "1"; - - let (agent, session) = if !debug_mode { - let agent = BskyAgent::builder().build().await?; - let session = agent - .login( - std::env::var("BLUESKY_USERNAME").unwrap_or_default(), - std::env::var("BLUESKY_PASSWORD").unwrap_or_default(), + let use_bsky = std::env::var("USE_BLUESKY").unwrap_or("0".to_string()) == "1"; + + let redis_state = + RedisState::new(std::env::var("REDIS_URL").unwrap_or("redis://localhost".to_string())) + .await + .expect("Initial redis connection failure"); + let bsky_state = if use_bsky { + Some( + BlueskyState::new_session( + std::env::var("BLUESKY_USERNAME").expect("Bluesky username not supplied"), + std::env::var("BLUESKY_PASSWORD") + .expect("Bluesky application password not supplied"), ) - .await?; - - (Some(Arc::new(Mutex::new(agent))), Some(session)) + .await + .expect("Could not connect to Bluesky with supplied credentials"), + ) } else { - (None, None) // Let's just simulate what the bot would post + None }; + let app_state = Arc::new(State { + redis: redis_state, + bsky_session: bsky_state, + }); + let sched = JobScheduler::new().await?; /* let event_filter = Arc::new(QuoteFilter { content: r"\b(?i:mother|mommy|mama|mom)\b".to_string(), - path: "test/**/ -*.txt".to_string(), + path: "test/**/*.txt".to_string(), dates: vec![], }); */ let regular_filter = Arc::new(QuoteFilter { content: r".*".to_string(), - path: if !debug_mode { "quotes/**/*.txt".to_string() } else { "test/**/*.txt".to_string() }, + path: if !debug_mode { + "quotes/**/*.txt".to_string() + } else { + "test/**/*.txt".to_string() + }, dates: vec![], }); - let (con_poster, con_event_monitor) = (con.clone(), con.clone()); - let (agent_poster, agent_event_monitor) = (agent.clone(), agent.clone()); - let posting_interval = if !debug_mode { POSTING_INTERVAL_CRON } else { POSTING_INTERVAL_DEBUG }; - // Add async job - sched - .add(Job::new_async(posting_interval, move |_uuid, _| { - let filter = regular_filter.clone(); - let con = con_poster.clone(); - let agent = agent_poster.clone(); - let session = session.clone(); - - Box::pin(async move { - let text: String = get_quote(&filter, con).await.unwrap(); - - if let (Some(agent), Some(session)) = (agent, session) { - let post = prepare_post(text.as_str()); - let agent = agent.lock().await; - - for _ in 0..POSTING_RETRIES { - if let Err(e) = agent.create_record(post.clone()).await { - eprintln!("Could not post quote: `{e}`"); - eprintln!("Attempting to refresh login..."); - - if let Err(e) = agent.resume_session(session.clone()).await { - eprintln!("Failed to resume sessions due to following error: {e}") - } - } else { - break; - } - } - } else { - // Let's just print the quote! - println!("{}\n", text); + let post_job = Job::new_async(posting_interval, move |_uuid, _| { + let filter = regular_filter.clone(); + let app_state = app_state.clone(); + + Box::pin(async move { + // We try fetching a new quote from our redis storage until we succeed + let text = match app_state.redis().fetch_quote(&filter).await { + Ok(text) => text, + Err(_) => { + eprintln!("Error fetching quote from redis storage."); + return; } - }) - })?) - .await?; + }; + + if let Some(bsky) = app_state.bsky() { + if let Err(_) = bsky.clone().submit_post(text).await { + eprintln!("Error posting to bluesky."); + return; + } + } else { + // Let's just print the quote! + println!("{}\n", text); + } + }) + })?; + + // Add async job + sched.add(post_job).await?; // sched // .add(Job::new_async(EVENT_UPDATE_INTERVAL, move |_uuid, _| { @@ -215,7 +308,10 @@ async fn main() -> Result<(), Box> { // })?) // .await?; - sched.start().await.unwrap(); + sched + .start() + .await + .expect("Error starting tokio scheduler. Shutting down..."); loop { tokio::time::sleep(Duration::from_secs(10)).await; }