extract some event handling

src/main.rs

···
 use url::Url;
 use jetstream::{
     JetstreamCompression, JetstreamConfig, JetstreamConnector,
-    events::{CommitOp, Cursor, EventKind},
+    events::{CommitOp, Cursor, EventKind, JetstreamEvent},
     exports::Nsid,
 };
 use jacquard::{
···
     Ok(())
 }
 
+fn event_to_create_label<T: for <'a> Deserialize<'a>>(event: JetstreamEvent) -> Result<(T, Cursor)> {
+    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
+    let t = serde_json::from_str(raw.get())?;
+    Ok((t, event.cursor))
+}
+
 /// com.bad-example.identity.resolveMiniDoc bit we care about
 #[derive(Deserialize)]
 struct MiniDocResponse {
···
     println!("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())?],
···
         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?;
-
-    let req_client = reqwest::Client::builder()
-        .user_agent("hacktober_bot")
-        .timeout(Duration::from_secs(9))
-        .build()?;
 
     println!("receiving jetstream messages...");
     loop {
···
             eprintln!("consumer: could not receive event, bailing");
             break;
         };
-        if event.kind != EventKind::Commit {
-            continue;
-        }
-        let Some(ref commit) = event.commit else {
-            eprintln!("consumer: commit event missing commit data, ignoring");
+
+        let Ok((CreateLabelRecord { add, subject }, cursor)) = event_to_create_label(event) else {
             continue;
         };
-        if commit.operation != CommitOp::Create {
-            continue;
-        }
-        let Some(ref record) = commit.record else {
-            eprintln!("consumer: commit update/delete missing record, ignoring");
-            continue;
-        };
-        let CreateLabelRecord { add, subject } = match serde_json::from_str(record.get()) {
-            Ok(v) => v,
-            Err(e) => {
-                eprintln!("consumer: record failed to parse, ignoring: {e}");
-                continue;
-            }
-        };
+
         let mut added_good_first_issue = false;
         for added in add {
             if added.key == "at://did:plc:wshs7t2adsemcrrd4snkeqli/sh.tangled.label.definition/good-first-issue" {
-                println!("found a good first issue label!! {:?}", event.cursor);
+                println!("found a good first issue label!! {:?}", cursor);
                 added_good_first_issue = true;
                 break; // inner
             }
···
             continue;
         }
 
-        let IssueRecord { title, repo } = match get_record(&req_client, &subject).await {
+        let IssueRecord { title, repo } = match get_record(&slingshot_client, &subject).await {
             Ok(m) => m,
             Err(e) => {
                 eprintln!("failed to get issue record: {e} for {subject}");
···
             continue;
         };
 
-        let RepoRecord { name: repo_name } = match get_record(&req_client, &repo).await {
+        let RepoRecord { name: repo_name } = match get_record(&slingshot_client, &repo).await {
             Ok(m) => m,
             Err(e) => {
                 eprintln!("failed to get repo record: {e} for {subject}");
···
 
         let nice_tangled_repo_id = match repo_uri.authority() {
             AtIdentifier::Handle(h) => format!("@{h}"),
-            AtIdentifier::Did(did) => match get_handle(&req_client, did.as_str()).await {
+            AtIdentifier::Did(did) => match get_handle(&slingshot_client, did.as_str()).await {
                 Err(e) => {
                     eprintln!("failed to get mini doc from repo identifier: {e} for {subject}");
                     continue;