remove rocketman (switch to crate)

.github/workflows/security.yml

···
-# yaml-language-server: $schema=https://json.schemastore.org/github-workflow.json
-
-name: Security
-
-on:
-  push:
-    branches: [main, develop]
-  pull_request:
-    branches: [main, develop]
-  schedule:
-    # Run security checks daily at 2 AM sunday
-    - cron: "0 2 * * 0"
-
-jobs:
-  codeql-analysis:
-    name: CodeQL Analysis
-    runs-on: ubuntu-latest
-    permissions:
-      actions: read
-      contents: read
-      security-events: write
-
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-
-      - name: Initialize CodeQL
-        uses: github/codeql-action/init@v3
-        with:
-          languages: "javascript,typescript,rust"
-          queries: security-extended,security-and-quality
-
-      - name: Setup environment for all languages
-        uses: ./.github/actions/setup
-        with:
-          setup-node: "true"
-          setup-rust: "true"
-
-      - name: Perform a full build for CodeQL
-        run: |
-          echo "Building Node.js projects..."
-          pnpm build
-          echo "Building Rust projects..."
-          (cd services && cargo build --all-features)
-          (cd apps/aqua && cargo build --all-features)
-
-      - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@v3
-
-  docker-security-scan:
-    name: Docker Security Scan
-    runs-on: ubuntu-latest
-    if: github.event_name == 'push' || github.event_name == 'schedule'
-    strategy:
-      matrix:
-        service: [aqua, cadet]
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-
-      - name: Setup environment
-        uses: ./.github/actions/setup
-        with:
-          setup-node: "true"
-          lexicons-only-rust: "true"
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
-
-      - name: Build Docker image
-        uses: docker/build-push-action@v5
-        with:
-          context: .
-          file: ${{ matrix.service == 'aqua' && './apps/aqua/Dockerfile' || './services/cadet/Dockerfile' }}
-          load: true
-          tags: ${{ matrix.service }}:latest
-          cache-from: type=gha,scope=${{ matrix.service }}
-          cache-to: type=gha,mode=max,scope=${{ matrix.service }}
-
-      - name: Run Trivy vulnerability scanner
-        uses: aquasecurity/trivy-action@master
-        with:
-          image-ref: "${{ matrix.service }}:latest"
-          format: "sarif"
-          output: "trivy-results-${{ matrix.service }}.sarif"
-          severity: "CRITICAL,HIGH"
-          exit-code: "1"
-
-      - name: Upload Trivy scan results to GitHub Security tab
-        uses: github/codeql-action/upload-sarif@v3
-        if: always()
-        with:
-          sarif_file: "trivy-results-${{ matrix.service }}.sarif"
-
-  secrets-scan:
-    name: Secrets Scan
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-
-      - name: Run TruffleHog OSS
-        uses: trufflesecurity/trufflehog@main
-        with:
-          path: ./
-          base: main
-          head: HEAD
-          extra_args: --debug --only-verified

Cargo.lock

···
  "atmst",
  "atrium-api",
  "axum",
- "base64",
+ "base64 0.22.1",
  "chrono",
  "clap",
  "dotenvy",
···
  "atrium-common",
  "atrium-xrpc",
  "chrono",
