Something went wrong. Try again.
announcing good-first-issue tags added on @tangled.sh (not affiliated with tangled!)
Something went wrong. Try again.
13 kB · 411 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412use clap::Parser;use jacquard::{ api::{ app_bsky::feed::post::Post, app_bsky::richtext::facet::{ByteSlice, Facet}, com_atproto::repo::create_record::CreateRecord, com_atproto::server::create_session::CreateSession, }, client::{BasicClient, Session}, types::{ collection::Collection, datetime::Datetime, ident::AtIdentifier, language::Language, string::AtUri, value::Data, },};use jetstream::{ JetstreamCompression, JetstreamConfig, JetstreamConnector, events::{CommitOp, Cursor, EventKind, JetstreamEvent}, exports::Nsid,};use url::Url;
use serde::Deserialize;use std::time::Duration;
type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
#[derive(Debug, Parser)]#[command(version, about, long_about = None)]struct Args { /// pds of the bot user #[arg(short, long, env = "BOT_PDS")] pds: Url, /// handle or did of the bot user #[arg(short, long, env = "BOT_HANDLE")] identifier: String, /// app password for the bot user #[arg(short, long, env = "BOT_APP_PASSWORD")] app_password: String, /// lightweight firehose #[arg(short, long, env = "BOT_JETSTREAM_URL")] #[clap(default_value = "wss://jetstream1.us-east.fire.hose.cam/subscribe")] jetstream_url: Url, /// optional: we can pick up from a past jetstream cursor /// /// the default is to just live-tail /// /// warning: setting this can lead to rapid bot posting #[arg(long)] jetstream_cursor: Option<u64>, /// don't actually post #[arg(long, action)] dry_run: bool, /// send a checkin to this url every 5 mins #[arg(long)] healthcheck_ping: Option<Url>,}
struct IssueDetails { repo_full_name: String, repo_url: String, title: String, issues_url: String,}
/// com.bad-example.identity.resolveMiniDoc bit we care about#[derive(Deserialize)]struct MiniDocResponse { handle: String,}
/// com.atproto.repo.getRecord wraps the record in a `value` key#[derive(Deserialize)]struct GetRecordResonse<T> { value: T,}
/// part of CreateLabelRecord: key is the label reference (ie for "good-first-issue")#[derive(Deserialize)]struct AddLabel { key: String,}
/// tangled's record for adding labels to an issue#[derive(Deserialize)]struct CreateLabelRecord { add: Vec<AddLabel>, subject: String,}
/// tangled issue record#[derive(Deserialize)]struct IssueRecord { title: String, repo: String,}
/// tangled repo record#[derive(Deserialize)]struct RepoRecord { name: String,}
/// get some atproto record content (from slingshot)async fn get_record<T: for<'a> Deserialize<'a>>( client: &reqwest::Client, at_uri: &str,) -> Result<T> { let mut url: Url = "https://slingshot.microcosm.blue".parse()?; url.set_path("/xrpc/com.bad-example.repo.getUriRecord"); url.query_pairs_mut().append_pair("at_uri", at_uri); let GetRecordResonse { value } = client .get(url) .send() .await? .error_for_status()? .json() .await?; Ok(value)}
/// try to resolve a bidirectionally verified handle from an identifier (via slingshot)async fn get_handle(client: &reqwest::Client, identifier: &str) -> Result<Option<String>> { let mut url: Url = "https://slingshot.microcosm.blue".parse()?; url.set_path("/xrpc/com.bad-example.identity.resolveMiniDoc"); url.query_pairs_mut().append_pair("identifier", identifier); let MiniDocResponse { handle } = client .get(url) .send() .await? .error_for_status()? .json() .await?; if handle == "handle.invalid" { Ok(None) } else { Ok(Some(handle)) }}
fn event_to_create_label<T: for<'a> Deserialize<'a>>(event: JetstreamEvent) -> Result<T> { if event.kind != EventKind::Commit { return Err("not a commit".into()); } let commit = event.commit.ok_or("commit event missing commit data")?; if commit.operation != CommitOp::Create { return Err("not a create event".into()); }
let raw = commit.record.ok_or("commit missing record")?;
// todo: delete post if label is removed // delete sample: at://did:plc:hdhoaan3xa3jiuq4fg4mefid/sh.tangled.label.op/3m2jvx4c6wf22 // tldr: has a "delete" array just like "add" on the same op collection Ok(serde_json::from_str(raw.get())?)}
async fn extract_issue_info( client: &reqwest::Client, adds: Vec<AddLabel>, subject: String,) -> Result<IssueDetails> { let mut added_good_first_issue = false; for added in adds { if added.key == "at://did:plc:wshs7t2adsemcrrd4snkeqli/sh.tangled.label.definition/good-first-issue" { log::info!("found a good first issue label!!"); added_good_first_issue = true; break; // inner } log::debug!("found a label but it wasn't good-first-issue, ignoring..."); } if !added_good_first_issue { return Err("good-first-issue label not found in added labels".into()); }
let IssueRecord { title, repo } = match get_record(client, &subject).await { Ok(m) => m, Err(e) => return Err(format!("failed to get issue record: {e} for {subject}").into()), };
let Ok(repo_uri) = AtUri::new(&repo) else { return Err("failed to parse repo to aturi for {subject}".into()); };
let RepoRecord { name: repo_name } = match get_record(client, &repo).await { Ok(m) => m, Err(e) => return Err(format!("failed to get repo record: {e} for {subject}").into()), };
let nice_tangled_repo_id = match repo_uri.authority() { AtIdentifier::Handle(h) => format!("@{h}"), AtIdentifier::Did(did) => match get_handle(client, did.as_str()).await { Err(e) => { return Err(format!( "failed to get mini doc from repo identifier: {e} for {subject}" ) .into()); } Ok(None) => did.to_string(), Ok(Some(h)) => format!("@{h}"), }, };
let repo_full_name = format!("{nice_tangled_repo_id}/{repo_name}"); let repo_url = format!("https://tangled.org/{nice_tangled_repo_id}/{repo_name}");
let issues_url = format!("https://tangled.org/{nice_tangled_repo_id}/{repo_name}/issues");
Ok(IssueDetails { repo_full_name, repo_url, title, issues_url, })}
async fn post( client: &BasicClient, identifier: &AtIdentifier<'_>, IssueDetails { repo_full_name, repo_url, title, issues_url, }: &IssueDetails,) -> Result<()> { let message = format!( r#"New from {repo_full_name}:
> {title}"# );
let pre_len = 9;
let repo_feature = serde_json::json!({ "$type": "app.bsky.richtext.facet#link", "uri": repo_url, }); let repo_facet = Facet { features: vec![Data::from_json(&repo_feature)?], index: ByteSlice { byte_start: pre_len, byte_end: pre_len + repo_full_name.len() as i64, extra_data: Default::default(), }, extra_data: Default::default(), };
let title_starts_at = pre_len + (repo_full_name.len() + 5) as i64;
let repo_issues_feature = serde_json::json!({ "$type": "app.bsky.richtext.facet#link", "uri": issues_url, }); let issues_facet = Facet { features: vec![Data::from_json(&repo_issues_feature)?], index: ByteSlice { byte_start: title_starts_at, byte_end: title_starts_at + title.len() as i64, extra_data: Default::default(), }, extra_data: Default::default(), };
// Make a post let post = Post { created_at: Datetime::now(), langs: Some(vec![Language::new("en")?]), text: message.into(), facets: Some(vec![repo_facet, issues_facet]), embed: Default::default(), entities: Default::default(), labels: Default::default(), reply: Default::default(), tags: Default::default(), extra_data: Default::default(), };
let json = serde_json::to_value(post)?; let data = Data::from_json(&json)?;
log::info!("\nposting..."); client .send( CreateRecord::new() .repo(identifier.clone()) .collection(Post::nsid()) .record(data) .build(), ) .await? .into_output()?;
Ok(())}
async fn hc_ping(url: Url, client: reqwest::Client) { let mut interval = tokio::time::interval(Duration::from_secs(5 * 60)); loop { interval.tick().await; log::trace!("sending healthcheck ping..."); if let Err(e) = client .get(url.clone()) .send() .await .and_then(reqwest::Response::error_for_status) { log::warn!("error sending healthcheck ping: {e}"); } }}
#[tokio::main]async fn main() -> Result<()> { env_logger::init(); let args = Args::parse();
// Create HTTP client and session let client = BasicClient::new(args.pds); let bot_id = AtIdentifier::new(&args.identifier)?; let create_session = CreateSession::new() .identifier(bot_id.to_string()) .password(&args.app_password) .build(); let session = Session::from(client.send(create_session.clone()).await?.into_output()?); log::debug!("logged in as {} ({})", session.handle, session.did); client.set_session(session).await?;
let slingshot_client = reqwest::Client::builder() .user_agent("hacktober_bot") .timeout(Duration::from_secs(9)) .build()?;
let jetstream_config: JetstreamConfig = JetstreamConfig { endpoint: args.jetstream_url.to_string(), wanted_collections: vec![Nsid::new("sh.tangled.label.op".to_string())?], user_agent: Some("hacktober_bot".to_string()), compression: JetstreamCompression::Zstd, replay_on_reconnect: true, channel_size: 1024, // buffer up to ~1s of jetstream events ..Default::default() }; let mut receiver = JetstreamConnector::new(jetstream_config)? .connect_cursor(args.jetstream_cursor.map(Cursor::from_raw_u64)) .await?;
if let Some(hc) = args.healthcheck_ping { log::info!("starting healthcheck ping task..."); tokio::task::spawn(hc_ping(hc.clone(), slingshot_client.clone())); }
log::info!("receiving jetstream messages..."); loop { let Some(event) = receiver.recv().await else { log::error!("consumer: could not receive event, bailing"); break; }; let cursor = event.cursor;
let CreateLabelRecord { add: adds, subject } = match event_to_create_label(event) { Ok(clr) => clr, Err(e) => { log::debug!("ignoring unparseable event (at {cursor:?}): {e}"); continue; } };
let issue_details = match extract_issue_info(&slingshot_client, adds, subject.clone()).await { Ok(deets) => deets, Err(e) => { log::warn!("failed to extract issue details (at {cursor:?}): {e}"); continue; } };
if args.dry_run { let IssueDetails { repo_full_name, repo_url, title, issues_url, } = issue_details; log::info!( r#"--dry-run, but would have posted:
good-first-issue label added for {repo_full_name} ({repo_url}):
> {title} ({issues_url})"# ); continue; }
if let Err(e) = post(&client, &bot_id, &issue_details).await { log::warn!("failed to post for {subject}: {e}, refreshing session for one retry..."); let session = Session::from(client.send(create_session.clone()).await?.into_output()?); log::debug!("logged in as {} ({})", session.handle, session.did); client.set_session(session).await?;
if let Err(e) = post(&client, &bot_id, &issue_details).await { log::error!( "failed to post after a session refresh: {e:?}, something is wrong. bye." ); break; } }; }
Ok(())}