hello ufos

.github/workflows/checks.yml

···
       - name: get nightly toolchain for jetstream fmt
         run: rustup toolchain install nightly --allow-downgrade -c rustfmt
       - name: fmt
-        run: cargo fmt --package links --package constellation -- --check
+        run: cargo fmt --package links --package constellation --package ufos -- --check
       - name: fmt jetstream (nightly)
         run: cargo +nightly fmt --package jetstream -- --check
       - name: clippy

Cargo.lock

···
 
 [[package]]
 name = "anyhow"
-version = "1.0.95"
+version = "1.0.97"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34ac096ce696dc2fcabef30516bb13c0a68a11d30131d3df6f04711467681b04"
+checksum = "dcfed56ad506cb2c684a14971b8861fdc3baaaae314b9e5f9bb532cbe3ba7a4f"
 
 [[package]]
 name = "arrayvec"
···
 
 [[package]]
 name = "clap"
-version = "4.5.28"
+version = "4.5.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e77c3243bd94243c03672cb5154667347c457ca271254724f9f393aee1c05ff"
+checksum = "027bb0d98429ae334a8698531da7077bdf906419543a35a55c2cb1b66437d767"
 dependencies = [
  "clap_builder",
  "clap_derive",
···
 
 [[package]]
 name = "clap_builder"
-version = "4.5.27"
+version = "4.5.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b26884eb4b57140e4d2d93652abfa49498b938b3c9179f9fc487b0acc3edad7"
+checksum = "5589e0cba072e0f3d23791efac0fd8627b49c829c196a492e88168e6a669d863"
 dependencies = [
  "anstream",
  "anstyle",
···
 
 [[package]]
 name = "serde_json"
-version = "1.0.139"
+version = "1.0.140"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44f86c3acccc9c65b153fe1b85a3be07fe5515274ec9f0653b4a0875731c72a6"
+checksum = "20068b6e96dc6c9bd23e01df8827e6c7e1f2fddd43c21810382803c136b99373"
 dependencies = [
  "itoa",
  "memchr",
···
 version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
+
+[[package]]
+name = "ufos"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "clap",
+ "jetstream",
+ "serde_json",
+ "tokio",
+]
 
 [[package]]
 name = "unicase"

Cargo.toml

···
     "links",
     "constellation",
     "jetstream",
+    "ufos",
 ]

Makefile

···
 	cargo test
 
 fmt:
-	cargo fmt --package links --package constellation
+	cargo fmt --package links --package constellation --package ufos
 	cargo +nightly fmt --package jetstream
 
 clippy:

jetstream/src/lib.rs

···
     USWestTwo,
 }
 
+impl DefaultJetstreamEndpoints {
+    /// Helper to reference official jetstream instances by shortcut
+    ///
+    /// This function will pass through a jetstream endpoint URL unless it matches a shortcut,
+    /// in which case it will be rewritten to the corresponding bluesky-operated jetstream endpoint
+    /// URL.
+    ///
+    /// The shortcuts available are
+    ///   - 'us-east-1'
+    ///   - 'us-east-2'
+    ///   - 'us-west-1'
+    ///   - 'us-west-2'
+    pub fn endpoint_or_shortcut(s: &str) -> String {
+        match s {
+            "us-east-1" => DefaultJetstreamEndpoints::USEastOne.into(),
+            "us-east-2" => DefaultJetstreamEndpoints::USEastTwo.into(),
+            "us-west-1" => DefaultJetstreamEndpoints::USWestOne.into(),
+            "us-west-2" => DefaultJetstreamEndpoints::USWestTwo.into(),
+            custom => custom.into(),
+        }
+    }
+}
+
 impl From<DefaultJetstreamEndpoints> for String {
     fn from(endpoint: DefaultJetstreamEndpoints) -> Self {
         match endpoint {

ufos/Cargo.toml

···
+[package]
+name = "ufos"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = "1.0.97"
+clap = { version = "4.5.31", features = ["derive"] }
+jetstream = { path = "../jetstream" }
+serde_json = "1.0.140"
+tokio = { version = "1.43.0", features = ["full", "sync", "time"] }

ufos/src/main.rs

···
+use clap::Parser;
+use std::path::PathBuf;
+use std::time::{Duration, Instant};
+
+use jetstream::{
+    events::{commit::CommitEvent, JetstreamEvent::Commit},
+    DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,
+};
+
+/// Aggregate links in the at-mosphere
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct Args {
+    /// Jetstream server to connect to (exclusive with --fixture). Provide either a wss:// URL, or a shorhand value:
+    /// 'us-east-1', 'us-east-2', 'us-west-1', or 'us-west-2'
+    #[arg(long)]
+    jetstream: String,
+    /// Location to store persist data to disk
+    #[arg(long)]
+    data: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Args::parse();
+
+    let config: JetstreamConfig<serde_json::Value> = JetstreamConfig {
+        endpoint: DefaultJetstreamEndpoints::endpoint_or_shortcut(&args.jetstream),
+        compression: JetstreamCompression::Zstd,
+        ..Default::default()
+    };
+
+    let jetstream: JetstreamConnector<serde_json::Value> = JetstreamConnector::new(config)?;
+    let receiver = jetstream.connect().await?;
+
+    println!("Jetstream ready");
+
+    let print_throttle = Duration::from_millis(400);
+    let mut last = Instant::now();
+    while let Ok(event) = receiver.recv_async().await {
+        if let Commit(CommitEvent::Create { commit, .. }) = event {
+            let now = Instant::now();
+            let since = now - last;
+            if since >= print_throttle {
+                let overshoot = since - print_throttle; // adjust to keep the rate on average
+                last = now - overshoot;
+                println!(
+                    "{}: {}",
+                    &*commit.info.collection,
+                    serde_json::to_string(&commit.record)?
+                );
+            }
+        }
+    }
+
+    Ok(())
+}