- "http",
+ "http 1.3.1",
  "ipld-core",
  "langtag",
  "regex",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0216ad50ce34e9ff982e171c3659e65dedaa2ed5ac2994524debdc9a9647ffa8"
 dependencies = [
- "http",
+ "http 1.3.1",
  "serde",
  "serde_html_form",
  "serde_json",
···
  "bytes",
  "form_urlencoded",
  "futures-util",
- "http",
+ "http 1.3.1",
  "http-body",
  "http-body-util",
  "hyper",
···
 dependencies = [
  "bytes",
  "futures-core",
- "http",
+ "http 1.3.1",
  "http-body",
  "http-body-util",
  "mime",
···
 
 [[package]]
 name = "base64"
+version = "0.21.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567"
+
+[[package]]
+name = "base64"
 version = "0.22.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
···
  "async-trait",
  "atmst",
  "atrium-api",
- "base64",
+ "base64 0.22.1",
  "chrono",
  "cid 0.11.1",
  "dotenvy",
···
  "sqlx",
  "time",
  "tokio",
- "tokio-tungstenite",
+ "tokio-tungstenite 0.24.0",
  "tracing",
  "tracing-subscriber",
  "types",
···
 
 [[package]]
 name = "core-foundation"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
+name = "core-foundation"
 version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6"
···
  "fnv",
  "futures-core",
  "futures-sink",
- "http",
+ "http 1.3.1",
  "indexmap",
  "slab",
  "tokio",
···
 
 [[package]]
 name = "http"
+version = "0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1"
+dependencies = [
+ "bytes",
+ "fnv",
+ "itoa",
+]
+
+[[package]]
+name = "http"
 version = "1.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f4a85d31aea989eead29a3aaf9e1115a180df8282431156e533de47660892565"
···
 checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
 dependencies = [
  "bytes",
- "http",
+ "http 1.3.1",
 ]
 
 [[package]]
···
 dependencies = [
  "bytes",
  "futures-core",
- "http",
+ "http 1.3.1",
  "http-body",
  "pin-project-lite",
 ]
···
  "futures-channel",
  "futures-util",
  "h2",
- "http",
+ "http 1.3.1",
  "http-body",
  "httparse",
  "httpdate",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58"
 dependencies = [
- "http",
+ "http 1.3.1",
  "hyper",
  "hyper-util",
- "rustls",
- "rustls-native-certs",
+ "rustls 0.23.31",
+ "rustls-native-certs 0.8.1",
  "rustls-pki-types",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.26.2",
  "tower-service",
  "webpki-roots 1.0.2",
 ]
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8d9b05277c7e8da2c93a568989bb6207bef0112e8d17df7a6eda4a3cf143bc5e"
 dependencies = [
- "base64",
+ "base64 0.22.1",
  "bytes",
  "futures-channel",
  "futures-core",
  "futures-util",
- "http",
+ "http 1.3.1",
  "http-body",
  "hyper",
  "ipnet",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dd7399781913e5393588a8d8c6a2867bf85fb38eaf2502fdce465aad2dc6f034"
 dependencies = [
- "base64",
+ "base64 0.22.1",
  "http-body-util",
  "hyper",
  "hyper-rustls",
···
  "bytes",
  "encoding_rs",
  "futures-util",
- "http",
+ "http 1.3.1",
  "httparse",
  "memchr",
  "mime",
···
  "quinn-proto",
  "quinn-udp",
  "rustc-hash 2.1.1",
- "rustls",
+ "rustls 0.23.31",
  "socket2 0.5.10",
  "thiserror 2.0.12",
  "tokio",
···
  "rand 0.9.2",
  "ring",
  "rustc-hash 2.1.1",
- "rustls",
+ "rustls 0.23.31",
  "rustls-pki-types",
  "slab",
  "thiserror 2.0.12",
···
 checksum = "cbc931937e6ca3a06e3b6c0aa7841849b160a90351d6ab467a8b9b9959767531"
 dependencies = [
  "async-compression",
- "base64",
+ "base64 0.22.1",
  "bytes",
  "futures-core",
  "futures-util",
- "http",
+ "http 1.3.1",
  "http-body",
  "http-body-util",
  "hyper",
···
  "percent-encoding",
  "pin-project-lite",
  "quinn",
- "rustls",
+ "rustls 0.23.31",
  "rustls-pki-types",
  "serde",
  "serde_json",
  "serde_urlencoded",
  "sync_wrapper",
  "tokio",
- "tokio-rustls",
+ "tokio-rustls 0.26.2",
  "tokio-util",
  "tower",
  "tower-http",
···
 [[package]]
 name = "rocketman"
 version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9928fe43979c19ff1f46f7920c30b76dfcead7a4d571c9836c4d02da8587f844"
 dependencies = [
  "anyhow",
  "async-trait",
···
  "derive_builder",
  "flume",
  "futures-util",
- "metrics 0.23.1",
+ "metrics 0.24.2",
  "rand 0.8.5",
  "serde",
  "serde_json",
  "tokio",
- "tokio-tungstenite",
+ "tokio-tungstenite 0.20.1",
  "tracing",
  "tracing-subscriber",
  "url",
···
 
 [[package]]
 name = "rustls"
+version = "0.21.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e"
+dependencies = [
+ "log",
+ "ring",
+ "rustls-webpki 0.101.7",
+ "sct",
+]
+
+[[package]]
+name = "rustls"
 version = "0.23.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c0ebcbd2f03de0fc1122ad9bb24b127a5a6cd51d72604a3f3c50ac459762b6cc"
···
  "once_cell",
  "ring",
  "rustls-pki-types",
- "rustls-webpki",
+ "rustls-webpki 0.103.4",
  "subtle",
  "zeroize",
 ]
 
 [[package]]
 name = "rustls-native-certs"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00"
+dependencies = [
+ "openssl-probe",
+ "rustls-pemfile",
+ "schannel",
+ "security-framework 2.11.1",
+]
+
+[[package]]
+name = "rustls-native-certs"
 version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7fcff2dd52b58a8d98a70243663a0d234c4e2b79235637849d15913394a247d3"
···
  "openssl-probe",
  "rustls-pki-types",
  "schannel",
- "security-framework",
+ "security-framework 3.2.0",
+]
+
+[[package]]
+name = "rustls-pemfile"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1c74cae0a4cf6ccbbf5f359f08efdf8ee7e1dc532573bf0db71968cb56b1448c"
+dependencies = [
+ "base64 0.21.7",
 ]
 
 [[package]]
···
 dependencies = [
  "web-time",
  "zeroize",
+]
+
+[[package]]
+name = "rustls-webpki"
+version = "0.101.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b6275d1ee7a1cd780b64aca7726599a1dbc893b1e64144529e55c3c2f745765"
+dependencies = [
+ "ring",
+ "untrusted",
 ]
 
 [[package]]
···
 checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
 [[package]]
+name = "sct"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da046153aa2352493d6cb7da4b6e5c0c057d8a1d0a9aa8560baffdd945acd414"
+dependencies = [
+ "ring",
+ "untrusted",
+]
+
+[[package]]
 name = "sec1"
 version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "security-framework"
+version = "2.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02"
+dependencies = [
+ "bitflags 2.9.1",
+ "core-foundation 0.9.4",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
+[[package]]
+name = "security-framework"
 version = "3.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "271720403f46ca04f7ba6f55d438f8bd878d6b8ca0a1046e8228c4145bcbb316"
 dependencies = [
  "bitflags 2.9.1",
- "core-foundation",
+ "core-foundation 0.10.1",
  "core-foundation-sys",
  "libc",
  "security-framework-sys",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6"
 dependencies = [
- "base64",
+ "base64 0.22.1",
  "bytes",
  "crc",
  "crossbeam-queue",
···
  "memchr",
  "once_cell",
  "percent-encoding",
- "rustls",
+ "rustls 0.23.31",
  "serde",
  "serde_json",
  "sha2",
···
 checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526"
 dependencies = [
  "atoi",
- "base64",
+ "base64 0.22.1",
  "bitflags 2.9.1",
  "byteorder",
  "bytes",
···
 checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46"
 dependencies = [
  "atoi",
- "base64",
+ "base64 0.22.1",
  "bitflags 2.9.1",
  "byteorder",
  "crc",
···
 
 [[package]]
 name = "tokio-rustls"
+version = "0.24.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081"
+dependencies = [
+ "rustls 0.21.12",
+ "tokio",
+]
+
+[[package]]
+name = "tokio-rustls"
 version = "0.26.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b"
 dependencies = [
- "rustls",
+ "rustls 0.23.31",
  "tokio",
 ]
 
···
 
 [[package]]
 name = "tokio-tungstenite"
+version = "0.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c"
+dependencies = [
+ "futures-util",
+ "log",
+ "rustls 0.21.12",
+ "rustls-native-certs 0.6.3",
+ "tokio",
+ "tokio-rustls 0.24.1",
+ "tungstenite 0.20.1",
+ "webpki-roots 0.25.4",
+]
+
+[[package]]
+name = "tokio-tungstenite"
 version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9"
 dependencies = [
  "futures-util",
  "log",
- "rustls",
+ "rustls 0.23.31",
  "rustls-pki-types",
  "tokio",
- "tokio-rustls",
- "tungstenite",
+ "tokio-rustls 0.26.2",
+ "tungstenite 0.24.0",
  "webpki-roots 0.26.11",
 ]
 
···
  "bitflags 2.9.1",
  "bytes",
  "futures-util",
- "http",
+ "http 1.3.1",
  "http-body",
  "iri-string",
  "pin-project-lite",
···
 
 [[package]]
 name = "tungstenite"
+version = "0.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e3dac10fd62eaf6617d3a904ae222845979aec67c615d1c842b4002c7666fb9"
+dependencies = [
+ "byteorder",
+ "bytes",
+ "data-encoding",
+ "http 0.2.12",
+ "httparse",
+ "log",
+ "rand 0.8.5",
+ "rustls 0.21.12",
+ "sha1",
+ "thiserror 1.0.69",
+ "url",
+ "utf-8",
+]
+
+[[package]]
+name = "tungstenite"
 version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a"
···
  "byteorder",
  "bytes",
  "data-encoding",
- "http",
+ "http 1.3.1",
  "httparse",
  "log",
  "rand 0.8.5",
- "rustls",
+ "rustls 0.23.31",
  "rustls-pki-types",
  "sha1",
  "thiserror 1.0.69",
···
  "atrium-api",
  "atrium-xrpc",
  "chrono",
- "http",
+ "http 1.3.1",
  "ipld-core",
  "langtag",
  "regex",
···
  "js-sys",
  "wasm-bindgen",
 ]
+
+[[package]]
+name = "webpki-roots"
+version = "0.25.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1"
 
 [[package]]
 name = "webpki-roots"

Cargo.toml

···
 [workspace]
-members = [
-    "apps/aqua",
-    "services/cadet",
-    "services/rocketman",
-    "tools/teal-cli",
-]
+members = ["apps/aqua", "services/cadet", "tools/teal-cli"]
 resolver = "2"
 
 [workspace.dependencies]
···
 chrono = "0.4"
 uuid = { version = "1.0", features = ["v4", "serde"] }
 types = { path = "services/types" }
-rocketman = { path = "services/rocketman" }
+rocketman = "0.2.3"
 
 # CAR and IPLD dependencies
 iroh-car = "0.5"

compose.db-test.yml

···
-version: "3.8"
-
-services:
-  postgres:
-    image: postgres:latest
-    container_name: postgres_test_db
-    environment:
-      POSTGRES_USER: postgres
-      POSTGRES_PASSWORD: testpass123
-      POSTGRES_DB: teal_test
-    ports:
-      - "5433:5432"
-    volumes:
-      - postgres_test_data:/var/lib/postgresql/data
-    networks:
-      - test_network
-    command: postgres -c log_statement=all -c log_destination=stderr
-
-networks:
-  test_network:
-    driver: bridge
-
-volumes:
-  postgres_test_data:

services/Cargo.toml

···
 [workspace]
-members = ["cadet", "rocketman", "satellite", "types"]
+members = ["cadet", "satellite", "types"]
 resolver = "2"
 
 [workspace.dependencies]
···
 chrono = { version = "0.4", features = ["serde"] }
 uuid = { version = "1.0", features = ["v4", "serde"] }
 types = { path = "types" }
-rocketman = { path = "rocketman" }
+rocketman = "0.2.5"
 
 # CAR and IPLD dependencies
 iroh-car = "0.4"

services/rocketman/Cargo.toml

···
-[package]
-name = "rocketman"
-version = "0.2.3"
-edition = "2021"
-
-license = "MIT"
-authors = ["Natalie B. <[e-mail]>"]
-repository = "https://github.com/espeon/cadet"
-
-readme = "readme.md"
-
-description = "A modular(ish) jetstream consumer."
-
-[dependencies]
-tokio.workspace = true
-tokio-tungstenite.workspace = true
-futures-util = "0.3"
-url.workspace = true
-rand.workspace = true
-tracing.workspace = true
-tracing-subscriber.workspace = true
-metrics.workspace = true
-derive_builder = "0.20.2"
-bon = "3.3.2"
-serde = { workspace = true, features = ["derive"] }
-serde_json.workspace = true
-flume.workspace = true
-anyhow.workspace = true
-async-trait.workspace = true
-zstd = { version = "0.13.3", optional = true }
-
-[features]
-default = ["zstd"]
-zstd = ["dep:zstd"]

services/rocketman/examples/spew-bsky-posts.rs

···
-use async_trait::async_trait;
-use rocketman::{
-    connection::JetstreamConnection,
-    handler,
-    ingestion::LexiconIngestor,
-    options::JetstreamOptions,
-    types::event::{Commit, Event},
-};
-use serde_json::Value;
-use std::{collections::HashMap, sync::Arc, sync::Mutex};
-
-#[tokio::main]
-async fn main() {
-    // init the builder
-    let opts = JetstreamOptions::builder()
-        // your EXACT nsids
-        .wanted_collections(vec!["app.bsky.feed.post".to_string()])
-        .build();
-    // create the jetstream connector
-    let jetstream = JetstreamConnection::new(opts);
-
-    // create your ingestors
-    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
-    ingestors.insert(
-        // your EXACT nsid
-        "app.bsky.feed.post".to_string(),
-        Box::new(MyCoolIngestor),
-    );
-
-    // tracks the last message we've processed
-    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
-
-    // get channels
-    let msg_rx = jetstream.get_msg_rx();
-    let reconnect_tx = jetstream.get_reconnect_tx();
-
-    // spawn a task to process messages from the queue.
-    // this is a simple implementation, you can use a more complex one based on needs.
-    let c_cursor = cursor.clone();
-    tokio::spawn(async move {
-        while let Ok(message) = msg_rx.recv_async().await {
-            if let Err(e) =
-                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
-                    .await
-            {
-                eprintln!("Error processing message: {}", e);
-            };
-        }
-    });
-
-    // connect to jetstream
-    // retries internally, but may fail if there is an extreme error.
-    if let Err(e) = jetstream.connect(cursor.clone()).await {
-        eprintln!("Failed to connect to Jetstream: {}", e);
-        std::process::exit(1);
-    }
-}
-
-pub struct MyCoolIngestor;
-
-/// A cool ingestor implementation. Will just print the message. Does not do verification.
-#[async_trait]
-impl LexiconIngestor for MyCoolIngestor {
-    async fn ingest(&self, message: Event<Value>) -> anyhow::Result<()> {
-        if let Some(Commit {
-            record: Some(record),
-            ..
-        }) = message.commit
-        {
-            if let Some(Value::String(text)) = record.get("text") {
-                println!("{text:?}");
-            }
-        }
-        Ok(())
-    }
-}

services/rocketman/package.json

···
-{
-  "name": "@repo/rocketman",
-  "private": true,
-  "scripts": {
-    "build": "cargo build --release",
-    "build:rust": "cargo build --release",
-    "dev": "cargo watch -x 'run'",
-    "test": "cargo test",
-    "test:rust": "cargo test"
-  }
-}

services/rocketman/readme.md

···
-## Rocketman
-
-A modular(ish) jetstream consumer. Backed by Tungstenite.
-
-
-### Installation
-```toml
-[dependencies]
-rocketman = "latest" # pyt the latest version here
-tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
-```
-### Usage
-```rs
-#[tokio::main]
-async fn main() {
-    // init the builder
-    let opts = JetstreamOptions::builder()
-        // your EXACT nsids
-        .wanted_collections(vec!["com.example.cool.nsid".to_string()])
-        .build();
-    // create the jetstream connector
-    let jetstream = JetstreamConnection::new(opts);
-
-    // create your ingestors
-    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
-    ingestors.insert(
-        // your EXACT nsid
-        "com.example.cool.nsid".to_string(),
-        Box::new(MyCoolIngestor),
-    );
-
-
-    // tracks the last message we've processed
-    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
-
-    // get channels
-    let msg_rx = jetstream.get_msg_rx();
-    let reconnect_tx = jetstream.get_reconnect_tx();
-
-    // spawn a task to process messages from the queue.
-    // this is a simple implementation, you can use a more complex one based on needs.
-    let c_cursor = cursor.clone();
-    tokio::spawn(async move {
-        while let Ok(message) = msg_rx.recv_async().await {
-            if let Err(e) =
-                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
-                    .await
-            {
-                error!("Error processing message: {}", e);
-            };
-        }
-    });
-
-    // connect to jetstream
-    // retries internally, but may fail if there is an extreme error.
-    if let Err(e) = jetstream.connect(cursor.clone()).await {
-        error!("Failed to connect to Jetstream: {}", e);
-        std::process::exit(1);
-    }
-}
-
-pub struct MyCoolIngestor;
-
-/// A cool ingestor implementation. Will just print the message. Does not do verification.
-impl LexiconIngestor for MyCoolIngestor {
-    async fn ingest(&self, message: Event<Value>) -> Result<()> {
-        info!("{:?}", message);
-        // Process message for default lexicon.
-        Ok(())
-    }
-}
-```
-### gratz
-Based heavily on [phil's jetstream consumer on atcosm constellation.](https://github.com/atcosm/links/blob/main/constellation/src/consumer/jetstream.rs)

services/rocketman/src/connection.rs

···
-use flume::{Receiver, Sender};
-use futures_util::StreamExt;
-use metrics::{counter, describe_counter, describe_histogram, histogram, Unit};
-use std::cmp::{max, min};
-use std::sync::{Arc, Mutex};
-use std::time::Instant;
-use tokio::time::{sleep, Duration};
-use tokio_tungstenite::{connect_async, tungstenite::Message};
-use tracing::{error, info};
-use url::Url;
-
-use crate::options::JetstreamOptions;
-use crate::time::system_time::SystemTimeProvider;
-use crate::time::TimeProvider;
-
-pub struct JetstreamConnection {
-    pub opts: JetstreamOptions,
-    reconnect_tx: flume::Sender<()>,
-    reconnect_rx: flume::Receiver<()>,
-    msg_tx: flume::Sender<Message>,
-    msg_rx: flume::Receiver<Message>,
-}
-
-impl JetstreamConnection {
-    pub fn new(opts: JetstreamOptions) -> Self {
-        let (reconnect_tx, reconnect_rx) = flume::bounded(opts.bound);
-        let (msg_tx, msg_rx) = flume::bounded(opts.bound);
-        Self {
-            opts,
-            reconnect_tx,
-            reconnect_rx,
-            msg_tx,
-            msg_rx,
-        }
-    }
-
-    pub fn get_reconnect_tx(&self) -> Sender<()> {
-        self.reconnect_tx.clone()
-    }
-
-    pub fn get_msg_rx(&self) -> Receiver<Message> {
-        self.msg_rx.clone()
-    }
-
-    fn build_ws_url(&self, cursor: Arc<Mutex<Option<u64>>>) -> String {
-        let mut url = Url::parse(&self.opts.ws_url.to_string()).unwrap();
-
-        // Append query params
-        if let Some(ref cols) = self.opts.wanted_collections {
-            for col in cols {
-                url.query_pairs_mut().append_pair("wantedCollections", col);
-            }
-        }
-        if let Some(ref dids) = self.opts.wanted_dids {
-            for did in dids {
-                url.query_pairs_mut().append_pair("wantedDids", did);
-            }
-        }
-        if let Some(cursor) = cursor.lock().unwrap().as_ref() {
-            url.query_pairs_mut()
-                .append_pair("cursor", &cursor.to_string());
-        }
-        #[cfg(feature = "zstd")]
-        if self.opts.compress {
-            url.query_pairs_mut().append_pair("compress", "true");
-        }
-
-        url.to_string()
-    }
-
-    pub async fn connect(
-        &self,
-        cursor: Arc<Mutex<Option<u64>>>,
-    ) -> Result<(), Box<dyn std::error::Error>> {
-        describe_counter!(
-            "jetstream.connection.attempt",
-            Unit::Count,
-            "attempts to connect to jetstream service"
-        );
-        describe_counter!(
-            "jetstream.connection.error",
-            Unit::Count,
-            "errors connecting to jetstream service"
-        );
-        describe_histogram!(
-            "jetstream.connection.duration",
-            Unit::Seconds,
-            "Time connected to jetstream service"
-        );
-        describe_counter!(
-            "jetstream.connection.reconnect",
-            Unit::Count,
-            "reconnects to jetstream service"
-        );
-        let mut retry_interval = 1;
-
-        let time_provider = SystemTimeProvider::new();
-
-        let mut start_time = time_provider.now();
-
-        loop {
-            counter!("jetstream.connection.attempt").increment(1);
-            info!("Connecting to {}", self.opts.ws_url);
-            let start = Instant::now();
-
-            let ws_url = self.build_ws_url(cursor.clone());
-
-            match connect_async(ws_url).await {
-                Ok((ws_stream, response)) => {
-                    let elapsed = start.elapsed();
-                    info!("Connected. HTTP status: {}", response.status());
-
-                    let (_, mut read) = ws_stream.split();
-
-                    loop {
-                        // Inner loop to handle messages, reconnect signals, and receive timeout
-                        let receive_timeout =
-                            sleep(Duration::from_secs(self.opts.timeout_time_sec as u64));
-                        tokio::pin!(receive_timeout);
-
-                        loop {
-                            tokio::select! {
-                                message_result = read.next() => {
-                                    match message_result {
-                                        Some(message) => {
-                                            // Reset timeout on message received
-                                            receive_timeout.as_mut().reset(tokio::time::Instant::now() + Duration::from_secs(self.opts.timeout_time_sec as u64));
-
-                                            histogram!("jetstream.connection.duration").record(elapsed.as_secs_f64());
-                                            match message {
-                                                Ok(message) => {
-                                                    if let Err(err) = self.msg_tx.send_async(message).await {
-                                                        counter!("jetstream.error").increment(1);
-                                                        error!("Failed to queue message: {}", err);
-                                                    }
-                                                }
-                                                Err(e) => {
-                                                    counter!("jetstream.error").increment(1);
-                                                    error!("Error: {}", e);
-                                                }
-                                            }
-                                        }
-                                        None => {
-                                            info!("Stream closed by server.");
-                                            counter!("jetstream.connection.reconnect").increment(1);
-                                            break; // Stream ended, break inner loop to reconnect
-                                        }
-                                    }
-                                }
-                                _ = self.reconnect_rx.recv_async() => {
-                                    info!("Reconnect signal received.");
-                                    counter!("jetstream.connection.reconnect").increment(1);
-                                    break;
-                                }
-                                _ = &mut receive_timeout => {
-                                    // last final poll, just in case
-                                    match read.next().await {
-                                        Some(Ok(message)) => {
-                                            if let Err(err) = self.msg_tx.send_async(message).await {
-                                                counter!("jetstream.error").increment(1);
-                                                error!("Failed to queue message: {}", err);
-                                            }
-                                            // Reset timeout to continue
-                                            receive_timeout.as_mut().reset(tokio::time::Instant::now() + Duration::from_secs(self.opts.timeout_time_sec as u64));
-                                        }
-                                        Some(Err(e)) => {
-                                            counter!("jetstream.error").increment(1);
-                                            error!("Error receiving message during final poll: {}", e);
-                                            counter!("jetstream.connection.reconnect").increment(1);
-                                            break;
-                                        }
-                                        None => {
-                                            info!("No commits received in {} seconds, reconnecting.", self.opts.timeout_time_sec);
-                                            counter!("jetstream.connection.reconnect").increment(1);
-                                            break;
-                                        }
-                                    }
-                                }
-                            }
-                        }
-                    }
-                }
-                Err(e) => {
-                    let elapsed_time = time_provider.elapsed(start_time);
-                    // reset if time connected > the time we set
-                    if elapsed_time.as_secs() > self.opts.max_retry_interval_seconds {
-                        retry_interval = 0;
-                        start_time = time_provider.now();
-                    }
-                    counter!("jetstream.connection.error").increment(1);
-                    error!("Connection error: {}", e);
-                }
-            }
-
-            let sleep_time = max(1, min(self.opts.max_retry_interval_seconds, retry_interval));
-            info!("Reconnecting in {} seconds...", sleep_time);
-            sleep(Duration::from_secs(sleep_time)).await;
-
-            if retry_interval > self.opts.max_retry_interval_seconds {
-                retry_interval = self.opts.max_retry_interval_seconds;
-            } else {
-                retry_interval *= 2;
-            }
-        }
-    }
-
-    pub fn force_reconnect(&self) -> Result<(), flume::SendError<()>> {
-        info!("Force reconnect requested.");
-        self.reconnect_tx.send(()) // Send a reconnect signal
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::sync::{Arc, Mutex};
-    use tokio::task;
-    use tokio::time::{timeout, Duration};
-    use tokio_tungstenite::tungstenite::Message;
-
-    #[test]
-    fn test_build_ws_url() {
-        let opts = JetstreamOptions {
-            wanted_collections: Some(vec!["col1".to_string(), "col2".to_string()]),
-            wanted_dids: Some(vec!["did1".to_string()]),
-            ..Default::default()
-        };
-        let connection = JetstreamConnection::new(opts);
-
-        let test = Arc::new(Mutex::new(Some(8373)));
-
-        let url = connection.build_ws_url(test);
-
-        assert!(url.starts_with("wss://"));
-        assert!(url.contains("cursor=8373"));
-        assert!(url.contains("wantedCollections=col1"));
-        assert!(url.contains("wantedCollections=col2"));
-        assert!(url.contains("wantedDids=did1"));
-    }
-
-    #[tokio::test]
-    async fn test_force_reconnect() {
-        let opts = JetstreamOptions::default();
-        let connection = JetstreamConnection::new(opts);
-
-        // Spawn a task to listen for the reconnect signal
-        let reconnect_rx = connection.reconnect_rx.clone();
-        let recv_task = task::spawn(async move {
-            reconnect_rx
-                .recv_async()
-                .await
-                .expect("Failed to receive reconnect signal");
-        });
-
-        connection
-            .force_reconnect()
-            .expect("Failed to send reconnect signal");
-
-        // Ensure reconnect signal was received
-        assert!(recv_task.await.is_ok());
-    }
-
-    #[tokio::test]
-    async fn test_message_queue() {
-        let opts = JetstreamOptions::default();
-        let connection = JetstreamConnection::new(opts);
-
-        let msg_rx = connection.get_msg_rx();
-        let msg = Message::Text("test message".into());
-
-        // Send a message to the queue
-        connection
-            .msg_tx
-            .send_async(msg.clone())
-            .await
-            .expect("Failed to send message");
-
-        // Receive and verify the message
-        let received = msg_rx
-            .recv_async()
-            .await
-            .expect("Failed to receive message");
-        assert_eq!(received, msg);
-    }
-
-    #[tokio::test]
-    async fn test_connection_retries_on_failure() {
-        let opts = JetstreamOptions::default();
-        let connection = Arc::new(JetstreamConnection::new(opts));
-
-        let cursor = Arc::new(Mutex::new(None));
-
-        // Timeout to prevent infinite loop
-        let result = timeout(Duration::from_secs(3), connection.connect(cursor)).await;
-
-        assert!(result.is_err(), "Expected timeout due to retry logic");
-    }
-
-    #[tokio::test]
-    async fn test_reconnect_after_receive_timeout() {
-        use tokio::net::TcpListener;
-        use tokio_tungstenite::accept_async;
-
-        let opts = JetstreamOptions {
-            ws_url: crate::endpoints::JetstreamEndpoints::Custom("ws://127.0.0.1:9001".to_string()),
-            bound: 5,
-            max_retry_interval_seconds: 1,
-            ..Default::default()
-        };
-        let connection = JetstreamConnection::new(opts);
-        let cursor = Arc::new(Mutex::new(None));
-
-        // set up dummy "websocket"
-        let listener = TcpListener::bind("127.0.0.1:9001")
-            .await
-            .expect("Failed to bind");
-        let server_handle = tokio::spawn(async move {
-            if let Ok((stream, _)) = listener.accept().await {
-                let ws_stream = accept_async(stream).await.expect("Failed to accept");
-                // send nothing
-                tokio::time::sleep(Duration::from_secs(6)).await;
-                drop(ws_stream);
-            }
-        });
-
-        // spawn, then run for >30 seconds to trigger reconnect
-        let connect_handle = tokio::spawn(async move {
-            tokio::time::timeout(Duration::from_secs(5), connection.connect(cursor))
-                .await
-                .ok();
-        });
-
-        let _ = tokio::join!(server_handle, connect_handle);
-    }
-}

services/rocketman/src/endpoints.rs

···
-use std::fmt::{Display, Formatter, Result};
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub enum JetstreamEndpointLocations {
-    UsEast,
-    UsWest,
-}
-
-impl Display for JetstreamEndpointLocations {
-    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
-        write!(
-            f,
-            "{}",
-            match self {
-                Self::UsEast => "us-east",
-                Self::UsWest => "us-west",
-            }
-        )
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub enum JetstreamEndpoints {
-    Public(JetstreamEndpointLocations, i8),
-    Custom(String),
-}
-
-impl Display for JetstreamEndpoints {
-    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
-        match self {
-            Self::Public(location, id) => write!(
-                f,
-                "wss://jetstream{}.{}.bsky.network/subscribe",
-                id, location
-            ),
-            Self::Custom(url) => write!(f, "{}", url),
-        }
-    }
-}
-
-impl Default for JetstreamEndpoints {
-    fn default() -> Self {
-        Self::Public(JetstreamEndpointLocations::UsEast, 2)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_display_public() {
-        let endpoint = JetstreamEndpoints::Public(JetstreamEndpointLocations::UsEast, 2);
-        assert_eq!(
-            endpoint.to_string(),
-            "wss://jetstream2.us-east.bsky.network/subscribe"
-        );
-    }
-
-    #[test]
-    fn test_display_custom() {
-        let endpoint = JetstreamEndpoints::Custom("wss://custom.bsky.network/subscribe".into());
-        assert_eq!(endpoint.to_string(), "wss://custom.bsky.network/subscribe");
-    }
-}

services/rocketman/src/err.rs

···
-// TODO: error types instead of using anyhow

services/rocketman/src/handler.rs

···
-use anyhow::Result;
-use flume::Sender;
-use metrics::{counter, describe_counter, Unit};
-use serde_json::Value;
-use std::{
-    collections::HashMap,
-    sync::{Arc, Mutex},
-};
-use tokio_tungstenite::tungstenite::{Error, Message};
-use tracing::{debug, error};
-
-#[cfg(feature = "zstd")]
-use std::io::Cursor as IoCursor;
-#[cfg(feature = "zstd")]
-use std::sync::LazyLock;
-#[cfg(feature = "zstd")]
-use zstd::dict::DecoderDictionary;
-
-use crate::{
-    ingestion::LexiconIngestor,
-    types::event::{Event, Kind},
-};
-
-/// The custom `zstd` dictionary used for decoding compressed Jetstream messages.
-///
-/// Sourced from the [official Bluesky Jetstream repo.](https://github.com/bluesky-social/jetstream/tree/main/pkg/models)
-#[cfg(feature = "zstd")]
-static ZSTD_DICTIONARY: LazyLock<DecoderDictionary> =
-    LazyLock::new(|| DecoderDictionary::copy(include_bytes!("../zstd/dictionary")));
-
-pub async fn handle_message(
-    message: Message,
-    ingestors: &HashMap<String, Box<dyn LexiconIngestor + Send + Sync>>,
-    reconnect_tx: Sender<()>,
-    cursor: Arc<Mutex<Option<u64>>>,
-) -> Result<()> {
-    describe_counter!(
-        "jetstream.event",
-        Unit::Count,
-        "number of event ingest attempts"
-    );
-    describe_counter!(
-        "jetstream.event.parse",
-        Unit::Count,
-        "events that were successfully processed"
-    );
-    describe_counter!(
-        "jetstream.event.fail",
-        Unit::Count,
-        "events that could not be read"
-    );
-    describe_counter!("jetstream.error", Unit::Count, "errors encountered");
-    match message {
-        Message::Text(text) => {
-            debug!("Text message received");
-            counter!("jetstream.event").increment(1);
-            let envelope: Event<Value> = serde_json::from_str(&text).map_err(|e| {
-                anyhow::anyhow!("Failed to parse message: {} with json string {}", e, text)
-            })?;
-            debug!("envelope: {:?}", envelope);
-            handle_envelope(envelope, cursor, ingestors).await?;
-            Ok(())
-        }
-        #[cfg(feature = "zstd")]
-        Message::Binary(bytes) => {
-            debug!("Binary message received");
-            counter!("jetstream.event").increment(1);
-            let decoder = zstd::stream::Decoder::with_prepared_dictionary(
-                IoCursor::new(bytes),
-                &ZSTD_DICTIONARY,
-            )?;
-            let envelope: Event<Value> = serde_json::from_reader(decoder)
-                .map_err(|e| anyhow::anyhow!("Failed to parse binary message: {}", e))?;
-            debug!("envelope: {:?}", envelope);
-            handle_envelope(envelope, cursor, ingestors).await?;
-            Ok(())
-        }
-        #[cfg(not(feature = "zstd"))]
-        Message::Binary(_) => {
-            debug!("Binary message received");
-            Err(anyhow::anyhow!(
-                "binary message received but zstd feature is not enabled"
-            ))
-        }
-        Message::Close(_) => {
-            debug!("Server closed connection");
-            if let Err(e) = reconnect_tx.send(()) {
-                counter!("jetstream.event.parse.error", "error" => "failed_to_send_reconnect_signal").increment(1);
-                error!("Failed to send reconnect signal: {}", e);
-            }
-            Err(Error::ConnectionClosed.into())
-        }
-        _ => Ok(()),
-    }
-}
-
-async fn handle_envelope(
-    envelope: Event<Value>,
-    cursor: Arc<Mutex<Option<u64>>>,
-    ingestors: &HashMap<String, Box<dyn LexiconIngestor + Send + Sync>>,
-) -> Result<()> {
-    if let Some(ref time_us) = envelope.time_us {
-        debug!("Time: {}", time_us);
-        if let Some(cursor) = cursor.lock().unwrap().as_mut() {
-            debug!("Cursor: {}", cursor);
-            if time_us > cursor {
-                debug!("Cursor is behind, resetting");
-                *cursor = *time_us;
-            }
-        }
-    }
-
-    match envelope.kind {
-        Kind::Commit => match extract_commit_nsid(&envelope) {
-            Ok(nsid) => {
-                if let Some(fun) = ingestors.get(&nsid) {
-                    match fun.ingest(envelope).await {
-                        Ok(_) => {
-                            counter!("jetstream.event.parse.commit", "nsid" => nsid).increment(1)
-                        }
-                        Err(e) => {
-                            error!("Error ingesting commit with nsid {}: {}", nsid, e);
-                            counter!("jetstream.error").increment(1);
-                            counter!("jetstream.event.fail").increment(1);
-                        }
-                    }
-                }
-            }
-            Err(e) => error!("Error parsing commit: {}", e),
-        },
-        Kind::Identity => {
-            counter!("jetstream.event.parse.identity").increment(1);
-        }
-        Kind::Account => {
-            counter!("jetstream.event.parse.account").increment(1);
-        }
-        Kind::Unknown(kind) => {
-            counter!("jetstream.event.parse.unknown", "kind" => kind).increment(1);
-        }
-    }
-    Ok(())
-}
-
-fn extract_commit_nsid(envelope: &Event<Value>) -> anyhow::Result<String> {
-    // if the type is not a commit
-    if envelope.commit.is_none() {
-        return Err(anyhow::anyhow!(
-            "Message has no commit, so there is no nsid attached."
-        ));
-    } else if let Some(ref commit) = envelope.commit {
-        return Ok(commit.collection.clone());
-    }
-
-    Err(anyhow::anyhow!("Failed to extract nsid: unknown error"))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::types::event::Event;
-    use anyhow::Result;
-    use async_trait::async_trait;
-    use flume::{Receiver, Sender};
-    use serde_json::json;
-    use std::{
-        collections::HashMap,
-        sync::{Arc, Mutex},
-    };
-    use tokio_tungstenite::tungstenite::Message;
-
-    // Dummy ingestor that records if it was called.
-    struct DummyIngestor {
-        pub called: Arc<Mutex<bool>>,
-    }
-
-    #[async_trait]
-    impl crate::ingestion::LexiconIngestor for DummyIngestor {
-        async fn ingest(&self, _event: Event<serde_json::Value>) -> Result<(), anyhow::Error> {
-            let mut called = self.called.lock().unwrap();
-            *called = true;
-            Ok(())
-        }
-    }
-
-    // Dummy ingestor that always returns an error.
-    struct ErrorIngestor;
-
-    #[async_trait]
-    impl crate::ingestion::LexiconIngestor for ErrorIngestor {
-        async fn ingest(&self, _event: Event<serde_json::Value>) -> Result<(), anyhow::Error> {
-            Err(anyhow::anyhow!("Ingest error"))
-        }
-    }
-
-    // Helper to create a reconnect channel.
-    fn setup_reconnect_channel() -> (Sender<()>, Receiver<()>) {
-        flume::unbounded()
-    }
-
-    #[tokio::test]
-    async fn test_valid_commit_success() {
-        let (reconnect_tx, _reconnect_rx) = setup_reconnect_channel();
-        let cursor = Arc::new(Mutex::new(Some(100)));
-        let called_flag = Arc::new(Mutex::new(false));
-
-        // Create a valid commit event JSON.
-        let event_json = json!({
-            "did": "did:example:123",
-            "time_us": 200,
-            "kind": "commit",
-            "commit": {
-                "rev": "1",
-                "operation": "create",
-                "collection": "ns1",
-                "rkey": "rkey1",
-                "record": { "foo": "bar" },
-                "cid": "cid123"
-            },
-        })
-        .to_string();
-
-        let mut ingestors: HashMap<
-            String,
-            Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>,
-        > = HashMap::new();
-        ingestors.insert(
-            "ns1".to_string(),
-            Box::new(DummyIngestor {
-                called: called_flag.clone(),
-            }),
-        );
-
-        let result = handle_message(
-            Message::Text(event_json),
-            &ingestors,
-            reconnect_tx,
-            cursor.clone(),
-        )
-        .await;
-        assert!(result.is_ok());
-        // Check that the ingestor was called.
-        assert!(*called_flag.lock().unwrap());
-        // Verify that the cursor got updated.
-        assert_eq!(*cursor.lock().unwrap(), Some(200));
-    }
-
-    #[cfg(feature = "zstd")]
-    #[tokio::test]
-    async fn test_binary_valid_commit() {
-        let (reconnect_tx, _reconnect_rx) = setup_reconnect_channel();
-        let cursor = Arc::new(Mutex::new(Some(100)));
-        let called_flag = Arc::new(Mutex::new(false));
-
-        let uncompressed_json = json!({
-            "did": "did:example:123",
-            "time_us": 200,
-            "kind": "commit",
-            "commit": {
-                "rev": "1",
-                "operation": "create",
-                "collection": "ns1",
-                "rkey": "rkey1",
-                "record": { "foo": "bar" },
-                "cid": "cid123"
-            },
-        })
-        .to_string();
-
-        let compressed_dest: IoCursor<Vec<u8>> = IoCursor::new(vec![]);
-        let mut encoder = zstd::Encoder::with_prepared_dictionary(
-            compressed_dest,
-            &zstd::dict::EncoderDictionary::copy(include_bytes!("../zstd/dictionary"), 0),
-        )
-        .unwrap();
-        std::io::copy(
-            &mut IoCursor::new(uncompressed_json.as_bytes()),
-            &mut encoder,
-        )
-        .unwrap();
-        let compressed_dest = encoder.finish().unwrap();
-
-        let mut ingestors: HashMap<
-            String,
-            Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>,
-        > = HashMap::new();
-        ingestors.insert(
-            "ns1".to_string(),
-            Box::new(DummyIngestor {
-                called: called_flag.clone(),
-            }),
-        );
-
-        let result = handle_message(
-            Message::Binary(compressed_dest.into_inner()),
-            &ingestors,
-            reconnect_tx,
-            cursor.clone(),
-        )
-        .await;
-
-        assert!(result.is_ok());
-        // Check that the ingestor was called.
-        assert!(*called_flag.lock().unwrap());
-        // Verify that the cursor got updated.
-        assert_eq!(*cursor.lock().unwrap(), Some(200));
-    }
-
-    #[tokio::test]
-    async fn test_commit_ingest_failure() {
-        let (reconnect_tx, _reconnect_rx) = setup_reconnect_channel();
-        let cursor = Arc::new(Mutex::new(Some(100)));
-
-        // Valid commit event with an ingestor that fails.
-        let event_json = json!({
-            "did": "did:example:123",
-            "time_us": 300,
-            "kind": "commit",
-            "commit": {
-                "rev": "1",
-                "operation": "create",
-                "collection": "ns_error",
-                "rkey": "rkey1",
-                "record": { "foo": "bar" },
-                "cid": "cid123"
-            },
-            "identity": null
-        })
-        .to_string();
-
-        let mut ingestors: HashMap<
-            String,
-            Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>,
-        > = HashMap::new();
-        ingestors.insert("ns_error".to_string(), Box::new(ErrorIngestor));
-
-        // Even though ingestion fails, handle_message returns Ok(()).
-        let result = handle_message(
-            Message::Text(event_json),
-            &ingestors,
-            reconnect_tx,
-            cursor.clone(),
-        )
-        .await;
-        assert!(result.is_ok());
-        // Cursor should still update because it comes before the ingest call.
-        assert_eq!(*cursor.lock().unwrap(), Some(300));
-    }
-
-    #[tokio::test]
-    async fn test_identity_message() {
-        let (reconnect_tx, _reconnect_rx) = setup_reconnect_channel();
-        let cursor = Arc::new(Mutex::new(None));
-        // Valid identity event.
-        let event_json = json!({
-            "did": "did:example:123",
-            "time_us": 150,
-            "kind": "identity",
-            "commit": null,
-            "identity": {
-                "did": "did:example:123",
-                "handle": "user",
-                "seq": 1,
-                "time": "2025-01-01T00:00:00Z"
-            }
-        })
-        .to_string();
-        let ingestors: HashMap<String, Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>> =
-            HashMap::new();
-
-        let result =
-            handle_message(Message::Text(event_json), &ingestors, reconnect_tx, cursor).await;
-        assert!(result.is_ok());
-    }
-
-    #[tokio::test]
-    async fn test_close_message() {
-        let (reconnect_tx, reconnect_rx) = setup_reconnect_channel();
-        let cursor = Arc::new(Mutex::new(None));
-        let ingestors: HashMap<String, Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>> =
-            HashMap::new();
-
-        let result = handle_message(Message::Close(None), &ingestors, reconnect_tx, cursor).await;
-        // Should return an error due to connection close.
-        assert!(result.is_err());
-        // Verify that a reconnect signal was sent.
-        let signal = reconnect_rx.recv_async().await;
-        assert!(signal.is_ok());
-    }
-
-    #[tokio::test]
-    async fn test_invalid_json() {
-        let (reconnect_tx, _reconnect_rx) = setup_reconnect_channel();
-        let cursor = Arc::new(Mutex::new(None));
-        let ingestors: HashMap<String, Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>> =
-            HashMap::new();
-
-        let invalid_json = "this is not json".to_string();
-        let result = handle_message(
-            Message::Text(invalid_json),
-            &ingestors,
-            reconnect_tx,
-            cursor,
-        )
-        .await;
-        assert!(result.is_err());
-    }
-
-    #[tokio::test]
-    async fn test_cursor_not_updated_if_lower() {
-        let (reconnect_tx, _reconnect_rx) = setup_reconnect_channel();
-        // Set an initial cursor value.
-        let cursor = Arc::new(Mutex::new(Some(300)));
-        let event_json = json!({
-            "did": "did:example:123",
-            "time_us": 200,
-            "kind": "commit",
-            "commit": {
-                "rev": "1",
-                "operation": "create",
-                "collection": "ns1",
-                "rkey": "rkey1",
-                "record": { "foo": "bar" },
-                "cid": "cid123"
-            },
-            "identity": null
-        })
-        .to_string();
-
-        // Use a dummy ingestor that does nothing.
-        let mut ingestors: HashMap<
-            String,
-            Box<dyn crate::ingestion::LexiconIngestor + Send + Sync>,
-        > = HashMap::new();
-        ingestors.insert(
-            "ns1".to_string(),
-            Box::new(DummyIngestor {
-                called: Arc::new(Mutex::new(false)),
-            }),
-        );
-
-        let result = handle_message(
-            Message::Text(event_json),
-            &ingestors,
-            reconnect_tx,
-            cursor.clone(),
-        )
-        .await;
-        assert!(result.is_ok());
-        // Cursor should remain unchanged.
-        assert_eq!(*cursor.lock().unwrap(), Some(300));
-    }
-}

services/rocketman/src/ingestion.rs

···
-use anyhow::Result;
-use async_trait::async_trait;
-use serde_json::Value;
-use tracing::info;
-
-use crate::types::event::Event;
-
-#[async_trait]
-pub trait LexiconIngestor {
-    async fn ingest(&self, message: Event<Value>) -> Result<()>;
-}
-
-pub struct DefaultLexiconIngestor;
-
-#[async_trait]
-impl LexiconIngestor for DefaultLexiconIngestor {
-    async fn ingest(&self, message: Event<Value>) -> Result<()> {
-        info!("Default lexicon processing: {:?}", message);
-        // Process message for default lexicon.
-        Ok(())
-    }
-}

services/rocketman/src/lib.rs

···
-// lib.rs
-pub mod connection;
-pub mod endpoints;
-pub mod handler;
-pub mod ingestion;
-pub mod options;
-pub mod time;
-pub mod types;

services/rocketman/src/options.rs

···
-use bon::Builder;
-
-use crate::endpoints::JetstreamEndpoints;
-
-#[derive(Builder, Debug)]
-pub struct JetstreamOptions {
-    #[builder(default)]
-    pub ws_url: JetstreamEndpoints,
-    #[builder(default)]
-    pub max_retry_interval_seconds: u64,
-    #[builder(default)]
-    pub connection_success_time_seconds: u64,
-    #[builder(default)]
-    pub bound: usize,
-    #[builder(default)]
-    pub timeout_time_sec: usize,
-    #[cfg(feature = "zstd")]
-    #[builder(default = true)]
-    pub compress: bool,
-    pub wanted_collections: Option<Vec<String>>,
-    pub wanted_dids: Option<Vec<String>>,
-    pub cursor: Option<String>,
-}
-
-impl Default for JetstreamOptions {
-    fn default() -> Self {
-        Self {
-            ws_url: JetstreamEndpoints::default(),
-            max_retry_interval_seconds: 120,
-            connection_success_time_seconds: 60,
-            bound: 65536,
-            timeout_time_sec: 40,
-            #[cfg(feature = "zstd")]
-            compress: true,
-            wanted_collections: None,
-            wanted_dids: None,
-            cursor: None,
-        }
-    }
-}

services/rocketman/src/time/mod.rs

···
-use std::time::{Duration, Instant, SystemTime};
-
-pub mod system_time;
-
-pub trait TimeProvider {
-    fn new() -> Self;
-    fn now(&self) -> SystemTime; // Get the current time
-    fn elapsed(&self, earlier: SystemTime) -> Duration; // Calculate the elapsed time.
-    fn instant_now(&self) -> Instant; // For compatibility with your existing code (if needed)
-    fn instant_elapsed(&self, earlier: Instant) -> Duration;
-}

services/rocketman/src/time/system_time.rs

···
-use std::time::{Duration, Instant, SystemTime};
-
-use super::TimeProvider;
-
-#[derive(Default, Clone, Copy)] // Add these derives for ease of use
-pub struct SystemTimeProvider; // No fields needed, just a marker type
-
-impl TimeProvider for SystemTimeProvider {
-    fn new() -> Self {
-        Self
-    }
-
-    fn now(&self) -> SystemTime {
-        SystemTime::now()
-    }
-
-    fn elapsed(&self, earlier: SystemTime) -> Duration {
-        earlier.elapsed().unwrap_or_else(|_| Duration::from_secs(0))
-    }
-
-    fn instant_now(&self) -> Instant {
-        Instant::now()
-    }
-
-    fn instant_elapsed(&self, earlier: Instant) -> Duration {
-        earlier.elapsed()
-    }
-}

services/rocketman/src/types/event.rs

···
-use serde::{Deserialize, Deserializer, Serialize};
-
-#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
-#[serde(rename_all = "lowercase")]
-pub enum Kind {
-    Account,
-    Identity,
-    Commit,
-    Unknown(String),
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub struct Event<T> {
-    pub did: String,
-    pub time_us: Option<u64>,
-    pub kind: Kind,
-    pub commit: Option<Commit<T>>,
-    pub identity: Option<Identity>,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Identity {
-    did: String,
-    handle: Option<String>,
-    seq: u64,
-    time: String,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(rename_all = "lowercase")]
-enum AccountStatus {
-    TakenDown,
-    Suspended,
-    Deleted,
-    Activated,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct Account {
-    did: String,
-    handle: String,
-    seq: u64,
-    time: String,
-    status: AccountStatus,
-}
-
-#[derive(Debug, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct Commit<T> {
-    pub rev: String,
-    pub operation: Operation,
-    pub collection: String,
-    pub rkey: String,
-    pub record: Option<T>,
-    pub cid: Option<String>,
-}
-
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(rename_all = "lowercase")]
-pub enum Operation {
-    Create,
-    Update,
-    Delete,
-}
-
-/// Enforce that record is None only when operation is 'delete'
-impl<'de, T> Deserialize<'de> for Commit<T>
-where
-    T: Deserialize<'de>,
-{
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        // Helper struct to perform the deserialization.
-        #[derive(Deserialize)]
-        #[serde(rename_all = "camelCase")]
-        struct Helper<T> {
-            rev: String,
-            operation: Operation,
-            collection: String,
-            rkey: String,
-            record: Option<T>,
-            cid: Option<String>,
-        }
-
-        let helper = Helper::deserialize(deserializer)?;
-
-        match helper.operation {
-            Operation::Delete => {
-                if helper.record.is_some() || helper.cid.is_some() {
-                    return Err(<D::Error as serde::de::Error>::custom(
-                        "record and cid must be null when operation is delete",
-                    ));
-                }
-            }
-            _ => {
-                if helper.record.is_none() || helper.cid.is_none() {
-                    return Err(<D::Error as serde::de::Error>::custom(
-                        "record and cid must be present unless operation is delete",
-                    ));
-                }
-            }
-        }
-
-        Ok(Commit {
-            rev: helper.rev,
-            operation: helper.operation,
-            collection: helper.collection,
-            rkey: helper.rkey,
-            record: helper.record,
-            cid: helper.cid,
-        })
-    }
-}

services/rocketman/src/types/mod.rs

···
-pub mod event;