Something went wrong. Try again.
Source code for my personal quote bot project.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133pub mod data;pub mod sink;pub mod storage;
pub mod run { use std::time::Duration;
use crate::data::Handle; use crate::sink::{BskySink, DiscordSink, PostQuote, SinkManager, StdoutSink}; use crate::storage::queue::QueueManager; use crate::storage::source::{QuoteFilter, SourceManager}; use crate::storage::{ FetchQuote, QuoteCycle, queue::MemoryQueueStorage, queue::RedisQueueStorage, source::FsFilterSourceManager, }; use cron_lite::CronEvent; use futures::StreamExt; use tokio::time::timeout;
pub async fn entrypoint() -> Result<(), Box<dyn std::error::Error>> { // 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 mut sinks = SinkManager::new();
let use_bsky = std::env::var("USE_BLUESKY").unwrap_or("0".to_string()) == "1"; if use_bsky { sinks = sinks.add_sink( BskySink::new_from_env() .await .expect("Could not connect to Bluesky with supplied credentials"), ); }
let use_discord = std::env::var("USE_DISCORD").unwrap_or("0".to_string()) == "1"; if use_discord { sinks = sinks.add_sink(DiscordSink::new_from_env()); }
// If we're in "debug" mode, posts become more frequent. let is_debug = std::env::var("DEBUG").unwrap_or("0".to_string()) == "1"; let schedule: &str = if !is_debug { "0 */30 * * * * *" } else { // Debug schedule "*/10 * * * * * *" }; let source = if !is_debug { FsFilterSourceManager::new(QuoteFilter::new_from_glob("quotes/**/*.txt".to_string())) } else { // Debug quotes fall back to the default configuration FsFilterSourceManager::default() };
if is_debug { sinks = sinks.add_sink(StdoutSink); };
let use_redis = std::env::var("USE_REDIS").unwrap_or("0".to_string()) == "1"; if use_redis { let queue = RedisQueueStorage::new_from_env().unwrap(); run_cycle(schedule, source, queue, sinks).await } else { let queue = MemoryQueueStorage::new(); run_cycle(schedule, source, queue, sinks).await } }
async fn run_cycle<S: SourceManager, Q: QueueManager>( schedule: &str, source: S, queue: Q, mut sink: SinkManager, ) -> Result<(), Box<dyn std::error::Error>> { let mut cycle = { QuoteCycle::with_thread_rng(source, queue) };
use cron_lite::Schedule; const POSTING_TIMEOUT: Duration = Duration::from_secs(60); let schedule = Schedule::new(schedule).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; }
use tracing::Instrument; let post_span = tracing::debug_span!("post_cycle", time = chrono::Utc::now().to_string());
// 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 mut next_post_iteration = async || -> Result<(), Box<dyn std::error::Error>> { tracing::debug!("Fetching next quote"); let next_quote = cycle .handle(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. tracing::debug!( quote = next_quote.get().to_string(), "Forwarding quote to sink" ); sink.handle(PostQuote(next_quote)).await?; println!();
Ok(()) };
if let Err(e) = timeout(POSTING_TIMEOUT, next_post_iteration().instrument(post_span)).await { tracing::error!( error = e.to_string(), "Could not submit post in time to all sinks.", ); } }
Ok(()) }}