use crate::{AmselBot, BskyFeed, BskyRecordRef, CursorStore, Result, record::post::BskyPost};
use anyhow::Context;
use bsky_sdk::{BskyAgent, agent::config::Config, api::agent::atp_agent::AtpSession};
use jetstream_oxide::{
DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,
events::commit::CommitEvent,
};
pub struct BskySession {
#[allow(dead_code)]
session: AtpSession,
pub agent: BskyAgent,
}
impl BskySession {
pub async fn new_from_env() -> Result<BskySession> {
if let Err(err) = dotenvy::dotenv() {
if !err.not_found() {
return Err(err.into());
}
}
BskySession::new(
&std::env::var("AMSEL_BOT_ENDPOINT").with_context(|| "AMSEL_BOT_ENDPOINT")?,
&std::env::var("AMSEL_BOT_HANDLE").with_context(|| "AMSEL_BOT_HANDLE")?,
&std::env::var("AMSEL_BOT_PASSWORD").with_context(|| "AMSEL_BOT_PASSWORD")?,
)
.await
}
pub async fn new(endpoint: &str, identifier: &str, password: &str) -> Result<BskySession> {
let agent = BskyAgent::builder()
.config(Config {
endpoint: endpoint.to_string(),
session: None,
labelers_header: None,
proxy_header: None,
})
.build()
.await?;
let session = agent
.login(identifier, password)
.await
.map_err(anyhow::Error::from)?;
Ok(Self { session, agent })
}
pub async fn run_with_jetstream_config<T: AmselBot + CursorStore>(
&self,
mut bot: T,
config: JetstreamConfig,
) -> Result<()> {
self.run_inner(&mut bot, config).await
}
pub async fn run<T: AmselBot + CursorStore>(&self, mut bot: T) -> Result<()> {
let default_config = JetstreamConfig {
endpoint: DefaultJetstreamEndpoints::USEastTwo.into(),
wanted_collections: vec![],
wanted_dids: vec![],
compression: JetstreamCompression::None,
cursor: None, max_retries: 10,
max_delay_ms: 30_000,
base_delay_ms: 1_000,
reset_retries_min_ms: 30_000,
};
self.run_inner(&mut bot, default_config).await
}
async fn run_inner<T: AmselBot + CursorStore>(
&self,
bot: &mut T,
mut config: JetstreamConfig,
) -> Result<()> {
config.cursor = bot.cursor_load().map(|time_us| {
chrono::DateTime::from_timestamp_micros(time_us.try_into().unwrap()).unwrap()
});
let jetstream = JetstreamConnector::new(config).map_err(anyhow::Error::from)?;
let receiver = jetstream.connect().await.map_err(anyhow::Error::from)?;
while let Ok(event) = receiver.recv_async().await {
match event {
jetstream_oxide::events::JetstreamEvent::Commit(commit_event) => match commit_event
{
CommitEvent::Create { info, commit } => {
match commit.record {
atrium_api::record::KnownRecord::AppBskyActorProfile(_) => {}
atrium_api::record::KnownRecord::AppBskyFeedGenerator(feed) => {
let feed = BskyFeed::new(
feed.data,
BskyRecordRef {
cid: commit.cid,
uri: format!(
"at://{}/app.bsky.feed.generator/{}",
info.did.as_str(),
commit.info.rkey
),
},
self.agent.clone(),
);
bot.on_feed_sync(&feed);
let _future = bot.on_feed(feed).await;
}
atrium_api::record::KnownRecord::AppBskyFeedLike(_) => {}
atrium_api::record::KnownRecord::AppBskyFeedPost(post) => {
let post = BskyPost::new(
post.data,
BskyRecordRef {
cid: commit.cid,
uri: format!(
"at://{}/app.bsky.feed.post/{}",
info.did.as_str(),
commit.info.rkey
),
},
self.agent.clone(),
);
bot.on_post_sync(&post);
let _future = bot.on_post(post).await;
}
atrium_api::record::KnownRecord::AppBskyFeedPostgate(_) => {}
atrium_api::record::KnownRecord::AppBskyFeedRepost(_) => {}
atrium_api::record::KnownRecord::AppBskyFeedThreadgate(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphBlock(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphFollow(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphList(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphListblock(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphListitem(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphStarterpack(_) => {}
atrium_api::record::KnownRecord::AppBskyGraphVerification(_) => {}
atrium_api::record::KnownRecord::AppBskyLabelerService(_) => {}
atrium_api::record::KnownRecord::ChatBskyActorDeclaration(_) => {}
atrium_api::record::KnownRecord::ComAtprotoLexiconSchema(_) => {}
}
bot.cursor_store(info.time_us);
}
CommitEvent::Update { info: _, commit: _ } => {}
CommitEvent::Delete { info: _, commit: _ } => {}
},
jetstream_oxide::events::JetstreamEvent::Identity(_identity_event) => {}
jetstream_oxide::events::JetstreamEvent::Account(_account_event) => {}
}
}
Ok(())
}
}