diff --git a/Cargo.lock b/Cargo.lock index 1c06c3d..12f038a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -142,6 +142,7 @@ name = "audquotes" version = "0.1.0" dependencies = [ "bsky-sdk", + "chrono", "cron-lite", "futures", "glob", diff --git a/Cargo.toml b/Cargo.toml index 2993fbc..bf57551 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,3 @@ -cargo-features = ["edition2024"] # For rust-analyzer to work - [package] name = "audquotes" version = "0.1.0" @@ -8,6 +6,7 @@ rust-version = "1.85" [dependencies] bsky-sdk = "0.1.16" +chrono = "0.4.42" cron-lite = { version = "0.3.0", features = ["async"] } futures = "0.3.31" glob = "0.3.2" diff --git a/src/lib.rs b/src/lib.rs index 2586603..89931de 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,20 +3,39 @@ pub mod sink; pub mod storage; pub mod run { - use crate::sink::{PostQuote, SinkManager, StdoutSink}; + use std::time::Duration; + + use crate::sink::{BskySink, PostQuote, SinkManager, StdoutSink}; use crate::storage::{ FetchQuote, QuoteCycle, queue::MemoryQueueStorage, source::FsFilterSourceManager, }; + use cron_lite::CronEvent; + use futures::StreamExt; use kameo::prelude::*; + use tokio::time::timeout; pub async fn entrypoint() -> Result<(), Box> { // TODO: Clean up this function's internals. // The current structure is alright, but it was stitched together // quickly just to confirm that everything is functioning as it should. + let use_bsky = std::env::var("USE_BLUESKY").unwrap_or("0".to_string()) == "1"; + let bsky = if use_bsky { + Some(BskySink::spawn( + BskySink::new_session( + std::env::var("BLUESKY_USERNAME").expect("Bluesky username not supplied"), + std::env::var("BLUESKY_PASSWORD") + .expect("Bluesky application password not supplied"), + ) + .await + .expect("Could not connect to Bluesky with supplied credentials"), + )) + } else { + None + }; let sink = { let stdout = StdoutSink::spawn(StdoutSink); - SinkManager::spawn(SinkManager::new(Some(stdout))) + SinkManager::spawn(SinkManager::new(Some(stdout), bsky)) }; let cycle = { @@ -26,14 +45,50 @@ pub mod run { QuoteCycle::spawn(QuoteCycle::with_thread_rng(source, queue)) }; - loop { - let next_quote = cycle - .ask(FetchQuote) - .await - .map_err(|_| "fetch quote should always succeed")?; - sink.tell(PostQuote(next_quote)).await?; - tokio::time::sleep(std::time::Duration::from_secs(3)).await; - println!() + use cron_lite::Schedule; + const POSTING_TIMEOUT: Duration = Duration::from_secs(60); + const POSTING_INTERVAL: &str = "*/10 * * * * * *"; + let schedule = + Schedule::new(POSTING_INTERVAL).expect("Schedule should be a valid cron expression"); + let now = chrono::Utc::now(); + + let mut tick_stream = schedule.stream(&now); + + while let Some(tick) = tick_stream.next().await { + if let CronEvent::Missed(missed_at) = tick { + eprintln!( + "Missed event tick at {}. Current time: {}. Skipping post.", + missed_at, + chrono::Utc::now() + ); + continue; + } + + // We store the code to perform the next posting iteration as one atomic future which we wrap with a timeout. + // This means that, if we miss a posting window due to the timeout, we will not get multiple consecutive or late posts. + let next_post_iteration = async || -> Result<(), Box> { + let next_quote = cycle + .ask(FetchQuote) + .await + .map_err(|_| "fetch quote should always succeed")?; + + // Note: By using `tell`, we don't know when each sink's code will have completed. + // If any sink uses, say, a file or stdout, that resource may well be contested between + // consecutive iterations of this loop. + sink.tell(PostQuote(next_quote)).await?; + println!(); + + Ok(()) + }; + + if let Err(e) = timeout(POSTING_TIMEOUT, next_post_iteration()).await { + eprintln!( + "Could not submit post in time to all sinks. Timeout error: {}", + e + ); + } } + + Ok(()) } } diff --git a/src/sink.rs b/src/sink.rs index d0898de..92711c9 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -1,4 +1,5 @@ use crate::data::Quote; +use bsky_sdk::{BskyAgent, api::types::Object}; use kameo::prelude::*; /// A newtype over [Quote] used to prompt the [SinkManager] to @@ -52,6 +53,68 @@ impl Message for StdoutSink { } } +/// A [QuoteSink] which will post the contents of each quote to Bluesky. +#[derive(Actor)] +pub struct BskySink { + bsky_agent: BskyAgent, + bsky_session: Object, +} + +impl BskySink { + 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, + }) + } + + async fn submit_post(&mut self, quote: Quote) -> Result<(), ()> { + let post = bsky_sdk::api::app::bsky::feed::post::RecordData { + text: quote.into(), + created_at: bsky_sdk::api::types::string::Datetime::now(), + embed: None, + entities: None, + facets: None, + labels: None, + langs: None, + reply: None, + tags: None, + }; + + if let Err(e) = self + .bsky_agent + .resume_session(self.bsky_session.clone()) + .await + { + eprintln!("Failed to resume sessions due to following error: {e}"); + return Err(()); + } + + match self.bsky_agent.create_record(post.clone()).await { + Ok(_) => Ok(()), + Err(_) => Err(()), + } + } +} + +impl Message for BskySink { + type Reply = PostResult; + + async fn handle( + &mut self, + PostQuote(quote): PostQuote, + _ctx: &mut Context, + ) -> Self::Reply { + match self.submit_post(quote).await { + Ok(_) => Ok(()), + Err(_) => Err(PostFailure::Unrecoverable), + } + } +} + /// Supervises all [QuoteSink] actors within the program, forwarding /// [PostQuote] messages to them as they are received. /// The SinkManager will attempt to reinitialize failed sinks upon @@ -63,12 +126,19 @@ pub struct SinkManager { // do asynchronous dynamic dispatch for it here. // I've decided I'll limit this to one sink per implementation right now. stdout_sink: Option>, + bsky_sink: Option>, // ... } impl SinkManager { - pub fn new(stdout_sink: Option>) -> Self { - Self { stdout_sink } + pub fn new( + stdout_sink: Option>, + bsky_sink: Option>, + ) -> Self { + Self { + stdout_sink, + bsky_sink, + } } } @@ -84,12 +154,17 @@ impl Message for SinkManager { ) -> Self::Reply { use futures::future::join_all; - // We'll see if this monstrosity actually works - let sinks = [self.stdout_sink.clone()]; - let futures = sinks - .iter() - .flatten() + let stdout_result = self + .stdout_sink + .as_ref() .map(|s| s.ask(msg.clone()).into_future()); + + let bsky_result = self + .bsky_sink + .as_ref() + .map(|s| s.ask(msg.clone()).into_future()); + + let futures = [stdout_result, bsky_result].into_iter().flatten(); let results = join_all(futures).await; results.iter().map(|r| r.clone().or(Err(()))).collect() @@ -102,7 +177,7 @@ mod test { use super::*; let stdout = StdoutSink::spawn(StdoutSink); - let manager = SinkManager::spawn(SinkManager::new(Some(stdout))); + let manager = SinkManager::spawn(SinkManager::new(Some(stdout), None)); let messages = ["First test!", "Second test.", "Third..."]; for msg in messages {