actor: primitive Bluesky Sink implementation.

This is working somewhat well, but we definitely
need to clean up the entrypoint and make each
sink adequately configurable.

Cargo.lock

···
 version = "0.1.0"
 dependencies = [
  "bsky-sdk",
+ "chrono",
  "cron-lite",
  "futures",
  "glob",

Cargo.toml

···
-cargo-features = ["edition2024"] # For rust-analyzer to work
-
 [package]
 name = "audquotes"
 version = "0.1.0"
···
 
 [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"

src/lib.rs

···
 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<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 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 = {
···
             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<dyn std::error::Error>> {
+                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(())
     }
 }

src/sink.rs

···
 use crate::data::Quote;
+use bsky_sdk::{BskyAgent, api::types::Object};
 use kameo::prelude::*;
 
 /// A newtype over [Quote] used to prompt the [SinkManager] to
···
     }
 }
 
+/// A [QuoteSink] which will post the contents of each quote to Bluesky.
+#[derive(Actor)]
+pub struct BskySink {
+    bsky_agent: BskyAgent,
+    bsky_session: Object<bsky_sdk::api::com::atproto::server::create_session::OutputData>,
+}
+
+impl BskySink {
+    pub async fn new_session(username: String, password: String) -> Result<Self, ()> {
+        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<PostQuote> for BskySink {
+    type Reply = PostResult;
+
+    async fn handle(
+        &mut self,
+        PostQuote(quote): PostQuote,
+        _ctx: &mut Context<Self, Self::Reply>,
+    ) -> 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
···
     // do asynchronous dynamic dispatch for it here.
     // I've decided I'll limit this to one sink per implementation right now.
     stdout_sink: Option<ActorRef<StdoutSink>>,
+    bsky_sink: Option<ActorRef<BskySink>>,
     // ...
 }
 
 impl SinkManager {
-    pub fn new(stdout_sink: Option<ActorRef<StdoutSink>>) -> Self {
-        Self { stdout_sink }
+    pub fn new(
+        stdout_sink: Option<ActorRef<StdoutSink>>,
+        bsky_sink: Option<ActorRef<BskySink>>,
+    ) -> Self {
+        Self {
+            stdout_sink,
+            bsky_sink,
+        }
     }
 }
 
···
     ) -> 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()
···
         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 {